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.tsBlame148 lines · 1 contributor

Normal view Raw

6522084Claude	1	/**
	2	* Server-Sent Events (SSE) infrastructure.
	3	*
	4	* Manages live connections so gate runs, notifications, and CI updates
	5	* stream to the browser in real time instead of requiring page refreshes.
	6	*/
	7
	8	type SSEChannel = "gate" \| "notification" \| "pr" \| "ci";
	9
	10	interface SSEClient {
	11	id: string;
	12	controller: ReadableStreamDefaultController;
	13	userId?: string;
	14	channels: Set<string>; // e.g. "gate:repoId", "pr:prId", "notification:userId"
	15	connectedAt: number;
	16	}
	17
	18	const clients = new Map<string, SSEClient>();
	19
	20	let clientIdCounter = 0;
	21
	22	function nextClientId(): string {
	23	return `sse_${++clientIdCounter}_${Date.now().toString(36)}`;
	24	}
	25
	26	/**
	27	* Create an SSE Response for a Hono handler.
	28	* The caller subscribes to channels, and this function returns a streaming Response.
	29	*/
	30	export function createSSEStream(
	31	channels: string[],
	32	userId?: string
	33	): Response {
	34	const clientId = nextClientId();
	35
	36	const stream = new ReadableStream({
	37	start(controller) {
	38	const client: SSEClient = {
	39	id: clientId,
	40	controller,
	41	userId,
	42	channels: new Set(channels),
	43	connectedAt: Date.now(),
	44	};
	45	clients.set(clientId, client);
	46
	47	// Send initial connection event
	48	const data = `event: connected\ndata: ${JSON.stringify({ clientId, channels })}\n\n`;
	49	controller.enqueue(new TextEncoder().encode(data));
	50	},
	51	cancel() {
	52	clients.delete(clientId);
	53	},
	54	});
	55
	56	return new Response(stream, {
	57	headers: {
	58	"Content-Type": "text/event-stream",
	59	"Cache-Control": "no-cache",
	60	Connection: "keep-alive",
	61	"X-Accel-Buffering": "no",
	62	},
	63	});
	64	}
	65
	66	/**
	67	* Broadcast an event to all clients subscribed to a channel.
	68	*/
	69	export function broadcast(
	70	channel: string,
	71	event: string,
	72	data: unknown
	73	): number {
	74	const encoded = new TextEncoder();
	75	const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
	76	const bytes = encoded.encode(payload);
	77	let sent = 0;
	78
	79	for (const [id, client] of clients) {
	80	if (client.channels.has(channel)) {
	81	try {
	82	client.controller.enqueue(bytes);
	83	sent++;
	84	} catch {
	85	clients.delete(id);
	86	}
	87	}
	88	}
	89
	90	return sent;
	91	}
	92
	93	/**
	94	* Send an event to a specific user (by userId) across all their connections.
	95	*/
	96	export function sendToUser(
	97	userId: string,
	98	event: string,
	99	data: unknown
	100	): number {
	101	const encoded = new TextEncoder();
	102	const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
	103	const bytes = encoded.encode(payload);
	104	let sent = 0;
	105
	106	for (const [id, client] of clients) {
	107	if (client.userId === userId) {
	108	try {
	109	client.controller.enqueue(bytes);
	110	sent++;
	111	} catch {
	112	clients.delete(id);
	113	}
	114	}
	115	}
	116
	117	return sent;
	118	}
	119
	120	/**
	121	* Get active connection count (for monitoring).
	122	*/
	123	export function getActiveConnections(): number {
	124	return clients.size;
	125	}
	126
	127	/**
	128	* Clean up stale connections (call periodically).
	129	*/
	130	export function cleanupStaleConnections(maxAgeMs = 30 * 60 * 1000): number {
	131	const cutoff = Date.now() - maxAgeMs;
	132	let removed = 0;
	133
	134	for (const [id, client] of clients) {
	135	if (client.connectedAt < cutoff) {
	136	try {
	137	client.controller.close();
	138	} catch {}
	139	clients.delete(id);
	140	removed++;
	141	}
	142	}
	143
	144	return removed;
	145	}
	146
	147	// Periodic cleanup every 5 minutes
	148	setInterval(() => cleanupStaleConnections(), 5 * 60 * 1000);