EventSnap/frontend/src/lib/sse.ts

// 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<string, EventHandler[]> = new Map();

/** Consecutive reconnect attempts since last successful onopen. Reset on success. */
let reconnectAttempt = 0;
let reconnectTimer: ReturnType<typeof setTimeout> | 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<StreamTicketResponse>('/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<void> {
	try {
		const response = await api.get<DeltaResponse>(`/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();