sse-helpers.ts
147 lines · browse/src/sse-helpers.ts at 3f79d81
1// SSE endpoint helper — shared cleanup contract for stream endpoints.
2//
3// Pre-helper, /activity/stream and /inspector/events implemented the same
4// pattern in parallel and both leaked subscribers when enqueue failed
5// without a corresponding abort signal (e.g. Chromium MV3 service-worker
6// suspend dropped the TCP without an abort edge). The subscriber closure
7// stayed in the Set, capturing the ReadableStreamDefaultController plus
8// any payloads queued behind it. Over a multi-day sidebar session this
9// compounded into multi-MB of retained controllers per dead connection.
10//
11// Centralizing the cleanup contract here means any future SSE endpoint
12// inherits the invariant — cleanup runs on abort, enqueue failure, AND
13// heartbeat failure, exactly once, regardless of which edge fires first.
15// sanitizeReplacer strips lone UTF-16 surrogates from string values before
16// they get escape-encoded — required at every SSE egress that ships
17// page-content-derived fields. See CLAUDE.md "Unicode sanitization at
18// server egress" and the canonical implementation in sanitize.ts.
19import { sanitizeReplacer } from './sanitize';
21/** Send an SSE event. Handles JSON encoding + lone-surrogate sanitization. */
22export type SseSender = (event: string, data: unknown) => void;
24export interface SseEndpointConfig<T> {
25 /**
26 * Optional. Runs once after the stream opens, before subscribing for live
27 * events. Use for initial event replay (activity gap detection, history
28 * burst) or a current-state snapshot (inspector). The `send` helper
29 * handles JSON encoding with sanitizeReplacer and SSE framing; pass
30 * any event name and any payload object.
31 */
32 initialReplay?: (send: SseSender) => void;
34 /**
35 * Subscribe to the live event source. Receives a `notify` callback;
36 * returns an unsubscribe function. The callback routes through the
37 * helper's safeEnqueue + cleanup-on-throw, so a dead consumer ends up
38 * removed from the subscriber set on the very next event (instead of
39 * waiting for an abort that may never fire).
40 */
41 subscribe: (notify: (entry: T) => void) => () => void;
43 /**
44 * SSE event name for live events. `data: <JSON.stringify(entry)>\n\n`
45 * is wrapped automatically. /activity/stream uses 'activity';
46 * /inspector/events uses 'inspector'.
47 */
48 liveEventName: string;
50 /** Heartbeat interval in ms. Default: 15000. */
51 heartbeatMs?: number;
52}
54/**
55 * Build a streaming Response that owns the cleanup contract:
56 * - safeEnqueue catches enqueue throws → cleanup
57 * - 15s heartbeat catches dead peers; failure → cleanup
58 * - req.signal abort → cleanup
59 * - cleanup is idempotent (clearInterval + unsubscribe + try close)
60 */
61export function createSseEndpoint<T>(
62 req: Request,
63 config: SseEndpointConfig<T>,
64): Response {
65 const heartbeatMs = config.heartbeatMs ?? 15000;
66 const encoder = new TextEncoder();
68 const stream = new ReadableStream({
69 start(controller) {
70 let cleanedUp = false;
71 let heartbeat: ReturnType<typeof setInterval> | null = null;
72 let unsubscribe: (() => void) | null = null;
74 const cleanup = (): void => {
75 if (cleanedUp) return;
76 cleanedUp = true;
77 if (heartbeat !== null) {
78 clearInterval(heartbeat);
79 heartbeat = null;
80 }
81 if (unsubscribe !== null) {
82 unsubscribe();
83 unsubscribe = null;
84 }
85 try {
86 controller.close();
87 } catch {
88 // Expected: stream already closed by the consumer.
89 }
90 };
92 const send: SseSender = (event, data) => {
93 if (cleanedUp) return;
94 try {
95 controller.enqueue(
96 encoder.encode(
97 `event: ${event}\ndata: ${JSON.stringify(data, sanitizeReplacer)}\n\n`,
98 ),
99 );
100 } catch {
101 // Consumer disconnected mid-write. Tear down so this subscriber
102 // doesn't sit in the set forever.
103 cleanup();
104 }
105 };
107 // Initial replay (caller-provided).
108 if (config.initialReplay) {
109 try {
110 config.initialReplay(send);
111 } catch {
112 cleanup();
113 return;
114 }
115 if (cleanedUp) return;
116 }
118 // Subscribe for live events.
119 unsubscribe = config.subscribe((entry) => {
120 send(config.liveEventName, entry);
121 });
123 // Heartbeat keeps NAT boxes and proxies from dropping idle SSE,
124 // and serves as a liveness probe: an enqueue failure here is the
125 // cheapest way to learn the consumer is gone without waiting for
126 // an abort signal that may never arrive.
127 heartbeat = setInterval(() => {
128 if (cleanedUp) return;
129 try {
130 controller.enqueue(encoder.encode(`: heartbeat\n\n`));
131 } catch {
132 cleanup();
133 }
134 }, heartbeatMs);
136 req.signal.addEventListener('abort', cleanup);
137 },
138 });
140 return new Response(stream, {
141 headers: {
142 'Content-Type': 'text/event-stream',
143 'Cache-Control': 'no-cache',
144 'Connection': 'keep-alive',
145 },
146 });
147}