Blame · Line-by-line history

`streaming-review.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.

streaming-review.tsBlame402 lines · 1 contributor

Normal view Raw

2be65d0Claude	1	/**
	2	* Streaming PR review routes.
	3	*
	4	* Two endpoints:
	5	*
	6	* GET /:owner/:repo/pulls/:number/review/stream
	7	* SSE endpoint — streams a real-time AI review of the PR diff.
	8	* If a review is already in progress, sends a waiting message and
	9	* polls every 2s until it finishes (max 60s), then streams the result.
	10	* On completion, saves the full review as a PR comment (idempotent).
	11	*
	12	* GET /:owner/:repo/pulls/:number/review/stream-ui
	13	* Returns an HTML fragment (<div id="stream-review">) with inline JS
	14	* that opens an EventSource to ./review/stream and renders tokens in
	15	* real-time. Can be embedded in any page via an iframe or fetch-insert.
	16	*
	17	* Neither endpoint modifies src/routes/pulls.tsx.
	18	*/
	19
	20	import { Hono } from "hono";
	21	import { eq, and } from "drizzle-orm";
	22	import { db } from "../db";
	23	import { repositories, users, pullRequests } from "../db/schema";
	24	import { softAuth } from "../middleware/auth";
	25	import type { AuthEnv } from "../middleware/auth";
	26	import {
	27	streamPrReview,
	28	isReviewStreaming,
	29	type StreamingReviewToken,
	30	} from "../lib/streaming-review";
	31
	32	const app = new Hono<AuthEnv>();
	33
	34	// ---------------------------------------------------------------------------
	35	// Helpers
	36	// ---------------------------------------------------------------------------
	37
	38	/** Resolve owner + repo from URL params. Returns null if not found. */
	39	async function resolveRepo(ownerName: string, repoName: string) {
	40	const [owner] = await db
	41	.select()
	42	.from(users)
	43	.where(eq(users.username, ownerName))
	44	.limit(1);
	45	if (!owner) return null;
	46
	47	const [repo] = await db
	48	.select()
	49	.from(repositories)
	50	.where(
	51	and(
	52	eq(repositories.ownerId, owner.id),
	53	eq(repositories.name, repoName)
	54	)
	55	)
	56	.limit(1);
	57	if (!repo) return null;
	58
	59	return { owner, repo };
	60	}
	61
	62	/** Resolve a PR by repo id + PR number. Returns null if not found. */
	63	async function resolvePr(repositoryId: string, prNumber: number) {
	64	const [pr] = await db
	65	.select()
	66	.from(pullRequests)
	67	.where(
	68	and(
	69	eq(pullRequests.repositoryId, repositoryId),
	70	eq(pullRequests.number, prNumber)
	71	)
	72	)
	73	.limit(1);
	74	return pr ?? null;
	75	}
	76
	77	/** Encode a SSE data line from a StreamingReviewToken. */
	78	function encodeToken(token: StreamingReviewToken): string {
	79	return `data: ${JSON.stringify(token)}\n\n`;
	80	}
	81
	82	// ---------------------------------------------------------------------------
	83	// SSE stream endpoint
	84	// ---------------------------------------------------------------------------
	85
	86	app.get("/:owner/:repo/pulls/:number/review/stream", softAuth, async (c) => {
	87	const { owner: ownerName, repo: repoName, number: numberStr } = c.req.param();
	88	const prNumber = parseInt(numberStr, 10);
	89	if (isNaN(prNumber) \|\| prNumber < 1) {
	90	return c.json({ error: "Invalid PR number" }, 400);
	91	}
	92
	93	// Resolve repo
	94	const resolved = await resolveRepo(ownerName, repoName);
	95	if (!resolved) {
	96	return c.json({ error: "Repository not found" }, 404);
	97	}
	98	const { repo } = resolved;
	99
	100	// Resolve PR
	101	const pr = await resolvePr(repo.id, prNumber);
	102	if (!pr) {
	103	return c.json({ error: "Pull request not found" }, 404);
	104	}
	105
	106	const prId = pr.id;
	107	const encoder = new TextEncoder();
	108
	109	const stream = new ReadableStream<Uint8Array>({
	110	async start(controller) {
	111	let closed = false;
	112
	113	const safeEnqueue = (chunk: string) => {
	114	if (closed) return;
	115	try {
	116	controller.enqueue(encoder.encode(chunk));
	117	} catch {
	118	closed = true;
	119	}
	120	};
	121
	122	const close = () => {
	123	if (closed) return;
	124	closed = true;
	125	try {
	126	controller.close();
	127	} catch {
	128	// already closed
	129	}
	130	};
	131
	132	// Flush headers on proxy buffering
	133	safeEnqueue(": open\n\n");
	134
	135	// If a review is already streaming, wait for it (max 60s, poll 2s)
	136	if (isReviewStreaming(prId)) {
	137	safeEnqueue(
	138	encodeToken({
	139	type: "token",
	140	content: "A review is already in progress. Waiting for it to complete...\n",
	141	})
	142	);
	143
	144	const MAX_WAIT_MS = 60_000;
	145	const POLL_MS = 2_000;
	146	const deadline = Date.now() + MAX_WAIT_MS;
	147
	148	while (isReviewStreaming(prId) && Date.now() < deadline && !closed) {
	149	await new Promise((r) => setTimeout(r, POLL_MS));
	150	}
	151
	152	if (isReviewStreaming(prId)) {
	153	safeEnqueue(
	154	encodeToken({
	155	type: "error",
	156	error: "Timed out waiting for in-progress review to complete.",
	157	})
	158	);
	159	close();
	160	return;
	161	}
	162
	163	// Review finished while we were waiting — inform the client
	164	safeEnqueue(
	165	encodeToken({
	166	type: "done",
	167	})
	168	);
	169	close();
	170	return;
	171	}
	172
	173	// Start the streaming review
	174	try {
	175	for await (const token of streamPrReview(
	176	prId,
	177	ownerName,
	178	repoName,
	179	pr.baseBranch,
	180	pr.headBranch
	181	)) {
	182	if (closed) break;
	183	safeEnqueue(encodeToken(token));
	184	if (token.type === "done" \|\| token.type === "error") {
	185	break;
	186	}
	187	}
	188	} catch (err) {
	189	const message = err instanceof Error ? err.message : String(err);
	190	safeEnqueue(encodeToken({ type: "error", error: message }));
	191	}
	192
	193	close();
	194	},
	195	});
	196
	197	// Handle client disconnect via AbortSignal — the ReadableStream cancel
	198	// is not directly wired, but closing the outer Response suffices for
	199	// Bun's HTTP layer to GC the stream controller.
	200	return new Response(stream, {
	201	status: 200,
	202	headers: {
	203	"Content-Type": "text/event-stream; charset=utf-8",
	204	"Cache-Control": "no-cache, no-transform",
	205	"Connection": "keep-alive",
	206	"X-Accel-Buffering": "no",
	207	},
	208	});
	209	});
	210
	211	// ---------------------------------------------------------------------------
	212	// Embeddable widget endpoint
	213	// ---------------------------------------------------------------------------
	214
	215	app.get("/:owner/:repo/pulls/:number/review/stream-ui", softAuth, async (c) => {
	216	const { owner: ownerName, repo: repoName, number: numberStr } = c.req.param();
	217	const prNumber = parseInt(numberStr, 10);
	218	if (isNaN(prNumber) \|\| prNumber < 1) {
	219	return c.html("<p>Invalid PR number</p>", 400);
	220	}
	221
	222	const streamUrl = `/${ownerName}/${repoName}/pulls/${prNumber}/review/stream`;
	223
	224	const html = `<!DOCTYPE html>
	225	<html lang="en">
	226	<head>
	227	<meta charset="utf-8">
	228	<meta name="viewport" content="width=device-width, initial-scale=1">
	229	<title>AI Stream Review — ${ownerName}/${repoName} #${prNumber}</title>
	230	<style>
	231	* { box-sizing: border-box; margin: 0; padding: 0; }
	232	body {
	233	font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
	234	background: #0d1117;
	235	color: #e6edf3;
	236	padding: 16px;
	237	}
	238	#stream-review {
	239	border: 1px solid #30363d;
	240	border-radius: 8px;
	241	background: #161b22;
	242	padding: 16px;
	243	position: relative;
	244	}
	245	.sr-header {
	246	display: flex;
	247	align-items: center;
	248	gap: 10px;
	249	margin-bottom: 14px;
	250	font-size: 14px;
	251	font-weight: 600;
	252	color: #8b949e;
	253	text-transform: uppercase;
	254	letter-spacing: 0.04em;
	255	}
	256	.sr-spinner {
	257	display: inline-block;
	258	width: 14px; height: 14px;
	259	border: 2px solid #30363d;
	260	border-top-color: #58a6ff;
	261	border-radius: 50%;
	262	animation: spin 0.8s linear infinite;
	263	}
	264	@keyframes spin { to { transform: rotate(360deg); } }
	265	.sr-section {
	266	margin-bottom: 14px;
	267	}
	268	.sr-section-label {
	269	font-size: 11px;
	270	font-weight: 600;
	271	text-transform: uppercase;
	272	letter-spacing: 0.06em;
	273	color: #58a6ff;
	274	margin-bottom: 6px;
	275	padding: 2px 6px;
	276	border-left: 2px solid #58a6ff;
	277	}
	278	.sr-section-label.finding { color: #f0883e; border-left-color: #f0883e; }
	279	.sr-section-label.verdict { color: #3fb950; border-left-color: #3fb950; }
	280	pre.sr-content {
	281	font-family: "SFMono-Regular", Consolas, monospace;
	282	font-size: 13px;
	283	line-height: 1.6;
	284	white-space: pre-wrap;
	285	word-break: break-word;
	286	color: #e6edf3;
	287	background: none;
	288	border: none;
	289	padding: 0;
	290	}
	291	.sr-done-msg {
	292	font-size: 12px;
	293	color: #3fb950;
	294	margin-top: 12px;
	295	padding: 6px 10px;
	296	border: 1px solid #3fb950;
	297	border-radius: 4px;
	298	display: none;
	299	}
	300	.sr-error-msg {
	301	font-size: 12px;
	302	color: #f85149;
	303	margin-top: 12px;
	304	padding: 6px 10px;
	305	border: 1px solid #f85149;
	306	border-radius: 4px;
	307	display: none;
	308	}
	309	</style>
	310	</head>
	311	<body>
	312	<div id="stream-review">
	313	<div class="sr-header">
	314	<span class="sr-spinner" id="sr-spinner"></span>
	315	<span>AI Stream Review</span>
	316	</div>
	317	<div id="sr-body">
	318	<div class="sr-section" id="sr-section-summary">
	319	<div class="sr-section-label summary">Summary</div>
	320	<pre class="sr-content" id="sr-pre-summary"></pre>
	321	</div>
	322	<div class="sr-section" id="sr-section-finding" style="display:none">
	323	<div class="sr-section-label finding">Findings</div>
	324	<pre class="sr-content" id="sr-pre-finding"></pre>
	325	</div>
	326	<div class="sr-section" id="sr-section-verdict" style="display:none">
	327	<div class="sr-section-label verdict">Verdict</div>
	328	<pre class="sr-content" id="sr-pre-verdict"></pre>
	329	</div>
	330	</div>
	331	<div class="sr-done-msg" id="sr-done">Review complete. Comment saved to PR.</div>
	332	<div class="sr-error-msg" id="sr-error"></div>
	333	</div>
	334
	335	<script>
	336	(function() {
	337	var streamUrl = ${JSON.stringify(streamUrl)};
	338	var currentSection = "summary";
	339	var sectionPres = {
	340	summary: document.getElementById("sr-pre-summary"),
	341	finding: document.getElementById("sr-pre-finding"),
	342	verdict: document.getElementById("sr-pre-verdict")
	343	};
	344	var sectionDivs = {
	345	summary: document.getElementById("sr-section-summary"),
	346	finding: document.getElementById("sr-section-finding"),
	347	verdict: document.getElementById("sr-section-verdict")
	348	};
	349	var spinner = document.getElementById("sr-spinner");
	350	var doneMsg = document.getElementById("sr-done");
	351	var errorMsg = document.getElementById("sr-error");
	352
	353	function showSection(name) {
	354	if (sectionDivs[name]) {
	355	sectionDivs[name].style.display = "";
	356	}
	357	currentSection = name;
	358	}
	359
	360	var es = new EventSource(streamUrl);
	361
	362	es.onmessage = function(e) {
	363	var token;
	364	try { token = JSON.parse(e.data); } catch { return; }
	365
	366	if (token.type === "token" && token.content) {
	367	var pre = sectionPres[currentSection];
	368	if (pre) pre.textContent += token.content;
	369	} else if (token.type === "section_start" && token.section) {
	370	showSection(token.section);
	371	} else if (token.type === "done") {
	372	es.close();
	373	if (spinner) spinner.style.display = "none";
	374	if (doneMsg) doneMsg.style.display = "";
	375	} else if (token.type === "error") {
	376	es.close();
	377	if (spinner) spinner.style.display = "none";
	378	if (errorMsg) {
	379	errorMsg.textContent = "Error: " + (token.error \|\| "Unknown error");
	380	errorMsg.style.display = "";
	381	}
	382	}
	383	};
	384
	385	es.onerror = function() {
	386	es.close();
	387	if (spinner) spinner.style.display = "none";
	388	if (errorMsg) {
	389	errorMsg.textContent = "Connection error. The review stream was interrupted.";
	390	errorMsg.style.display = "";
	391	}
	392	};
	393	})();
	394	</script>
	395	</body>
	396	</html>`;
	397
	398	return c.html(html);
	399	});
	400
	401	export { app as streamingReviewRoutes };
	402	export default app;