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.tsBlame150 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
40e7738Claude	32	// kind is lowercase letters; id may include colons so multi-segment topics
	33	// like `ai:repo:<uuid>` parse cleanly into kind="ai", id="repo:<uuid>".
	34	const TOPIC_RE = /^[a-z]+:[a-zA-Z0-9:\-]+$/;
febd4f0Claude	35	const HEARTBEAT_MS = 25_000;
	36
	37	app.get("/live-events/:topic", softAuth, async (c) => {
	38	const topic = c.req.param("topic");
	39	if (!topic \|\| !TOPIC_RE.test(topic)) {
	40	return c.json({ error: "Invalid topic" }, 400);
	41	}
	42
	43	const user = c.get("user") ?? null;
	44	const colon = topic.indexOf(":");
	45	const kind = topic.slice(0, colon);
	46	const id = topic.slice(colon + 1);
	47
	48	// For repo topics, gate on read access. Other topic kinds pass through.
	49	if (kind === "repo") {
	50	try {
	51	const [repo] = await db
	52	.select({ id: repositories.id, isPrivate: repositories.isPrivate })
	53	.from(repositories)
	54	.where(eq(repositories.id, id))
	55	.limit(1);
	56
	57	if (!repo) {
	58	return c.json({ error: "Not found" }, 404);
	59	}
	60
	61	const access = await resolveRepoAccess({
	62	repoId: repo.id,
	63	userId: user?.id ?? null,
	64	isPublic: !repo.isPrivate,
	65	});
	66
	67	if (!satisfiesAccess(access, "read")) {
	68	return c.json({ error: "Forbidden" }, 403);
	69	}
	70	} catch {
	71	return c.json({ error: "Not found" }, 404);
	72	}
	73	}
	74
	75	const encoder = new TextEncoder();
	76
	77	const stream = new ReadableStream<Uint8Array>({
	78	start(controller) {
	79	let closed = false;
	80
	81	const safeEnqueue = (chunk: string) => {
	82	if (closed) return;
	83	try {
	84	controller.enqueue(encoder.encode(chunk));
	85	} catch {
	86	// Controller already closed — mark local state so we stop trying.
	87	closed = true;
	88	}
	89	};
	90
	91	// Initial comment flushes headers on some proxies.
	92	safeEnqueue(": open\n\n");
	93
	94	const unsubscribe = subscribe(topic, (event: SSEEvent) => {
	95	let payload = "";
	96	if (event.id !== undefined) payload += `id: ${event.id}\n`;
	97	if (event.event !== undefined) payload += `event: ${event.event}\n`;
	98	const data =
	99	typeof event.data === "string"
	100	? event.data
	101	: JSON.stringify(event.data);
	102	// SSE `data:` lines must not contain raw newlines — split if present.
	103	for (const line of data.split("\n")) {
	104	payload += `data: ${line}\n`;
	105	}
	106	payload += "\n";
	107	safeEnqueue(payload);
	108	});
	109
	110	const heartbeat = setInterval(() => {
	111	safeEnqueue(": ping\n\n");
	112	}, HEARTBEAT_MS);
	113
	114	const cleanup = () => {
	115	if (closed) return;
	116	closed = true;
	117	clearInterval(heartbeat);
	118	unsubscribe();
	119	try {
	120	controller.close();
	121	} catch {
	122	// Already closed — nothing to do.
	123	}
	124	};
	125
	126	// Client-side abort (navigation, tab close) surfaces via the request's
	127	// AbortSignal. Bun's fetch-style request exposes this on `c.req.raw`.
	128	const signal = c.req.raw.signal;
	129	if (signal) {
	130	if (signal.aborted) {
	131	cleanup();
	132	} else {
	133	signal.addEventListener("abort", cleanup, { once: true });
	134	}
	135	}
	136	},
	137	});
	138
	139	return new Response(stream, {
	140	status: 200,
	141	headers: {
	142	"Content-Type": "text/event-stream; charset=utf-8",
	143	"Cache-Control": "no-cache, no-transform",
	144	Connection: "keep-alive",
	145	"X-Accel-Buffering": "no",
	146	},
	147	});
	148	});
	149
	150	export default app;