Commit6522084unknown_key
feat: add SSE streaming for real-time gate updates and notifications
feat: add SSE streaming for real-time gate updates and notifications New SSE infrastructure (src/lib/sse.ts) with channel-based pub/sub. Gate checks now broadcast start/completion events so the browser can show live progress without polling. Events route at /api/events/stream accepts channel subscriptions (gate:repoId, pr:prId, notification). Includes auto-cleanup of stale connections every 5 minutes. https://claude.ai/code/session_01M6EgoTaZ6HAZiVu77xp11g
4 files changed+224−1652208418702957d36376aae5e38701f62b40beb
4 changed files+224−1
Modifiedsrc/app.tsx+4−0View fileUnifiedSplit
@@ -75,6 +75,7 @@ import signingKeysRoutes from "./routes/signing-keys";
7575import followsRoutes from "./routes/follows";
7676import rulesetsRoutes from "./routes/rulesets";
7777import commitStatusesRoutes from "./routes/commit-statuses";
78import eventsRoutes from "./routes/events";
7879import webRoutes from "./routes/web";
7980
8081const app = new Hono();
@@ -103,6 +104,9 @@ app.route("/", healthRoutes);
103104// Inbound API hooks (GateTest callback + backup PAT-authed /api/v1/gate-runs)
104105app.route("/", hookRoutes);
105106
107// SSE event streams (real-time gate updates, notifications)
108app.route("/", eventsRoutes);
109
106110// REST API
107111app.route("/", apiRoutes);
108112
Modifiedsrc/lib/gate.ts+33−1View fileUnifiedSplit
@@ -23,6 +23,7 @@ import { repairSecrets, repairSecurityIssues } from "./auto-repair";
2323import { readFile } from "fs/promises";
2424import { join } from "path";
2525import { updateGateMetrics, extractPatterns } from "./flywheel";
26import { broadcast } from "./sse";
2627
2728export interface GateCheckResult {
2829 name: string;
@@ -317,6 +318,17 @@ export async function runAllGateChecks(
317318 const runAiReview = settings?.aiReviewEnabled !== false;
318319 const enableRepair = opts.enableAutoRepair !== false && settings?.autoFixEnabled !== false;
319320
321 // SSE: broadcast that gate checks are starting
322 const sseChannel = repoRow ? `gate:${repoRow.id}` : null;
323 if (sseChannel) {
324 broadcast(sseChannel, "gate:started", {
325 repoId: repoRow!.id,
326 prId: opts.pullRequestId,
327 sha: headSha,
328 gates: ["GateTest", "Secret scan", "Security scan", "Merge check", "AI Review"],
329 });
330 }
331
320332 const [gateTestResult, mergeResult, scanResults] = await Promise.all([
321333 runGateTest
322334 ? runGateTestScan(owner, repo, `refs/heads/${headBranch}`, headSha)
@@ -418,8 +430,28 @@ export async function runAllGateChecks(
418430 }
419431 }
420432
421 return {
433 const result = {
422434 allPassed: checks.every((c) => c.passed || c.skipped),
423435 checks,
424436 };
437
438 // SSE: broadcast gate completion with full results
439 if (sseChannel) {
440 broadcast(sseChannel, "gate:completed", {
441 repoId: repoRow!.id,
442 prId: opts.pullRequestId,
443 sha: headSha,
444 allPassed: result.allPassed,
445 checks: result.checks.map((c) => ({
446 name: c.name,
447 passed: c.passed,
448 skipped: c.skipped,
449 repaired: c.repaired,
450 details: c.details,
451 })),
452 durationMs: Date.now() - started,
453 });
454 }
455
456 return result;
425457}
Addedsrc/lib/sse.ts+148−0View fileUnifiedSplit
@@ -0,0 +1,148 @@
1/**
2 * Server-Sent Events (SSE) infrastructure.
3 *
4 * Manages live connections so gate runs, notifications, and CI updates
5 * stream to the browser in real time instead of requiring page refreshes.
6 */
7
8type SSEChannel = "gate" | "notification" | "pr" | "ci";
9
10interface SSEClient {
11 id: string;
12 controller: ReadableStreamDefaultController;
13 userId?: string;
14 channels: Set<string>; // e.g. "gate:repoId", "pr:prId", "notification:userId"
15 connectedAt: number;
16}
17
18const clients = new Map<string, SSEClient>();
19
20let clientIdCounter = 0;
21
22function nextClientId(): string {
23 return `sse_${++clientIdCounter}_${Date.now().toString(36)}`;
24}
25
26/**
27 * Create an SSE Response for a Hono handler.
28 * The caller subscribes to channels, and this function returns a streaming Response.
29 */
30export function createSSEStream(
31 channels: string[],
32 userId?: string
33): Response {
34 const clientId = nextClientId();
35
36 const stream = new ReadableStream({
37 start(controller) {
38 const client: SSEClient = {
39 id: clientId,
40 controller,
41 userId,
42 channels: new Set(channels),
43 connectedAt: Date.now(),
44 };
45 clients.set(clientId, client);
46
47 // Send initial connection event
48 const data = `event: connected\ndata: ${JSON.stringify({ clientId, channels })}\n\n`;
49 controller.enqueue(new TextEncoder().encode(data));
50 },
51 cancel() {
52 clients.delete(clientId);
53 },
54 });
55
56 return new Response(stream, {
57 headers: {
58 "Content-Type": "text/event-stream",
59 "Cache-Control": "no-cache",
60 Connection: "keep-alive",
61 "X-Accel-Buffering": "no",
62 },
63 });
64}
65
66/**
67 * Broadcast an event to all clients subscribed to a channel.
68 */
69export function broadcast(
70 channel: string,
71 event: string,
72 data: unknown
73): number {
74 const encoded = new TextEncoder();
75 const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
76 const bytes = encoded.encode(payload);
77 let sent = 0;
78
79 for (const [id, client] of clients) {
80 if (client.channels.has(channel)) {
81 try {
82 client.controller.enqueue(bytes);
83 sent++;
84 } catch {
85 clients.delete(id);
86 }
87 }
88 }
89
90 return sent;
91}
92
93/**
94 * Send an event to a specific user (by userId) across all their connections.
95 */
96export function sendToUser(
97 userId: string,
98 event: string,
99 data: unknown
100): number {
101 const encoded = new TextEncoder();
102 const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
103 const bytes = encoded.encode(payload);
104 let sent = 0;
105
106 for (const [id, client] of clients) {
107 if (client.userId === userId) {
108 try {
109 client.controller.enqueue(bytes);
110 sent++;
111 } catch {
112 clients.delete(id);
113 }
114 }
115 }
116
117 return sent;
118}
119
120/**
121 * Get active connection count (for monitoring).
122 */
123export function getActiveConnections(): number {
124 return clients.size;
125}
126
127/**
128 * Clean up stale connections (call periodically).
129 */
130export function cleanupStaleConnections(maxAgeMs = 30 * 60 * 1000): number {
131 const cutoff = Date.now() - maxAgeMs;
132 let removed = 0;
133
134 for (const [id, client] of clients) {
135 if (client.connectedAt < cutoff) {
136 try {
137 client.controller.close();
138 } catch {}
139 clients.delete(id);
140 removed++;
141 }
142 }
143
144 return removed;
145}
146
147// Periodic cleanup every 5 minutes
148setInterval(() => cleanupStaleConnections(), 5 * 60 * 1000);
Addedsrc/routes/events.ts+39−0View fileUnifiedSplit
@@ -0,0 +1,39 @@
1/**
2 * SSE event stream routes — real-time updates for gate runs, PRs, and notifications.
3 *
4 * GET /api/events/stream?channels=gate:repoId,pr:prId,notification
5 */
6
7import { Hono } from "hono";
8import { softAuth } from "../middleware/auth";
9import type { AuthEnv } from "../middleware/auth";
10import { createSSEStream, getActiveConnections } from "../lib/sse";
11
12const events = new Hono<AuthEnv>();
13
14events.get("/api/events/stream", softAuth, (c) => {
15 const user = c.get("user");
16 const channelParam = c.req.query("channels") || "";
17 const requestedChannels = channelParam
18 .split(",")
19 .map((ch) => ch.trim())
20 .filter(Boolean);
21
22 if (requestedChannels.length === 0) {
23 return c.json({ error: "No channels specified" }, 400);
24 }
25
26 // Auto-add user-specific notification channel if authenticated
27 const channels = [...requestedChannels];
28 if (user && !channels.some((ch) => ch.startsWith("notification"))) {
29 channels.push(`notification:${user.id}`);
30 }
31
32 return createSSEStream(channels, user?.id);
33});
34
35events.get("/api/events/health", (c) => {
36 return c.json({ connections: getActiveConnections() });
37});
38
39export default events;
040