Build reliable clients

import { EventBrokerSocket, type BrokerEvent } from '@thuum/client';

interface Grant {
	baseUrl: string;
	domain: string;
	token: string;
}

const MAX_BACKOFF_MS = 30_000;
const MAX_REMEMBERED_IDS = 2_000;

export class LiveTopic {
	status: 'connecting' | 'online' | 'reconnecting' = 'connecting';

	#socket: EventBrokerSocket | null = null;
	#retryTimer: ReturnType<typeof setTimeout> | null = null;
	#attempt = 0;
	#stopped = false;
	#lastTs: number | undefined;
	readonly #seen = new Set<string>();

	constructor(
		private readonly topic: string,
		private readonly onEvent: (event: BrokerEvent) => void,
		/** Your server's history route: events of `topic` since `since` (inclusive), newest first. */
		private readonly loadHistory: (since?: number) => Promise<BrokerEvent[]>
	) {
		void this.#connect();
		addEventListener('online', this.#retryNow);
	}

	stop(): void {
		this.#stopped = true;
		removeEventListener('online', this.#retryNow);
		if (this.#retryTimer) clearTimeout(this.#retryTimer);
		this.#socket?.close();
	}

	async #connect(): Promise<void> {
		try {
			const response = await fetch('/api/thuum-token', { cache: 'no-store' });
			if (!response.ok) throw new Error(`token route: ${response.status}`);
			const grant = (await response.json()) as Grant;
			if (this.#stopped) return;

			const socket = new EventBrokerSocket(grant);
			this.#socket = socket;
			socket.onMessage((event) => this.#deliver(event));
			socket.onClose(() => this.#socket === socket && this.#lost());
			await socket.ready;

			await socket.sub(this.topic); // 1. subscribe first…
			const missed = await this.loadHistory(this.#lastTs); // 2. …then catch up
			for (const event of missed.reverse()) this.#deliver(event); // oldest first

			this.#attempt = 0;
			this.status = 'online';
		} catch {
			this.#lost();
		}
	}

	#deliver(event: BrokerEvent): void {
		if (event.topic !== this.topic || this.#seen.has(event.id)) return; // 3. de-duplicate
		this.#seen.add(event.id);
		if (this.#seen.size > MAX_REMEMBERED_IDS) {
			this.#seen.delete(this.#seen.values().next().value!);
		}
		this.#lastTs = Math.max(this.#lastTs ?? 0, event.ts);
		this.onEvent(event);
	}

	#lost(): void {
		if (this.#stopped || this.#retryTimer) return;
		const socket = this.#socket;
		this.#socket = null;
		socket?.close();
		this.status = 'reconnecting';

		// Exponential backoff with jitter: 0.5–1 s, 1–2 s, 2–4 s, … up to 15–30 s.
		const ceiling = Math.min(MAX_BACKOFF_MS, 1_000 * 2 ** this.#attempt++);
		const delay = ceiling / 2 + Math.random() * (ceiling / 2);
		this.#retryTimer = setTimeout(() => {
			this.#retryTimer = null;
			void this.#connect();
		}, delay);
	}

	/** The browser is back online: don't wait for the backoff. */
	#retryNow = () => {
		if (!this.#retryTimer) return;
		clearTimeout(this.#retryTimer);
		this.#retryTimer = null;
		void this.#connect();
	};
}

// GET /api/rooms/:room/history?since=1792590697447
const events = await acme.eventBroker.data.list(`chat:${room}`, {
	since: since ? Number(since) : undefined,
	limit: 200
});

socket.onChange(apply);
await socket.ready;
await socket.sub(`board:${id}`); // 1. subscribe first…
const states = await fetch(`/api/boards/${id}/states`).then((r) => r.json()); // 2. …then read (your server lists them)
for (const s of states) apply({ topic: s.topic, key: s.key, value: s.value, version: s.version });

await socket.sub(`scoreboard:${id}`);
const { value, version } = await socket.tell(`scoreboard:${id}`, 'board');
apply({ topic: `scoreboard:${id}`, key: 'board', value, version });

const versions = new Map<string, number>();

function apply({
	topic,
	key,
	value,
	version
}: {
	topic: string;
	key: string;
	value: unknown;
	version: number | null;
}) {
	const id = `${topic}\n${key}`;
	if (version === null) {
		versions.delete(id);
		return remove(id); // deleted
	}
	if (version < (versions.get(id) ?? 0)) return; // older than what we show
	versions.set(id, version);
	render(id, value);
}

const topic = `presence:${room}`;
const write = () => socket.set(topic, clientId, { name, typing: false }).catch(() => {});

await socket.sub(topic);
await write();
const heartbeat = setInterval(write, 20_000);

addEventListener('pagehide', () => {
	clearInterval(heartbeat);
	void socket.del(topic, clientId);
});