Blame · Line-by-line history

`live-events.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.

live-events.tsBlame148 lines · 1 contributor

Normal view Raw

febd4f0Claude	1	/**
	2	* SSE endpoint: `GET /live-events/:topic`.
	3	*
	4	* Topic format: `repo:{repoId}`, `pr:{prId}`, `user:{userId}`. The regex
	5	* `^[a-z]+:[a-zA-Z0-9\-]+$` is enforced; anything else is a 400.
	6	*
	7	* Auth / authorization:
	8	* - Runs behind softAuth so we have the viewer (or null).
	9	* - For `repo:{repoId}` topics, we do a cheap DB check that the viewer has
	10	* read access via `resolveRepoAccess`. `pr:` and `user:` topics currently
	11	* only require a valid topic string — when we add PR-level privacy we'll
	12	* extend this handler in place.
	13	*
	14	* Transport:
	15	* - `text/event-stream` with keep-alive + nginx-friendly `X-Accel-Buffering`.
	16	* - We write `id:` / `event:` / `data:` blocks per SSEEvent and send a
	17	* `: ping` comment every 25s to keep intermediaries from timing out.
	18	* - On stream close we unsubscribe and clear the heartbeat timer.
	19	*/
	20
	21	import { Hono } from "hono";
	22	import { eq } from "drizzle-orm";
	23	import { softAuth } from "../middleware/auth";
	24	import type { AuthEnv } from "../middleware/auth";
	25	import { db } from "../db";
	26	import { repositories } from "../db/schema";
	27	import { resolveRepoAccess, satisfiesAccess } from "../middleware/repo-access";
	28	import { subscribe, type SSEEvent } from "../lib/sse";
	29
	30	const app = new Hono<AuthEnv>();
	31
	32	const TOPIC_RE = /^[a-z]+:[a-zA-Z0-9\-]+$/;
	33	const HEARTBEAT_MS = 25_000;
	34
	35	app.get("/live-events/:topic", softAuth, async (c) => {
	36	const topic = c.req.param("topic");
	37	if (!topic \|\| !TOPIC_RE.test(topic)) {
	38	return c.json({ error: "Invalid topic" }, 400);
	39	}
	40
	41	const user = c.get("user") ?? null;
	42	const colon = topic.indexOf(":");
	43	const kind = topic.slice(0, colon);
	44	const id = topic.slice(colon + 1);
	45
	46	// For repo topics, gate on read access. Other topic kinds pass through.
	47	if (kind === "repo") {
	48	try {
	49	const [repo] = await db
	50	.select({ id: repositories.id, isPrivate: repositories.isPrivate })
	51	.from(repositories)
	52	.where(eq(repositories.id, id))
	53	.limit(1);
	54
	55	if (!repo) {
	56	return c.json({ error: "Not found" }, 404);
	57	}
	58
	59	const access = await resolveRepoAccess({
	60	repoId: repo.id,
	61	userId: user?.id ?? null,
	62	isPublic: !repo.isPrivate,
	63	});
	64
	65	if (!satisfiesAccess(access, "read")) {
	66	return c.json({ error: "Forbidden" }, 403);
	67	}
	68	} catch {
	69	return c.json({ error: "Not found" }, 404);
	70	}
	71	}
	72
	73	const encoder = new TextEncoder();
	74
	75	const stream = new ReadableStream<Uint8Array>({
	76	start(controller) {
	77	let closed = false;
	78
	79	const safeEnqueue = (chunk: string) => {
	80	if (closed) return;
	81	try {
	82	controller.enqueue(encoder.encode(chunk));
	83	} catch {
	84	// Controller already closed — mark local state so we stop trying.
	85	closed = true;
	86	}
	87	};
	88
	89	// Initial comment flushes headers on some proxies.
	90	safeEnqueue(": open\n\n");
	91
	92	const unsubscribe = subscribe(topic, (event: SSEEvent) => {
	93	let payload = "";
	94	if (event.id !== undefined) payload += `id: ${event.id}\n`;
	95	if (event.event !== undefined) payload += `event: ${event.event}\n`;
	96	const data =
	97	typeof event.data === "string"
	98	? event.data
	99	: JSON.stringify(event.data);
	100	// SSE `data:` lines must not contain raw newlines — split if present.
	101	for (const line of data.split("\n")) {
	102	payload += `data: ${line}\n`;
	103	}
	104	payload += "\n";
	105	safeEnqueue(payload);
	106	});
	107
	108	const heartbeat = setInterval(() => {
	109	safeEnqueue(": ping\n\n");
	110	}, HEARTBEAT_MS);
	111
	112	const cleanup = () => {
	113	if (closed) return;
	114	closed = true;
	115	clearInterval(heartbeat);
	116	unsubscribe();
	117	try {
	118	controller.close();
	119	} catch {
	120	// Already closed — nothing to do.
	121	}
	122	};
	123
	124	// Client-side abort (navigation, tab close) surfaces via the request's
	125	// AbortSignal. Bun's fetch-style request exposes this on `c.req.raw`.
	126	const signal = c.req.raw.signal;
	127	if (signal) {
	128	if (signal.aborted) {
	129	cleanup();
	130	} else {
	131	signal.addEventListener("abort", cleanup, { once: true });
	132	}
	133	}
	134	},
	135	});
	136
	137	return new Response(stream, {
	138	status: 200,
	139	headers: {
	140	"Content-Type": "text/event-stream; charset=utf-8",
	141	"Cache-Control": "no-cache, no-transform",
	142	Connection: "keep-alive",
	143	"X-Accel-Buffering": "no",
	144	},
	145	});
	146	});
	147
	148	export default app;