Lucas Berger
1652a68c51
- Wire publishListEvent fan-out in lists.ts after every write mutation (item:added/updated/deleted, list:updated/deleted)
- Add GET /api/sse/lists scoped endpoint in sse.ts: resolveUserId → 401 on null; getAccessibleListIds → subscribe only to accessible channels; 30s heartbeat; cleanup on disconnect (D-04/T-04-01/T-04-02)
- Create useListSSE.ts: bounded-backoff EventSource wrapper (250ms→500ms→1s→2s→4s→cap 8s); MAX_ATTEMPTS=6; withCredentials:true; close-before-retry prevents reconnect storm (Pitfall 3); invalidates ['list', listId] on open (D-10) and on each event; onStateChange('disconnected') after exhaustion (D-11)
- Create LiveSyncIndicator.tsx: connected=green dot; reconnecting=pulsing muted dot + label; disconnected=red dot + 'Updates paused' (role=alert); correct ARIA per UI-SPEC
- Wire useListSSE + LiveSyncIndicator into ListDetail header; retain refetchInterval:30000 polling fallback (D-12)
- All 8 useListSSE tests pass; all 54 API tests pass; both typechecks pass
- playwright-cli: live update confirmed (eggs item added via API appeared in browser without manual refresh)
122 lines
4.3 KiB
TypeScript
122 lines
4.3 KiB
TypeScript
/**
|
|
* SSE router — streaming endpoints for real-time events.
|
|
*
|
|
* Routes:
|
|
* GET /heartbeat — Pangolin smoke-test (Phase 1 entry gate)
|
|
* GET /lists — Scoped live-list fan-out stream (LIST-04, D-04)
|
|
*
|
|
* Both routes sit behind oidcAuthMiddleware in index.ts (T-04-01).
|
|
*
|
|
* Source: https://hono.dev/docs/helpers/streaming
|
|
* RESEARCH Pattern 5 (heartbeat) + Finding 1 (lists scoped fan-out)
|
|
*/
|
|
|
|
import { Hono } from 'hono'
|
|
import type { Context } from 'hono'
|
|
import { streamSSE } from 'hono/streaming'
|
|
import { getAuth } from '../auth/middleware.js'
|
|
import { upsertUser, deriveDisplayName } from '../auth/user.js'
|
|
import { subscribeListEvents } from '../lib/listEmitter.js'
|
|
import { getAccessibleListIds } from '../lib/listAccess.js'
|
|
// Side-effect import: brings in the ContextVariableMap augmentation for c.get('user')
|
|
import '../auth/devBypass.js'
|
|
|
|
export const sseRouter = new Hono()
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Auth helper — same pattern as lists.ts (per-router duplication convention).
|
|
// Resolution order: dev-bypass user first, then OIDC.
|
|
// ---------------------------------------------------------------------------
|
|
async function resolveUserId(c: Context): Promise<number | null> {
|
|
const devUser = c.get('user') as { id: number } | undefined
|
|
if (devUser) return devUser.id
|
|
|
|
const auth = await getAuth(c)
|
|
if (!auth) return null
|
|
|
|
const iss = (auth.iss as string | undefined) ?? ''
|
|
const sub = auth.sub ?? ''
|
|
const displayName = deriveDisplayName(auth)
|
|
const user = await upsertUser(iss, sub, displayName)
|
|
return user?.id ?? null
|
|
}
|
|
|
|
/**
|
|
* GET /heartbeat
|
|
* Streams SSE heartbeat events every 10 seconds until client disconnects.
|
|
* Response: text/event-stream with events of type "heartbeat".
|
|
*/
|
|
sseRouter.get('/heartbeat', (c) => {
|
|
return streamSSE(c, async (stream) => {
|
|
let id = 0
|
|
while (!stream.aborted) {
|
|
await stream.writeSSE({
|
|
data: JSON.stringify({ ts: new Date().toISOString(), id }),
|
|
event: 'heartbeat',
|
|
id: String(id++),
|
|
})
|
|
await stream.sleep(10_000)
|
|
}
|
|
})
|
|
})
|
|
|
|
/**
|
|
* GET /lists
|
|
*
|
|
* Scoped live-list SSE stream (LIST-04, D-04).
|
|
*
|
|
* Security (T-04-01, T-04-02):
|
|
* - resolveUserId → 401 on null (same OIDC guard as all /api/sse/* routes).
|
|
* - Subscribes ONLY to list channels the caller can access (owner + shares).
|
|
* Events for private lists of other members are never delivered (D-04).
|
|
*
|
|
* Behavior:
|
|
* - Resolves the caller's accessible list IDs via getAccessibleListIds(userId).
|
|
* - Opens one subscribeListEvents subscription per accessible list.
|
|
* - Each event is forwarded as `event: event.type; data: JSON.stringify(event)`.
|
|
* - A 30s heartbeat keeps the Pangolin connection alive (proven in smoke test).
|
|
* - On client disconnect (stream.aborted), all subscriptions are cleaned up.
|
|
*
|
|
* D-10: Client receives minimal { type, listId } payload and full-refetches.
|
|
* Payload is not relied on for cache updates — only triggers invalidation.
|
|
* D-11: Bounded backoff + give-up logic lives in the PWA useListSSE hook.
|
|
* D-12: PWA refetchInterval: 30000 polling fallback always active.
|
|
*/
|
|
sseRouter.get('/lists', async (c) => {
|
|
const userId = await resolveUserId(c)
|
|
if (userId === null) return c.json({ error: 'Unauthorized' }, 401)
|
|
|
|
const accessibleListIds = await getAccessibleListIds(userId)
|
|
|
|
return streamSSE(c, async (stream) => {
|
|
const unsubscribers: Array<() => void> = []
|
|
|
|
// Subscribe to each accessible list's channel (D-04 — scoped, not global)
|
|
for (const listId of accessibleListIds) {
|
|
const unsub = subscribeListEvents(listId, async (event) => {
|
|
if (stream.aborted) return
|
|
await stream.writeSSE({
|
|
data: JSON.stringify(event),
|
|
event: event.type,
|
|
id: `${listId}-${Date.now()}`,
|
|
})
|
|
})
|
|
unsubscribers.push(unsub)
|
|
}
|
|
|
|
// 30s heartbeat — keeps Pangolin connection alive (smoke-tested in Phase 1)
|
|
let tick = 0
|
|
while (!stream.aborted) {
|
|
await stream.writeSSE({
|
|
data: JSON.stringify({ ts: new Date().toISOString() }),
|
|
event: 'heartbeat',
|
|
id: String(tick++),
|
|
})
|
|
await stream.sleep(30_000)
|
|
}
|
|
|
|
// Cleanup all subscriptions on client disconnect
|
|
unsubscribers.forEach((unsub) => unsub())
|
|
})
|
|
})
|