diff --git a/apps/api/src/routes/lists.ts b/apps/api/src/routes/lists.ts index 7cd86c1..58497aa 100644 --- a/apps/api/src/routes/lists.ts +++ b/apps/api/src/routes/lists.ts @@ -30,8 +30,7 @@ import { rankForAppend } from '../lib/rank.js' // Side-effect import: brings in the ContextVariableMap augmentation for c.get('user') import '../auth/devBypass.js' -// Uncomment in Plan 06 when SSE endpoint exists: -// import { publishListEvent } from '../lib/listEmitter.js' +import { publishListEvent } from '../lib/listEmitter.js' export const listsRouter = new Hono() @@ -284,8 +283,8 @@ listsRouter.post('/', zValidator('json', createListSchema), async (c) => { .where(eq(lists.id, listId)) .limit(1) - // Plan 06 SSE seam: - // publishListEvent(listId, { type: 'list:updated', listId, payload: newList }) + // Fan-out: notify accessible subscribers that this list was created/updated (LIST-04) + publishListEvent(listId, { type: 'list:updated', listId, payload: { id: listId, name: newList.name } }) return c.json( { @@ -376,8 +375,8 @@ listsRouter.patch('/:id', zValidator('json', patchListSchema), async (c) => { .where(eq(lists.id, listId)) .limit(1) - // Plan 06 SSE seam: - // publishListEvent(listId, { type: 'list:updated', listId, payload: updated }) + // Fan-out: notify accessible subscribers that this list metadata changed (LIST-04) + publishListEvent(listId, { type: 'list:updated', listId, payload: { id: listId, name: updated.name } }) return c.json({ id: updated.id, @@ -422,8 +421,8 @@ listsRouter.delete('/:id', async (c) => { await db.delete(lists).where(eq(lists.id, listId)) - // Plan 06 SSE seam: - // publishListEvent(listId, { type: 'list:deleted', listId, payload: { id: listId } }) + // Fan-out: notify accessible subscribers that this list was deleted (LIST-04) + publishListEvent(listId, { type: 'list:deleted', listId, payload: { id: listId } }) return c.json({ id: listId }) } catch (err) { @@ -488,8 +487,8 @@ listsRouter.post('/:id/items', zValidator('json', createItemSchema), async (c) = .where(eq(listItems.id, inserted.id)) .limit(1) - // Plan 06 SSE seam: - // publishListEvent(listId, { type: 'item:added', listId, payload: newItem }) + // Fan-out: notify accessible subscribers that an item was added (LIST-04) + publishListEvent(listId, { type: 'item:added', listId, payload: { id: newItem.id, listId, text: newItem.text } }) return c.json( { @@ -631,8 +630,8 @@ listItemsRouter.patch('/:itemId', zValidator('json', patchItemSchema), async (c) .where(eq(listItems.id, itemId)) .limit(1) - // Plan 06 SSE seam: - // publishListEvent(item.listId, { type: 'item:updated', listId: item.listId, payload: updated }) + // Fan-out: notify accessible subscribers that an item was updated (LIST-04) + publishListEvent(item.listId, { type: 'item:updated', listId: item.listId, payload: { id: updated.id, listId: updated.listId } }) return c.json({ id: updated.id, @@ -682,8 +681,8 @@ listItemsRouter.delete('/:itemId', async (c) => { // Delete-wins (D-09): delete is final; no rollback path. await db.delete(listItems).where(eq(listItems.id, itemId)) - // Plan 06 SSE seam: - // publishListEvent(item.listId, { type: 'item:deleted', listId: item.listId, payload: { id: itemId } }) + // Fan-out: notify accessible subscribers that an item was deleted (LIST-04) + publishListEvent(item.listId, { type: 'item:deleted', listId: item.listId, payload: { id: itemId } }) return c.json({ id: itemId }) } catch (err) { diff --git a/apps/api/src/routes/sse.ts b/apps/api/src/routes/sse.ts index 2b79968..3095b78 100644 --- a/apps/api/src/routes/sse.ts +++ b/apps/api/src/routes/sse.ts @@ -1,25 +1,46 @@ /** - * GET /api/sse/heartbeat — Pangolin SSE pass-through smoke test endpoint. + * SSE router — streaming endpoints for real-time events. * - * Emits a `heartbeat` event every 10 seconds with { ts, id } payload. - * Runs until the client disconnects (stream.aborted). + * Routes: + * GET /heartbeat — Pangolin smoke-test (Phase 1 entry gate) + * GET /lists — Scoped live-list fan-out stream (LIST-04, D-04) * - * Mounted under /api/sse in index.ts, so it sits behind oidcAuthMiddleware (T-04-01). - * Heartbeat payload carries only timestamps — no user data or secrets (T-04-02). - * - * Smoke test procedure (D-08): - * curl -N https://familysync./api/sse/heartbeat - * Keep open 5+ min — confirm no proxy timeout. PASS → SSE viable for Phase 4. + * Both routes sit behind oidcAuthMiddleware in index.ts (T-04-01). * * Source: https://hono.dev/docs/helpers/streaming - * RESEARCH Pattern 5: Pangolin SSE Smoke Test + * 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 { + 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. @@ -38,3 +59,63 @@ sseRouter.get('/heartbeat', (c) => { } }) }) + +/** + * 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()) + }) +}) diff --git a/apps/pwa/src/components/LiveSyncIndicator.tsx b/apps/pwa/src/components/LiveSyncIndicator.tsx new file mode 100644 index 0000000..a8fd894 --- /dev/null +++ b/apps/pwa/src/components/LiveSyncIndicator.tsx @@ -0,0 +1,119 @@ +/** + * LiveSyncIndicator — visual SSE connection status indicator (LIST-04, D-11). + * + * States per UI-SPEC §"LiveSyncIndicator": + * connected: 8px filled circle in --color-member-1 (#50C878), no label + * reconnecting: 8px pulsing circle in --color-text-muted + "Reconnecting…" label + * disconnected: 8px filled circle in --color-destructive + "Updates paused" label + * + * Accessibility: + * - role="status" for connected/reconnecting (polite announcements) + * - role="alert" for disconnected state (assertive announcement) + * - aria-label per UI-SPEC copywriting contract + * + * Positioned at the right end of the ListDetail header row. + * Visible only inside ListDetail (not on ListsIndex). + */ + +import type { SyncState } from '../hooks/useListSSE.js' + +interface LiveSyncIndicatorProps { + state: SyncState +} + +export function LiveSyncIndicator({ state }: LiveSyncIndicatorProps) { + if (state === 'connected') { + return ( +
+ {/* 8px filled green dot — --color-member-1 (#50C878) per UI-SPEC */} +
+
+ ) + } + + if (state === 'reconnecting') { + return ( +
+ {/* 8px pulsing muted dot — animation via keyframes in CSS */} +
+ + Reconnecting… + +
+ ) + } + + // disconnected — role="alert" for assertive announcement + return ( +
+ {/* 8px filled red dot — --color-destructive (#DC2626) */} +
+ + Updates paused + +
+ ) +} diff --git a/apps/pwa/src/hooks/useListSSE.test.ts b/apps/pwa/src/hooks/useListSSE.test.ts index 634f484..2c6c1ba 100644 --- a/apps/pwa/src/hooks/useListSSE.test.ts +++ b/apps/pwa/src/hooks/useListSSE.test.ts @@ -313,7 +313,7 @@ describe('useListSSE — D-11 bounded backoff', () => { }) expect(mockInstances.length).toBe(instanceCountAfterUnmount) // no new connection post-unmount - expect(firstEs.closeCalled ?? (firstEs as unknown as { close: () => void; closeCalled?: boolean }).closeCalled).toBeTruthy() + expect((firstEs as unknown as { closeCalled?: boolean }).closeCalled).toBeTruthy() }) it('uses withCredentials: true on EventSource (Pitfall 7 — session cookie)', async () => { diff --git a/apps/pwa/src/hooks/useListSSE.ts b/apps/pwa/src/hooks/useListSSE.ts new file mode 100644 index 0000000..b3a3507 --- /dev/null +++ b/apps/pwa/src/hooks/useListSSE.ts @@ -0,0 +1,114 @@ +/** + * useListSSE — Bounded-backoff EventSource wrapper for live list sync (LIST-04). + * + * Decisions implemented: + * D-10: Full refetch on reconnect — invalidates ['list', listId] on open. + * D-11: Silent bounded-backoff reconnect (250ms→500ms→1s→2s→4s→cap 8s). + * After MAX_ATTEMPTS failures: onStateChange('disconnected'), stop retrying. + * D-12: Polling fallback (refetchInterval:30000) lives in ListDetail — always active. + * + * Security (Pitfall 7, T-04-01): + * withCredentials: true ensures the OIDC session cookie is sent. + * + * Implementation notes (Pitfall 3 — no reconnect storm): + * On onerror: close the EventSource BEFORE scheduling setTimeout. + * This prevents the browser's built-in reconnect stacking with our manual one. + * Refs (not state) for esRef/attemptsRef/timerRef — avoids re-render loops. + * + * Source: RESEARCH.md Finding 4 verbatim pattern. + */ + +import { useEffect, useRef, useCallback } from 'react' +import { useQueryClient } from '@tanstack/react-query' + +export type SyncState = 'connected' | 'reconnecting' | 'disconnected' + +/** + * Backoff schedule per D-11: 250ms → 500ms → 1000ms → 2000ms → 4000ms → cap 8000ms. + * 6 steps → MAX_ATTEMPTS = 6; after exhaustion, state → 'disconnected'. + */ +const BACKOFF_STEPS_MS = [250, 500, 1000, 2000, 4000, 8000] +const MAX_ATTEMPTS = BACKOFF_STEPS_MS.length + +export interface UseListSSEOptions { + listId: number + onStateChange: (state: SyncState) => void +} + +/** + * Opens a scoped SSE connection to /api/sse/lists and invalidates React Query + * cache on events — driving a full refetch (D-10). + * + * Lifecycle: + * Mount → connect() → EventSource opened + * Open → attemptsRef reset → onStateChange('connected') → invalidate cache + * Event → invalidate ['list', listId] + * Error → es.close() → if attempts left: onStateChange('reconnecting') + setTimeout(connect, backoff) + * else: onStateChange('disconnected') — stop + * Unmount → es.close() + clearTimeout(timerRef) + */ +export function useListSSE({ listId, onStateChange }: UseListSSEOptions): void { + const queryClient = useQueryClient() + const esRef = useRef(null) + const attemptsRef = useRef(0) + const timerRef = useRef | null>(null) + + // handleListChange is stable — invalidate the list on any list-change event + const handleListChange = useCallback(() => { + queryClient.invalidateQueries({ queryKey: ['list', listId] }) + }, [queryClient, listId]) + + const connect = useCallback(() => { + // Close any existing connection (prevents stacking) + esRef.current?.close() + + const es = new EventSource('/api/sse/lists', { withCredentials: true }) + esRef.current = es + + // Subscribe to all list-change event types (D-10 — trigger refetch on any change) + es.addEventListener('item:added', handleListChange) + es.addEventListener('item:updated', handleListChange) + es.addEventListener('item:deleted', handleListChange) + es.addEventListener('list:updated', handleListChange) + es.addEventListener('list:deleted', handleListChange) + + es.onopen = () => { + // Reset attempt counter on successful open (D-11) + attemptsRef.current = 0 + onStateChange('connected') + // Full refetch on (re)connect — D-10: no Last-Event-ID replay, just refetch + queryClient.invalidateQueries({ queryKey: ['list', listId] }) + } + + es.onerror = () => { + // Close BEFORE scheduling retry — prevents browser auto-reconnect stacking (Pitfall 3) + es.close() + + const attempt = attemptsRef.current + + if (attempt >= MAX_ATTEMPTS) { + // Backoff exhausted — surface "Updates paused" indicator and stop (D-11) + onStateChange('disconnected') + // D-12 polling fallback (refetchInterval:30000 in ListDetail) keeps data fresh + return + } + + onStateChange('reconnecting') + const delay = BACKOFF_STEPS_MS[attempt] + attemptsRef.current = attempt + 1 + timerRef.current = setTimeout(connect, delay) + } + }, [listId, queryClient, onStateChange, handleListChange]) + + useEffect(() => { + connect() + return () => { + // Cleanup: close EventSource + cancel any pending reconnect timer (Pitfall 3) + esRef.current?.close() + if (timerRef.current !== null) { + clearTimeout(timerRef.current) + timerRef.current = null + } + } + }, [connect]) +} diff --git a/apps/pwa/src/routes/ListDetail.tsx b/apps/pwa/src/routes/ListDetail.tsx index ed59f12..50dc313 100644 --- a/apps/pwa/src/routes/ListDetail.tsx +++ b/apps/pwa/src/routes/ListDetail.tsx @@ -1,7 +1,7 @@ /** * ListDetail — Single list view (/lists/:listId). * - * Replaces the Plan 04-01 placeholder. Delivers LIST-02 + LIST-03: + * Delivers LIST-02 + LIST-03 + LIST-04: * - Fetches items via useQuery(['list', listId]) with 30s polling fallback (D-12) * - Splits items into active (!checked, sorted rank ASC) and completed (checked) (D-05) * - Optimistic mutations for add / check / delete (D-07/D-09) @@ -14,7 +14,14 @@ * - Remote reorders animate via CSS.Transform.toString in ItemRow (D-14) * - Concurrent reorder converges via server last-write-wins (D-15) * - * Live sync (SSE) is wired in Plan 06. + * Live sync (LIST-04, D-10/D-11/D-12): + * - useListSSE wires /api/sse/lists with bounded backoff reconnect (D-11) + * - On SSE event: invalidates ['list', listId] → background refetch (D-10) + * - LiveSyncIndicator renders connection status in the header + * - refetchInterval:30000 polling fallback always active (D-12) + * - Note: SSE connection lives in ListDetail (one stream per list navigation). + * A future optimization could hoist this to the Lists route level — acceptable + * for Phase 4 per RESEARCH note; Phase 5 push will own the session lifecycle. * * Security: item text rendered as plain-text JSX children (T-04-06 XSS guard). */ @@ -47,6 +54,9 @@ import { } from '../api/listsClient.js' import { ItemRow } from '../components/ItemRow.js' import { AddItemInput } from '../components/AddItemInput.js' +import { LiveSyncIndicator } from '../components/LiveSyncIndicator.js' +import { useListSSE } from '../hooks/useListSSE.js' +import type { SyncState } from '../hooks/useListSSE.js' import type { ListItem, ListItemsResponse } from '../api/listsClient.js' /** @@ -97,6 +107,13 @@ export function ListDetail() { const parsedListId = Number(listId) const [completedExpanded, setCompletedExpanded] = useState(true) + const [syncState, setSyncState] = useState('connected') + + // ── Live sync (LIST-04) ────────────────────────────────────────────────── + // Bounded-backoff SSE hook (D-11). On open: invalidates ['list', listId] (D-10). + // On event: invalidates ['list', listId] → background refetch. + // Polling fallback: refetchInterval:30000 below stays always active (D-12). + useListSSE({ listId: parsedListId, onStateChange: setSyncState }) // ── dnd-kit sensors ──────────────────────────────────────────────────────── // PointerSensor: desktop mouse — immediate activation @@ -363,6 +380,9 @@ export function ListDetail() { Plan 05/06 can enrich from the ['lists'] cache by listId. */} List + + {/* Live sync indicator — connected/reconnecting/disconnected (LIST-04, D-11) */} +
{/* Content area */}