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 } 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<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',
	'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<StreamTicketResponse>('/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<void> {
	try {
		const response = await api.get<DeltaResponse>(
			`/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();
		}
	});
}