// 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 } from './api'; import type { DeltaResponse } from './types'; type StreamTicketResponse = { ticket: 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', 'user-banned', 'event-closed', 'event-opened', 'event-updated', 'export-progress', 'export-available', 'pin-reset' ] 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' | 'feed-reload'; /** * Advance the reconnect cursor using a **server** timestamp (an upload's * `created_at`), never the client clock. A phone clock skewed ahead would * otherwise make the cursor jump past events that happened while the tab was * backgrounded. ISO-8601 UTC strings compare correctly lexicographically. */ function noteServerTime(ts: string | null | undefined): void { if (!ts) return; if (!lastEventTime || ts > lastEventTime) lastEventTime = ts; } 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; try { const res = await api.post('/stream/ticket', {}); ticket = res.ticket; } 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 (server-derived) timestamp this is a reconnect // — fetch the gap. The cursor is only ever advanced from server // timestamps (noteServerTime), so client clock skew can't drop events. const since = lastEventTime; if (since) { void deltaFetchAndFan(since); } }; for (const eventName of KNOWN_EVENTS) { eventSource.addEventListener(eventName, (e) => dispatch(eventName, (e as MessageEvent).data) ); } 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 cursor from the server timestamp carried by a new upload. // Other event types don't carry one and must not bump it off the client clock. if (eventType === 'new-upload') { try { noteServerTime((JSON.parse(data) as { created_at?: string }).created_at); } catch { // payload not JSON — ignore } } const list = handlers.get(eventType); if (list) { for (const handler of list) { handler(data); } } } /** * 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): Promise { try { const response = await api.get( `/feed/delta?since=${encodeURIComponent(since)}` ); // Advance the cursor from the newest server timestamp in the delta. for (const u of response.uploads) noteServerTime(u.created_at); // The server clamped the window or hit the row cap — the partial delta // can't be trusted, so ask the page to do a full reload instead. if (response.reload_required) { dispatch('feed-reload', '{}'); return; } dispatch('feed-delta', JSON.stringify(response)); } catch { // non-fatal } } // Page Visibility API: close while hidden, reopen on focus. On reopen `connectSse`'s // `onopen` runs the delta fetch. if (typeof document !== 'undefined') { document.addEventListener('visibilitychange', () => { 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(); } }); }