// Thin EventSource wrapper with a per-event-type registration pattern. // // Subscribers register via `onSseEvent(type, handler)` and receive the raw payload // string. The list of event types we know how to relay lives in `KNOWN_EVENTS` so // adding one new is one constant entry — keeps the file friendly to extension. // // Lifecycle: // - Connection survives backgrounding via the visibility listener at the bottom of // this file (closes on hidden, reopens on visible). // - On reopen we fire a `feed-delta` synthetic event with the gap since last seen // to whoever subscribes. The feed page is the typical consumer; it merges the // delta into its in-memory list. import { getToken } from './auth'; import { api, ApiError } from './api'; import type { DeltaResponse } from './types'; type StreamTicketResponse = { ticket: string; server_time: string }; type EventHandler = (data: string) => void; let eventSource: EventSource | null = null; let lastEventTime: string | null = null; const handlers: Map = new Map(); /** Consecutive reconnect attempts since last successful onopen. Reset on success. */ let reconnectAttempt = 0; let reconnectTimer: ReturnType | null = null; /** * SSE event names emitted by the backend. Add new ones here as `state.sse_tx.send` * call sites grow — every entry becomes a relay registration below. */ const KNOWN_EVENTS = [ 'new-upload', 'upload-processed', 'upload-error', 'upload-deleted', 'like-update', 'new-comment', 'comment-deleted', 'user-hidden', 'event-closed', 'event-opened', 'event-updated', 'export-progress', 'export-available', 'pin-reset', // A guest asked a host to reset their PIN — hosts refresh their pending-request badge. 'pin-reset-requested' ] as const; /** * Synthetic event types — not emitted by the server, dispatched locally to fan out * cross-cutting state changes (e.g. delta-fetch results after a reconnect). */ export type SyntheticEvent = 'feed-delta'; export function onSseEvent(eventType: string, handler: EventHandler): () => void { if (!handlers.has(eventType)) { handlers.set(eventType, []); } handlers.get(eventType)!.push(handler); return () => { const list = handlers.get(eventType); if (list) { const idx = list.indexOf(handler); if (idx >= 0) list.splice(idx, 1); } }; } export function connectSse(): void { const token = getToken(); if (!token || eventSource) return; // EventSource can't send an Authorization header, so we exchange the JWT for // a short-lived single-use ticket via POST /stream/ticket (Bearer auth) and // pass that on the URL. The JWT itself never appears in URLs / access logs. void (async () => { let ticket: string; let serverTime: string; try { const res = await api.post('/stream/ticket', {}); ticket = res.ticket; serverTime = res.server_time; } catch { // Failed to mint a ticket (auth lapse, network blip). Back off and retry // via the existing error path. scheduleReconnect(); return; } // Auth flow may have torn things down while we were awaiting the ticket. if (!getToken() || eventSource) return; eventSource = new EventSource(`/api/v1/stream?ticket=${encodeURIComponent(ticket)}`); eventSource.onopen = () => { // Successful connection — reset the backoff counter. reconnectAttempt = 0; // If we have a previous timestamp this is a reconnect — fetch the gap. The // delta advances `lastEventTime` from the SERVER clock it returns. const since = lastEventTime; if (since) { void deltaFetchAndFan(since); } else { // First connect: seed the cursor from the server clock at ticket-mint time, // never `new Date()` — a skewed browser clock would otherwise shift the very // first reconnect window and could drop uploads. lastEventTime = serverTime; } }; for (const eventName of KNOWN_EVENTS) { eventSource.addEventListener(eventName, (e) => dispatch(eventName, (e as MessageEvent).data)); } // `resync` is emitted by the server when our broadcast subscription fell // behind and events were dropped. Rather than let those losses leave the feed // stale, fetch the gap since the last event we actually saw and fan it out as // a feed-delta (which reconciles new uploads AND deletions). Handled with its // own listener — not via `dispatch` — so reading `lastEventTime` as the gap // start isn't clobbered by dispatch bumping it to "now". eventSource.addEventListener('resync', () => { const since = lastEventTime; if (since) void deltaFetchAndFan(since); }); eventSource.onerror = () => { // EventSource auto-reconnects but the connection state can stay broken; close // and try again ourselves with exponential backoff capped at 60s. Prevents // retry storms (and lets the backend recover quietly) when the server is down // for a while or when 100+ guests reconnect simultaneously after an outage. disconnectSse(); scheduleReconnect(); }; })(); } function scheduleReconnect(): void { reconnectAttempt++; const delay = Math.min(60_000, 1_000 * 2 ** (reconnectAttempt - 1)); const jitter = Math.random() * 500; if (reconnectTimer) clearTimeout(reconnectTimer); reconnectTimer = setTimeout(connectSse, delay + jitter); } export function disconnectSse(): void { if (eventSource) { eventSource.close(); eventSource = null; } if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } } export function getLastEventTime(): string | null { return lastEventTime; } export function setLastEventTime(time: string): void { lastEventTime = time; } function dispatch(eventType: string, data: string): void { // Advance the reconnect cursor from the SERVER timestamp carried in the payload (when // present — e.g. a new upload's `created_at`), never the browser clock. Events without // a timestamp (likes, lock toggles) leave the cursor where it is; the next delta // re-fetches from the last content timestamp we saw, which merges idempotently. const ts = extractCreatedAt(data); if (ts) lastEventTime = ts; const list = handlers.get(eventType); if (list) { for (const handler of list) { handler(data); } } } /** Pull an ISO `created_at` out of an event payload if it has one, else undefined. */ function extractCreatedAt(data: string): string | undefined { try { const parsed = JSON.parse(data); if (parsed && typeof parsed.created_at === 'string') return parsed.created_at; } catch { // non-JSON payload (e.g. a plain count) — no timestamp to extract } return undefined; } /** * Fetch all feed activity since `since` and fan it out as a synthetic `feed-delta` * event. Subscribers (typically the feed page) merge the result into their * in-memory list. Swallows errors — a failed delta is non-fatal; the next live * SSE event will keep the feed moving. */ async function deltaFetchAndFan(since: string, attempt = 0): Promise { try { const response = await api.get(`/feed/delta?since=${encodeURIComponent(since)}`); // Advance the cursor to the server clock this delta was computed at, so the next // reconnect resumes exactly where the server left off (no browser-clock skew). lastEventTime = response.server_time; dispatch('feed-delta', JSON.stringify(response)); } catch (e) { // A throttled delta (429) must NOT be silently dropped: live events keep advancing // `lastEventTime`, so the next reconnect would resume PAST this un-fetched gap and // lose it. Retry the SAME `since` with backoff so the gap [since, now] is still // covered regardless of how the live cursor moves in the meantime. Bounded, and only // reachable by a rapidly flapping EventSource hitting the per-user delta limit. if (e instanceof ApiError && e.status === 429 && attempt < MAX_DELTA_RETRIES) { const delayMs = DELTA_RETRY_BASE_MS * 2 ** attempt; setTimeout(() => void deltaFetchAndFan(since, attempt + 1), delayMs); } // Other errors are non-fatal — the next live SSE event keeps the feed moving. } } /** Bounded backoff for a rate-limited reconnect delta (see `deltaFetchAndFan`). */ const MAX_DELTA_RETRIES = 4; const DELTA_RETRY_BASE_MS = 2000; // Page Visibility API: close while hidden, reopen on focus. On reopen `connectSse`'s // `onopen` runs the delta fetch. function handleVisibilityChange() { if (document.hidden) { disconnectSse(); } else { // User-initiated reconnect — clear backoff so we don't wait out a long // retry delay that was scheduled from a prior background error. reconnectAttempt = 0; connectSse(); } } let visibilityBound = false; /** Idempotent: safe to call more than once; only the first registration sticks. */ function bindVisibility() { if (visibilityBound || typeof document === 'undefined') return; document.addEventListener('visibilitychange', handleVisibilityChange); visibilityBound = true; } /** Remove the visibility listener (e.g. on teardown / test cleanup). */ export function teardownVisibility() { if (!visibilityBound || typeof document === 'undefined') return; document.removeEventListener('visibilitychange', handleVisibilityChange); visibilityBound = false; } bindVisibility();