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.
| 3c03977 | 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) => { | |
| 6cbc9cb | 55 | // softAuth POPULATES the viewer but never denies, and these handlers never |
| 56 | // read c.get("user") — so an anonymous caller could broadcast presence and | |
| 57 | // arbitrary edit patches into any live PR session it could name, injecting | |
| 58 | // fake cursors and edits into collaborators' views. Live collaboration is a | |
| 59 | // signed-in feature; require the session. | |
| 60 | if (!c.get("user")) return c.json({ error: "Authentication required" }, 401); | |
| 3c03977 | 61 | const prId = c.req.param("prId"); |
| 62 | if (!prId || !ID_RE.test(prId)) { | |
| 63 | return c.json({ error: "Invalid pr id" }, 400); | |
| 64 | } | |
| 65 | ||
| 66 | const user = c.get("user") ?? null; | |
| 67 | const topic = prLiveTopic(prId); | |
| 68 | ||
| 69 | // Auto-join — only humans for v1; agents will get an explicit | |
| 70 | // POST /join endpoint when the harness wants them on the stream. | |
| 71 | let sessionId: string | null = null; | |
| 72 | let sessionColor: string | null = null; | |
| 73 | if (user) { | |
| 74 | const joined = await joinSession({ prId, userId: user.id }); | |
| 75 | if (joined) { | |
| 76 | sessionId = joined.sessionId; | |
| 77 | sessionColor = joined.color; | |
| 78 | } | |
| 79 | } | |
| 80 | ||
| 81 | // Snapshot of current presence — sent as the first event so the | |
| 82 | // client can render avatars without an extra round-trip. | |
| 83 | const presence = await listLive(prId); | |
| 84 | ||
| 85 | const encoder = new TextEncoder(); | |
| 86 | ||
| 87 | const stream = new ReadableStream<Uint8Array>({ | |
| 88 | start(controller) { | |
| 89 | let closed = false; | |
| 90 | const safeEnqueue = (chunk: string) => { | |
| 91 | if (closed) return; | |
| 92 | try { | |
| 93 | controller.enqueue(encoder.encode(chunk)); | |
| 94 | } catch { | |
| 95 | closed = true; | |
| 96 | } | |
| 97 | }; | |
| 98 | ||
| 99 | const sendEvent = (event: SSEEvent) => { | |
| 100 | let payload = ""; | |
| 101 | if (event.id !== undefined) payload += `id: ${event.id}\n`; | |
| 102 | if (event.event !== undefined) payload += `event: ${event.event}\n`; | |
| 103 | const data = | |
| 104 | typeof event.data === "string" | |
| 105 | ? event.data | |
| 106 | : JSON.stringify(event.data); | |
| 107 | for (const line of data.split("\n")) { | |
| 108 | payload += `data: ${line}\n`; | |
| 109 | } | |
| 110 | payload += "\n"; | |
| 111 | safeEnqueue(payload); | |
| 112 | }; | |
| 113 | ||
| 114 | // Flush headers on proxies that buffer. | |
| 115 | safeEnqueue(": open\n\n"); | |
| 116 | ||
| 117 | // Initial "hello" with the snapshot + the joined session id (so | |
| 118 | // the client knows which row is theirs and can suppress echo | |
| 119 | // events from itself). | |
| 120 | sendEvent({ | |
| 121 | event: "hello", | |
| 122 | data: { | |
| 123 | sessionId, | |
| 124 | color: sessionColor, | |
| 125 | presence, | |
| 126 | }, | |
| 127 | }); | |
| 128 | ||
| 129 | const unsubscribe = subscribe(topic, sendEvent); | |
| 130 | ||
| 131 | const ping = setInterval(() => { | |
| 132 | safeEnqueue(": ping\n\n"); | |
| 133 | }, SSE_PING_MS); | |
| 134 | ||
| 135 | const cleanup = async () => { | |
| 136 | if (closed) return; | |
| 137 | closed = true; | |
| 138 | clearInterval(ping); | |
| 139 | unsubscribe(); | |
| 140 | if (sessionId) { | |
| 141 | // Fire-and-forget. The autopilot sweep is the durable | |
| 142 | // fallback if this never runs (process crash). | |
| 143 | try { | |
| 144 | await leaveSession(sessionId, prId); | |
| 145 | } catch { | |
| 146 | /* best-effort */ | |
| 147 | } | |
| 148 | } | |
| 149 | try { | |
| 150 | controller.close(); | |
| 151 | } catch { | |
| 152 | /* already closed */ | |
| 153 | } | |
| 154 | }; | |
| 155 | ||
| 156 | const signal = c.req.raw.signal; | |
| 157 | if (signal) { | |
| 158 | if (signal.aborted) { | |
| 159 | void cleanup(); | |
| 160 | } else { | |
| 161 | signal.addEventListener("abort", () => void cleanup(), { once: true }); | |
| 162 | } | |
| 163 | } | |
| 164 | }, | |
| 165 | }); | |
| 166 | ||
| 167 | return new Response(stream, { | |
| 168 | status: 200, | |
| 169 | headers: { | |
| 170 | "Content-Type": "text/event-stream; charset=utf-8", | |
| 171 | "Cache-Control": "no-cache, no-transform", | |
| 172 | Connection: "keep-alive", | |
| 173 | "X-Accel-Buffering": "no", | |
| 174 | }, | |
| 175 | }); | |
| 176 | }); | |
| 177 | ||
| 178 | // ---------- Control plane ---------- | |
| 179 | ||
| 180 | async function parseJsonBody(c: import("hono").Context): Promise<any> { | |
| 181 | try { | |
| 182 | return await c.req.json(); | |
| 183 | } catch { | |
| 184 | return null; | |
| 185 | } | |
| 186 | } | |
| 187 | ||
| 188 | app.post("/api/v2/pulls/:prId/live/cursor", softAuth, async (c) => { | |
| 6cbc9cb | 189 | // softAuth POPULATES the viewer but never denies, and these handlers never |
| 190 | // read c.get("user") — so an anonymous caller could broadcast presence and | |
| 191 | // arbitrary edit patches into any live PR session it could name, injecting | |
| 192 | // fake cursors and edits into collaborators' views. Live collaboration is a | |
| 193 | // signed-in feature; require the session. | |
| 194 | if (!c.get("user")) return c.json({ error: "Authentication required" }, 401); | |
| 3c03977 | 195 | const prId = c.req.param("prId"); |
| 196 | if (!prId || !ID_RE.test(prId)) return c.json({ error: "Invalid pr id" }, 400); | |
| 197 | const body = await parseJsonBody(c); | |
| 198 | const sessionId = String(body?.sessionId || ""); | |
| 199 | const position = body?.position as CursorPosition | undefined; | |
| 200 | if (!sessionId || !position || typeof position.field !== "string") { | |
| 201 | return c.json({ error: "Invalid body" }, 400); | |
| 202 | } | |
| 203 | await updateCursor(sessionId, prId, position); | |
| 204 | return c.json({ ok: true }); | |
| 205 | }); | |
| 206 | ||
| 207 | app.post("/api/v2/pulls/:prId/live/edit", softAuth, async (c) => { | |
| 6cbc9cb | 208 | // softAuth POPULATES the viewer but never denies, and these handlers never |
| 209 | // read c.get("user") — so an anonymous caller could broadcast presence and | |
| 210 | // arbitrary edit patches into any live PR session it could name, injecting | |
| 211 | // fake cursors and edits into collaborators' views. Live collaboration is a | |
| 212 | // signed-in feature; require the session. | |
| 213 | if (!c.get("user")) return c.json({ error: "Authentication required" }, 401); | |
| 3c03977 | 214 | const prId = c.req.param("prId"); |
| 215 | if (!prId || !ID_RE.test(prId)) return c.json({ error: "Invalid pr id" }, 400); | |
| 216 | const body = await parseJsonBody(c); | |
| 217 | const sessionId = String(body?.sessionId || ""); | |
| 218 | const patch = body?.patch as EditPatch | undefined; | |
| 219 | if (!sessionId || !patch || typeof patch.field !== "string") { | |
| 220 | return c.json({ error: "Invalid body" }, 400); | |
| 221 | } | |
| 222 | await broadcastEdit(sessionId, prId, patch); | |
| 223 | return c.json({ ok: true }); | |
| 224 | }); | |
| 225 | ||
| 226 | app.post("/api/v2/pulls/:prId/live/heartbeat", softAuth, async (c) => { | |
| 6cbc9cb | 227 | // softAuth POPULATES the viewer but never denies, and these handlers never |
| 228 | // read c.get("user") — so an anonymous caller could broadcast presence and | |
| 229 | // arbitrary edit patches into any live PR session it could name, injecting | |
| 230 | // fake cursors and edits into collaborators' views. Live collaboration is a | |
| 231 | // signed-in feature; require the session. | |
| 232 | if (!c.get("user")) return c.json({ error: "Authentication required" }, 401); | |
| 3c03977 | 233 | const prId = c.req.param("prId"); |
| 234 | if (!prId || !ID_RE.test(prId)) return c.json({ error: "Invalid pr id" }, 400); | |
| 235 | const body = await parseJsonBody(c); | |
| 236 | const sessionId = String(body?.sessionId || ""); | |
| 237 | if (!sessionId) return c.json({ error: "Invalid body" }, 400); | |
| 238 | await heartbeat(sessionId, prId); | |
| 239 | return c.json({ ok: true }); | |
| 240 | }); | |
| 241 | ||
| 242 | app.post("/api/v2/pulls/:prId/live/leave", softAuth, async (c) => { | |
| 6cbc9cb | 243 | // softAuth POPULATES the viewer but never denies, and these handlers never |
| 244 | // read c.get("user") — so an anonymous caller could broadcast presence and | |
| 245 | // arbitrary edit patches into any live PR session it could name, injecting | |
| 246 | // fake cursors and edits into collaborators' views. Live collaboration is a | |
| 247 | // signed-in feature; require the session. | |
| 248 | if (!c.get("user")) return c.json({ error: "Authentication required" }, 401); | |
| 3c03977 | 249 | const prId = c.req.param("prId"); |
| 250 | if (!prId || !ID_RE.test(prId)) return c.json({ error: "Invalid pr id" }, 400); | |
| 251 | // Accept JSON body OR a sendBeacon Blob — sendBeacon sets | |
| 252 | // content-type to text/plain;charset=UTF-8 by default. Try JSON | |
| 253 | // first, fall back to reading raw text and parsing it. | |
| 254 | let body = await parseJsonBody(c); | |
| 255 | if (!body) { | |
| 256 | try { | |
| 257 | const raw = await c.req.text(); | |
| 258 | body = raw ? JSON.parse(raw) : null; | |
| 259 | } catch { | |
| 260 | body = null; | |
| 261 | } | |
| 262 | } | |
| 263 | const sessionId = String(body?.sessionId || ""); | |
| 264 | if (!sessionId) return c.json({ error: "Invalid body" }, 400); | |
| 265 | await leaveSession(sessionId, prId); | |
| 266 | return c.json({ ok: true }); | |
| 267 | }); | |
| 268 | ||
| 269 | export default app; |