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