Blame · Line-by-line history

`sse.ts`

Each line is annotated with the commit that last touched it. Click any SHA to jump to that commit and see the surrounding change.

sse.tsBlame71 lines · 1 contributor

Normal view Raw

febd4f0Claude	1	/**
	2	* In-process topic-based pub/sub broadcaster for Server-Sent Events.
	3	*
	4	* Module-level `Map<topic, Set<handler>>`. `publish` iterates subscribers
	5	* synchronously (fire-and-forget). Handlers are expected not to throw — we
	6	* swallow exceptions defensively so one misbehaving subscriber cannot take
	7	* down the publisher or starve its peers.
	8	*
	9	* TODO(scale): this is deliberately single-process / in-memory. Horizontally
	10	* scaled deploys (multiple Bun instances behind a load balancer) will need
	11	* a cross-node fanout layer — likely Redis pub/sub or NATS — that feeds this
	12	* local broadcaster on each node. Until then, SSE subscribers only receive
	13	* events published by the same process handling their connection.
	14	*/
	15
	16	export type SSEEvent = {
	17	event?: string;
	18	data: unknown;
	19	id?: string;
	20	};
	21
	22	type Handler = (event: SSEEvent) => void;
	23
	24	const topics = new Map<string, Set<Handler>>();
	25
	26	/**
	27	* Publish an event to every subscriber of `topic`. No-op if the topic has
	28	* no subscribers. Handler exceptions are caught and swallowed so a single
	29	* broken subscriber cannot break fanout for its peers.
	30	*/
	31	export function publish(topic: string, event: SSEEvent): void {
	32	const subs = topics.get(topic);
	33	if (!subs \|\| subs.size === 0) return;
	34	for (const handler of subs) {
	35	try {
	36	handler(event);
	37	} catch {
	38	// Swallow — handlers are fire-and-forget and must not disrupt fanout.
	39	}
	40	}
	41	}
	42
	43	/**
	44	* Register a handler for `topic`. Returns a cleanup function that removes
	45	* the handler (and drops the topic's entry when its last subscriber leaves).
	46	*/
	47	export function subscribe(
	48	topic: string,
	49	handler: Handler
	50	): () => void {
	51	let subs = topics.get(topic);
	52	if (!subs) {
	53	subs = new Set<Handler>();
	54	topics.set(topic, subs);
	55	}
	56	subs.add(handler);
	57
	58	return () => {
	59	const current = topics.get(topic);
	60	if (!current) return;
	61	current.delete(handler);
	62	if (current.size === 0) {
	63	topics.delete(topic);
	64	}
	65	};
	66	}
	67
	68	/** Number of active subscribers on a topic (0 if unknown). */
	69	export function topicSubscriberCount(topic: string): number {
	70	return topics.get(topic)?.size ?? 0;
	71	}