Blame · Line-by-line history

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

pr-live.tsBlame239 lines · 1 contributor

Normal view Raw

3c03977Claude	1	/**
	2	* Live co-editing transport — SSE stream + REST control plane.
	3	*
	4	* GET /api/v2/pulls/:prId/live
	5	* SSE feed of presence + cursor + edit events for one PR. Auto-
	6	* joins on connect (if authed) and auto-leaves on stream close.
	7	*
	8	* POST /api/v2/pulls/:prId/live/cursor
	9	* { sessionId, position } — broadcast a cursor move.
	10	*
	11	* POST /api/v2/pulls/:prId/live/edit
	12	* { sessionId, patch } — broadcast a content patch.
	13	*
	14	* POST /api/v2/pulls/:prId/live/heartbeat
	15	* { sessionId } — keep-alive ping.
	16	*
	17	* POST /api/v2/pulls/:prId/live/leave
	18	* { sessionId } — explicit leave (the browser also fires this on
	19	* `beforeunload` via sendBeacon).
	20	*
	21	* Unauthed connections still receive the SSE feed (so anonymous repo
	22	* viewers see presence), but the POST control plane requires auth.
	23	* Agent tokens (`agt_*`) are accepted via the optional `?agent=1`
	24	* query string + Authorization header path — left to the writeup;
	25	* v1 wires humans only and falls back gracefully on missing principal.
	26	*/
	27
	28	import { Hono } from "hono";
	29	import { softAuth } from "../middleware/auth";
	30	import type { AuthEnv } from "../middleware/auth";
	31	import { subscribe, type SSEEvent } from "../lib/sse";
	32	import {
	33	joinSession,
	34	leaveSession,
	35	updateCursor,
	36	heartbeat,
	37	broadcastEdit,
	38	listLive,
	39	prLiveTopic,
	40	type CursorPosition,
	41	type EditPatch,
	42	} from "../lib/pr-live";
	43
	44	const app = new Hono<AuthEnv>();
	45
	46	// SSE heartbeat from the server side (separate from client-side
	47	// heartbeat broadcasts). Keeps intermediaries from idle-timing the
	48	// connection.
	49	const SSE_PING_MS = 25_000;
	50
	51	/** Strict UUID-ish guard (we don't lock to a specific RFC variant). */
	52	const ID_RE = /^[a-zA-Z0-9\-]{1,64}$/;
	53
	54	app.get("/api/v2/pulls/:prId/live", softAuth, async (c) => {
	55	const prId = c.req.param("prId");
	56	if (!prId \|\| !ID_RE.test(prId)) {
	57	return c.json({ error: "Invalid pr id" }, 400);
	58	}
	59
	60	const user = c.get("user") ?? null;
	61	const topic = prLiveTopic(prId);
	62
	63	// Auto-join — only humans for v1; agents will get an explicit
	64	// POST /join endpoint when the harness wants them on the stream.
	65	let sessionId: string \| null = null;
	66	let sessionColor: string \| null = null;
	67	if (user) {
	68	const joined = await joinSession({ prId, userId: user.id });
	69	if (joined) {
	70	sessionId = joined.sessionId;
	71	sessionColor = joined.color;
	72	}
	73	}
	74
	75	// Snapshot of current presence — sent as the first event so the
	76	// client can render avatars without an extra round-trip.
	77	const presence = await listLive(prId);
	78
	79	const encoder = new TextEncoder();
	80
	81	const stream = new ReadableStream<Uint8Array>({
	82	start(controller) {
	83	let closed = false;
	84	const safeEnqueue = (chunk: string) => {
	85	if (closed) return;
	86	try {
	87	controller.enqueue(encoder.encode(chunk));
	88	} catch {
	89	closed = true;
	90	}
	91	};
	92
	93	const sendEvent = (event: SSEEvent) => {
	94	let payload = "";
	95	if (event.id !== undefined) payload += `id: ${event.id}\n`;
	96	if (event.event !== undefined) payload += `event: ${event.event}\n`;
	97	const data =
	98	typeof event.data === "string"
	99	? event.data
	100	: JSON.stringify(event.data);
	101	for (const line of data.split("\n")) {
	102	payload += `data: ${line}\n`;
	103	}
	104	payload += "\n";
	105	safeEnqueue(payload);
	106	};
	107
	108	// Flush headers on proxies that buffer.
	109	safeEnqueue(": open\n\n");
	110
	111	// Initial "hello" with the snapshot + the joined session id (so
	112	// the client knows which row is theirs and can suppress echo
	113	// events from itself).
	114	sendEvent({
	115	event: "hello",
	116	data: {
	117	sessionId,
	118	color: sessionColor,
	119	presence,
	120	},
	121	});
	122
	123	const unsubscribe = subscribe(topic, sendEvent);
	124
	125	const ping = setInterval(() => {
	126	safeEnqueue(": ping\n\n");
	127	}, SSE_PING_MS);
	128
	129	const cleanup = async () => {
	130	if (closed) return;
	131	closed = true;
	132	clearInterval(ping);
	133	unsubscribe();
	134	if (sessionId) {
	135	// Fire-and-forget. The autopilot sweep is the durable
	136	// fallback if this never runs (process crash).
	137	try {
	138	await leaveSession(sessionId, prId);
	139	} catch {
	140	/* best-effort */
	141	}
	142	}
	143	try {
	144	controller.close();
	145	} catch {
	146	/* already closed */
	147	}
	148	};
	149
	150	const signal = c.req.raw.signal;
	151	if (signal) {
	152	if (signal.aborted) {
	153	void cleanup();
	154	} else {
	155	signal.addEventListener("abort", () => void cleanup(), { once: true });
	156	}
	157	}
	158	},
	159	});
	160
	161	return new Response(stream, {
	162	status: 200,
	163	headers: {
	164	"Content-Type": "text/event-stream; charset=utf-8",
	165	"Cache-Control": "no-cache, no-transform",
	166	Connection: "keep-alive",
	167	"X-Accel-Buffering": "no",
	168	},
	169	});
	170	});
	171
	172	// ---------- Control plane ----------
	173
	174	async function parseJsonBody(c: import("hono").Context): Promise<any> {
	175	try {
	176	return await c.req.json();
	177	} catch {
	178	return null;
	179	}
	180	}
	181
	182	app.post("/api/v2/pulls/:prId/live/cursor", softAuth, async (c) => {
	183	const prId = c.req.param("prId");
	184	if (!prId \|\| !ID_RE.test(prId)) return c.json({ error: "Invalid pr id" }, 400);
	185	const body = await parseJsonBody(c);
	186	const sessionId = String(body?.sessionId \|\| "");
	187	const position = body?.position as CursorPosition \| undefined;
	188	if (!sessionId \|\| !position \|\| typeof position.field !== "string") {
	189	return c.json({ error: "Invalid body" }, 400);
	190	}
	191	await updateCursor(sessionId, prId, position);
	192	return c.json({ ok: true });
	193	});
	194
	195	app.post("/api/v2/pulls/:prId/live/edit", softAuth, async (c) => {
	196	const prId = c.req.param("prId");
	197	if (!prId \|\| !ID_RE.test(prId)) return c.json({ error: "Invalid pr id" }, 400);
	198	const body = await parseJsonBody(c);
	199	const sessionId = String(body?.sessionId \|\| "");
	200	const patch = body?.patch as EditPatch \| undefined;
	201	if (!sessionId \|\| !patch \|\| typeof patch.field !== "string") {
	202	return c.json({ error: "Invalid body" }, 400);
	203	}
	204	await broadcastEdit(sessionId, prId, patch);
	205	return c.json({ ok: true });
	206	});
	207
	208	app.post("/api/v2/pulls/:prId/live/heartbeat", softAuth, async (c) => {
	209	const prId = c.req.param("prId");
	210	if (!prId \|\| !ID_RE.test(prId)) return c.json({ error: "Invalid pr id" }, 400);
	211	const body = await parseJsonBody(c);
	212	const sessionId = String(body?.sessionId \|\| "");
	213	if (!sessionId) return c.json({ error: "Invalid body" }, 400);
	214	await heartbeat(sessionId, prId);
	215	return c.json({ ok: true });
	216	});
	217
	218	app.post("/api/v2/pulls/:prId/live/leave", softAuth, async (c) => {
	219	const prId = c.req.param("prId");
	220	if (!prId \|\| !ID_RE.test(prId)) return c.json({ error: "Invalid pr id" }, 400);
	221	// Accept JSON body OR a sendBeacon Blob — sendBeacon sets
	222	// content-type to text/plain;charset=UTF-8 by default. Try JSON
	223	// first, fall back to reading raw text and parsing it.
	224	let body = await parseJsonBody(c);
	225	if (!body) {
	226	try {
	227	const raw = await c.req.text();
	228	body = raw ? JSON.parse(raw) : null;
	229	} catch {
	230	body = null;
	231	}
	232	}
	233	const sessionId = String(body?.sessionId \|\| "");
	234	if (!sessionId) return c.json({ error: "Invalid body" }, 400);
	235	await leaveSession(sessionId, prId);
	236	return c.json({ ok: true });
	237	});
	238
	239	export default app;