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;

/**
 * True once a stream has opened at least once this session. Distinguishes "first
 * connect" from "reconnect" now that `lastEventTime` is seeded at ticket-mint rather
 * than in `onopen` (see `connectSse`) — without it every boot would fire a pointless
 * zero-width delta.
 */
let streamEverOpened = false;

/**
 * `Date.now()` of the last thing we actually received on the live stream.
 *
 * Keep-alives are deliberately NOT observable here: the backend sends them as SSE
 * COMMENTS (`KeepAlive::new().text("ping")` emits `:ping`), and the EventSource parser
 * discards comments without dispatching anything at all. There is no browser API that
 * exposes them. So liveness cannot be decided by a plain silence timer — it could not
 * tell a dead socket from a genuinely quiet half hour, and would churn reconnects for
 * every guest through every lull. The backstop poll below decides it on evidence
 * instead: the server had news that this stream never delivered.
 */
let lastStreamActivity = 0;

/**
 * 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',
	// The mirror of `user-hidden`: a host lifted a ban, so the guest's uploads return to every
	// feed and the projector, and their own parked upload queue resumes.
	'user-shown',
	'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;
		}
		// Seed the reconnect cursor the moment we have a server clock — NOT in `onopen`,
		// which is where it used to live. `onopen` never fires behind a captive portal or
		// any proxy that buffers `text/event-stream`, so on exactly the networks where the
		// stream fails the cursor stayed `null` forever and the backstop poll below had no
		// `since` to fetch from. Still the server clock and never `new Date()`: a skewed
		// browser clock would shift the window and silently drop uploads.
		if (!lastEventTime) lastEventTime = serverTime;

		// Auth flow may have torn things down while we were awaiting the ticket — and the
		// phone may have gone to sleep during that round-trip. Opening a stream while
		// hidden is worse than not opening one: iOS reaps a backgrounded socket without
		// ever firing `onerror`, so `eventSource` stays non-null and the guard at the top
		// of this function then treats the corpse as a live connection for the rest of the
		// evening. The visibility handler reconnects us when the screen comes back.
		if (!getToken() || eventSource || (typeof document !== 'undefined' && document.hidden)) return;

		eventSource = new EventSource(`/api/v1/stream?ticket=${encodeURIComponent(ticket)}`);

		eventSource.onopen = () => {
			// Successful connection — reset the backoff counter.
			reconnectAttempt = 0;
			noteStreamActivity();
			// A reconnect has a gap to close; the very first open of a session does not
			// (the cursor was just seeded from this ticket's server clock). The delta
			// advances `lastEventTime` from the SERVER clock it returns.
			if (streamEverOpened && lastEventTime) void deltaFetchAndFan(lastEventTime);
			streamEverOpened = true;
		};

		for (const eventName of KNOWN_EVENTS) {
			eventSource.addEventListener(eventName, (e) => {
				noteStreamActivity();
				const data = (e as MessageEvent).data;
				noteDelivered(eventName, data);
				dispatch(eventName, 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', () => {
			noteStreamActivity();
			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));
	// Jitter must SCALE WITH the backoff, not be a flat 500ms. Every client that dropped
	// together shares the same `reconnectAttempt`, so they compute an identical `delay` —
	// a fixed 500ms window spreads 100 phones over half a second no matter how long the
	// backoff grew, which is the thundering herd the backoff exists to prevent. Each
	// reconnect costs a ticket POST + stream GET + feed-delta fetch, so the herd lands on
	// the DB pool three times over. Scaling the jitter to the delay (floored at 1s so the
	// first, most synchronised retry is spread too) turns that into a smooth ramp.
	const jitter = Math.random() * Math.max(delay, 1_000);
	if (reconnectTimer) clearTimeout(reconnectTimer);
	reconnectTimer = setTimeout(connectSse, delay + jitter);
}

export function disconnectSse(): void {
	if (eventSource) {
		eventSource.close();
		eventSource = null;
	}
	// A new stream has delivered nothing, so anything the next delta returns is genuinely
	// undelivered as far as THIS connection is concerned. Keeping the old set would suppress the
	// liveness signal for content that arrived during the gap — the opposite failure to H1.
	forgetDelivered();
	if (reconnectTimer) {
		clearTimeout(reconnectTimer);
		reconnectTimer = null;
	}
}

function noteStreamActivity(): void {
	lastStreamActivity = Date.now();
}

// ── Stream backstop ────────────────────────────────────────────────────────────────
//
// EVERY feed update is triggered by an SSE event — there is no periodic refetch — so a
// stream that stops delivering freezes a guest's gallery for the rest of the evening.
// And a venue produces exactly the two failures that the reconnect path cannot see:
//
//   • a captive portal or any proxy that buffers `text/event-stream` never forwards a
//     byte, so `onopen` may never fire and neither does `onerror`;
//   • a phone that roams between APs leaves a HALF-OPEN socket — the connection is gone
//     but `readyState` still reads OPEN, no error is raised, and `connectSse` early-
//     returns on its non-null `eventSource`, so nothing ever reconnects.
//
// Neither surfaces anything to react to, which is why the only honest backstop is to
// ask the server. `/feed/delta` is idempotent and answers with an empty payload when
// nothing changed, and one request per 60–120s is ~1/60th of the per-user feed limit —
// cheap insurance against a guest staring at a frozen feed all night. The diashow uses
// the same reconcile-on-a-timer for the same reason (`RECONCILE_INTERVAL_MS` there).
const BACKSTOP_MIN_MS = 60_000;
const BACKSTOP_MAX_MS = 120_000;
let backstopTimer: ReturnType<typeof setTimeout> | null = null;
let backstopEnabled = false;

/**
 * Start the poll-based liveness/completeness backstop. Opt-in per page (the feed is the
 * consumer that needs it; the export page's SSE use is a status ping and the diashow
 * runs its own full reconcile), and idempotent.
 */
export function startStreamBackstop(): void {
	if (typeof document === 'undefined' || backstopEnabled) return;
	backstopEnabled = true;
	scheduleBackstop();
}

export function stopStreamBackstop(): void {
	backstopEnabled = false;
	if (backstopTimer) {
		clearTimeout(backstopTimer);
		backstopTimer = null;
	}
}

function scheduleBackstop(): void {
	if (backstopTimer) clearTimeout(backstopTimer);
	// Jittered for the same reason the reconnect backoff is: ~100 phones that joined
	// within the same few minutes would otherwise poll in permanent lockstep.
	const delay = BACKSTOP_MIN_MS + Math.random() * (BACKSTOP_MAX_MS - BACKSTOP_MIN_MS);
	backstopTimer = setTimeout(() => void runBackstop(), delay);
}

async function runBackstop(): Promise<void> {
	backstopTimer = null;
	try {
		// A hidden tab has no stream (the visibility handler closed it) and cannot show a
		// result anyway; the reopen path already fetches the gap.
		if (document.hidden || !getToken()) return;
		// No stream, or one the browser has admitted is closed, while we are visible: there
		// is no other way back, because `connectSse`'s `eventSource` guard cannot tell a
		// corpse from a live connection.
		if (!eventSource || eventSource.readyState === EventSource.CLOSED) {
			disconnectSse();
			reconnectAttempt = 0;
			connectSse();
			return;
		}
		const since = lastEventTime;
		if (!since) return;
		const activityBefore = lastStreamActivity;
		const carried = await deltaFetchAndFan(since);
		// The server had news that this stream never delivered. That is the evidence a
		// half-open socket cannot otherwise give us — reconnect, or every remaining update
		// tonight arrives at poll latency instead of instantly.
		if (carried && lastStreamActivity === activityBefore) {
			disconnectSse();
			reconnectAttempt = 0;
			connectSse();
		}
	} finally {
		if (backstopEnabled) scheduleBackstop();
	}
}

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);
		}
	}
}

/**
 * Ids the LIVE STREAM has actually delivered to us — the evidence the liveness backstop needs.
 *
 * The backstop asks "did the poll find something the stream never pushed?", and the old answer was
 * simply "did the poll return any rows?". Those are not the same question, and on a live event they
 * diverge constantly: `dispatch` advances the cursor to an upload's `created_at`, which is always
 * EARLIER than the `server_time` the previous delta stored, so the cursor rewinds onto
 * `feed_delta`'s deliberately inclusive `>=` boundary and the poll re-returns an upload the stream
 * had already delivered a moment ago. Row count > 0, so the backstop concluded the socket was dead
 * and tore down a perfectly healthy stream — then reset `reconnectAttempt` to 0, bypassing the
 * jittered backoff. At 100 guests that is roughly one reconnect per second, sustained, all evening,
 * each costing ~10 queries.
 *
 * Bounded FIFO: an event runs for hours and this must not grow without limit. The cap only needs to
 * exceed what one delta window can return (`DELTA_LIMIT` server-side), because anything older than
 * the current window cannot be re-returned as "new".
 */
const DELIVERED_MEMORY = 500;
const deliveredIds: string[] = [];
const deliveredSet = new Set<string>();

function rememberDelivered(id: string): void {
	if (deliveredSet.has(id)) return;
	deliveredSet.add(id);
	deliveredIds.push(id);
	if (deliveredIds.length > DELIVERED_MEMORY) {
		const evicted = deliveredIds.shift();
		if (evicted !== undefined) deliveredSet.delete(evicted);
	}
}

/** Record whatever ids a stream payload carried, so a later delta can be recognised as a repeat.
 *
 * Only the id that IS the thing the liveness check tests, per event — not every id field present.
 *
 * This used to harvest `id`, `upload_id` and `user_id` from every payload, which quietly disarmed
 * two thirds of the backstop. `new-upload` carries `id` + `user_id`, and `like-update` /
 * `new-comment` carry `upload_id` + `user_id`, so by the time anything was deleted or anyone was
 * banned their ids were already in `deliveredSet` — recorded from ordinary traffic about content
 * that was still perfectly live. `carried`'s `deleted_ids` and `hidden_user_ids` clauses were then
 * false essentially always.
 *
 * The cost: a socket that goes half-open (a phone roaming APs leaves `readyState === OPEN`, so the
 * cheap check misses it) is only noticed once a genuinely NEW upload appears. A host moderating
 * three photos, or banning a guest, produced a delta whose every id was "already delivered" — so
 * the stream stayed dead and the host kept moderating into a feed nobody's app was listening to.
 */
function noteDelivered(eventName: string, data: string): void {
	try {
		const p = JSON.parse(data) as { id?: unknown; upload_id?: unknown; user_id?: unknown };
		// Mirrors the three clauses in `carried`: uploads by upload id, deletions by upload id,
		// ban-hides by user id.
		const relevant =
			eventName === 'new-upload'
				? p.id
				: // `upload-processed` carries `upload_id`, not `id` (see compression.rs) — reading
					// `p.id` recorded nothing at all, so this branch quietly did the opposite of what
					// the comment above claims. Harmless today only because the delta cursor is
					// anchored by `new-upload`, which is not a property worth depending on.
					eventName === 'upload-processed' || eventName === 'upload-deleted'
					? (p.upload_id ?? p.id)
					: // Only `user-hidden` has a matching clause in `carried` (`hidden_user_ids` comes
						// from `uploads_hidden = TRUE`). Recording an UNBAN's user id could only ever
						// suppress a later genuine signal, so it is deliberately not recorded.
						eventName === 'user-hidden'
						? p.user_id
						: undefined;
		if (typeof relevant === 'string') rememberDelivered(relevant);
	} catch {
		// non-JSON payload — nothing to record
	}
}

/** Reset on disconnect: a fresh stream has delivered nothing yet. */
function forgetDelivered(): void {
	deliveredIds.length = 0;
	deliveredSet.clear();
}

/** 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.
 *
 * Resolves to whether the delta actually CARRIED something. `runBackstop` uses that as
 * its liveness signal: content the poll found but the stream never pushed means the
 * stream is dead in the way the browser will not report.
 */
async function deltaFetchAndFan(since: string, attempt = 0): Promise<boolean> {
	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));
		// "Did the poll find something the STREAM never delivered?" — not "did it return rows?".
		// See `deliveredIds`: on a live event the cursor rewinds onto the inclusive `>=` boundary
		// and re-returns uploads the stream already pushed, so a row count made every healthy
		// stream look dead and produced a sustained reconnect storm.
		//
		// Ids the stream delivered while we were connected are excluded. Anything genuinely new —
		// including everything that arrived while the socket was half-open — still counts, which is
		// the signal this backstop exists for.
		return (
			response.uploads.some((u) => !deliveredSet.has(u.id)) ||
			response.deleted_ids.some((id) => !deliveredSet.has(id)) ||
			response.hidden_user_ids.some((id) => !deliveredSet.has(id))
		);
	} 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) {
			// Jittered for the same reason `scheduleReconnect` is: the clients that hit this
			// 429 are the ones that just reconnected together after a venue-wide wifi blip,
			// so they share `attempt` and would otherwise retry in lockstep at exactly 2s,
			// 4s, 8s — re-tripping the same per-user limit in a synchronised wave.
			const base = DELTA_RETRY_BASE_MS * 2 ** attempt;
			setTimeout(() => void deltaFetchAndFan(since, attempt + 1), base + Math.random() * base);
		}
		// Other errors are non-fatal — the next live SSE event keeps the feed moving.
		return false;
	}
}

/** 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 {
		// Tear down UNCONDITIONALLY before reconnecting rather than leaning on
		// `connectSse`'s `eventSource` guard. iOS can reap a backgrounded socket without
		// ever firing `onerror`, which leaves a non-null but permanently dead EventSource —
		// and the guard would then read that as "already connected" and never reconnect,
		// for the rest of the evening. Closing an already-closed EventSource is a no-op.
		disconnectSse();
		// 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();