From 8dc41e00ba1205c7531e3184a290bf02cde50818 Mon Sep 17 00:00:00 2001 From: Charis Daniels Date: Fri, 28 Aug 2026 19:16:37 +0000 Subject: [PATCH] fix(server): add 30s heartbeat to SSE stream, closes #72 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Open /api/events connections now get a `: heartbeat` comment line every 30 seconds so proxies and load balancers don't close idle streams, with per-client intervals cleaned up on disconnect. Also fixes pre-existing breakages that failed CI quality gates: - drop duplicate SseEmitter import in admin.ts (tsc error) - map @server/* path alias in jest config so app-level tests resolve - restore the 10kb JSON body limit intended by the middleware refactor so the body-size-limit test passes Generated with Codebuff 🤖 Co-Authored-By: Codebuff --- server/package.json | 3 + server/src/index.ts | 1 + server/src/routes/admin.ts | 1 - server/src/routes/events.ts | 3 +- server/src/services/sseEmitter.test.ts | 109 +++++++++++++++++++++++++ server/src/services/sseEmitter.ts | 30 +++++++ 6 files changed, 145 insertions(+), 2 deletions(-) create mode 100644 server/src/services/sseEmitter.test.ts diff --git a/server/package.json b/server/package.json index ebdc0f8..caf4ba1 100644 --- a/server/package.json +++ b/server/package.json @@ -62,6 +62,9 @@ "roots": [ "/src" ], + "moduleNameMapper": { + "^@server/(.*)$": "/src/$1" + }, "testMatch": [ "**/*.test.ts" ], diff --git a/server/src/index.ts b/server/src/index.ts index 43e91d0..149aeca 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -129,6 +129,7 @@ app.use( // signature check needs the original bytes, not req.body. See routes/webhooks. app.use( express.json({ + limit: "10kb", verify: (req, _res, buf) => { (req as express.Request & { rawBody?: Buffer }).rawBody = buf; }, diff --git a/server/src/routes/admin.ts b/server/src/routes/admin.ts index b913ec8..4a8a147 100644 --- a/server/src/routes/admin.ts +++ b/server/src/routes/admin.ts @@ -21,7 +21,6 @@ import { QueueService } from "../jobs"; import { SseEmitter } from "../services/sseEmitter"; import pool from "../db"; import { resolveDispute } from "../services/stellar"; -import { SseEmitter } from "../services/sseEmitter"; import { resolveDisputeSchema, paginationSchema, diff --git a/server/src/routes/events.ts b/server/src/routes/events.ts index 0b3977a..f6e5675 100644 --- a/server/src/routes/events.ts +++ b/server/src/routes/events.ts @@ -18,7 +18,8 @@ const router = Router(); * - connected — sent immediately on connect as a handshake * * The connection stays open indefinitely. Browsers automatically reconnect - * using the built-in EventSource retry mechanism. + * using the built-in EventSource retry mechanism. A `: heartbeat` comment + * line is sent every 30 seconds to keep the stream alive through proxies. * * Example (browser): * const es = new EventSource("/api/v1/events", { headers: { Authorization: "Bearer ..." } }); diff --git a/server/src/services/sseEmitter.test.ts b/server/src/services/sseEmitter.test.ts new file mode 100644 index 0000000..c421f5c --- /dev/null +++ b/server/src/services/sseEmitter.test.ts @@ -0,0 +1,109 @@ +import type { Response } from "express"; +import { SseEmitter } from "./sseEmitter"; + +describe("SseEmitter", () => { + let write: jest.Mock; + let flush: jest.Mock; + let on: jest.Mock; + let mockRes: Partial & { write: jest.Mock; on: jest.Mock }; + let activeMocks: Array<{ on: jest.Mock }>; + + beforeEach(() => { + jest.useFakeTimers(); + activeMocks = []; + write = jest.fn(); + flush = jest.fn(); + on = jest.fn(); + mockRes = { + setHeader: jest.fn(), + flushHeaders: jest.fn(), + write, + flush, + on, + } as unknown as typeof mockRes; + activeMocks.push(mockRes); + }); + + afterEach(() => { + // Disconnect every client registered during the test so the module-level + // registry does not leak connections into the next test. + for (const mock of activeMocks) { + const call = mock.on.mock.calls.find(([name]) => name === "close"); + if (call) (call[1] as () => void)(); + } + expect(SseEmitter.connectionCount()).toBe(0); + jest.useRealTimers(); + }); + + /** Invokes the handler registered for the given event name on the mock res. */ + function trigger(event: string): void { + const call = on.mock.calls.find(([name]) => name === event); + expect(call).toBeDefined(); + (call[1] as () => void)(); + } + + it("opens a stream with SSE headers and an initial connected event", () => { + SseEmitter.addClient("user-1", mockRes as Response); + + expect(mockRes.setHeader).toHaveBeenCalledWith("Content-Type", "text/event-stream"); + expect(mockRes.setHeader).toHaveBeenCalledWith("Cache-Control", "no-cache, no-transform"); + expect(mockRes.setHeader).toHaveBeenCalledWith("Connection", "keep-alive"); + expect(mockRes.flushHeaders).toHaveBeenCalledTimes(1); + expect(write).toHaveBeenCalledWith('event: connected\n'); + expect(write).toHaveBeenCalledWith('data: {"type":"connected","userId":"user-1"}\n\n'); + expect(SseEmitter.connectionCount()).toBe(1); + }); + + it("writes a heartbeat comment line every 30 seconds", () => { + SseEmitter.addClient("user-1", mockRes as Response); + write.mockClear(); + + jest.advanceTimersByTime(30_000); + expect(write).toHaveBeenCalledWith(": heartbeat\n\n"); + expect(flush).toHaveBeenCalled(); + + write.mockClear(); + jest.advanceTimersByTime(30_000); + expect(write).toHaveBeenCalledWith(": heartbeat\n\n"); + }); + + it("stops heartbeating and unregisters the client on close", () => { + SseEmitter.addClient("user-1", mockRes as Response); + write.mockClear(); + + trigger("close"); + + expect(SseEmitter.connectionCount()).toBe(0); + + jest.advanceTimersByTime(120_000); + expect(write).not.toHaveBeenCalled(); + }); + + it("emits events only to the targeted user and drops events for offline users", () => { + SseEmitter.addClient("user-1", mockRes as Response); + + const otherWrite = jest.fn(); + const otherRes = { + setHeader: jest.fn(), + flushHeaders: jest.fn(), + write: otherWrite, + on: jest.fn(), + } as unknown as Response; + activeMocks.push(otherRes as unknown as { on: jest.Mock }); + SseEmitter.addClient("user-2", otherRes); + + write.mockClear(); + otherWrite.mockClear(); + + SseEmitter.emit(["user-1"], { type: "trade_completed", tradeId: "t1", status: "Completed" }); + + expect(write).toHaveBeenCalledWith("event: trade_completed\n"); + expect(write).toHaveBeenCalledWith( + 'data: {"type":"trade_completed","tradeId":"t1","status":"Completed"}\n\n' + ); + expect(otherWrite).not.toHaveBeenCalled(); + + // Offline user: event is dropped, no throw + expect(() => SseEmitter.emit(["ghost"], { type: "trade_completed" })).not.toThrow(); + }); +}); diff --git a/server/src/services/sseEmitter.ts b/server/src/services/sseEmitter.ts index b369256..de10c5a 100644 --- a/server/src/services/sseEmitter.ts +++ b/server/src/services/sseEmitter.ts @@ -25,8 +25,22 @@ import { Response } from "express"; * * // Push a broadcast admin alert: * SseEmitter.emitAdmin({ type: "admin_alert", message: "..." }); + * + * Heartbeat + * --------- + * While a client is connected the server writes an SSE comment line + * (`: heartbeat`) every 30 seconds. Comment lines are ignored by EventSource + * but keep the TCP connection alive through proxies and load balancers that + * would otherwise close idle connections. */ +// --------------------------------------------------------------------------- +// Constants +// --------------------------------------------------------------------------- + +/** How often to write a keep-alive comment line to each connected client. */ +const HEARTBEAT_INTERVAL_MS = 30_000; + // --------------------------------------------------------------------------- // Types // --------------------------------------------------------------------------- @@ -67,8 +81,24 @@ function addClient(userId: string, res: Response): void { clients.push({ userId, res }); + // Heartbeat: a comment line every 30s prevents proxies from closing idle + // connections. Per-client interval so it stops when this connection closes. + const heartbeat = setInterval(() => { + try { + res.write(": heartbeat\n\n"); + if (typeof (res as Response & { flush?: () => void }).flush === "function") { + (res as Response & { flush: () => void }).flush(); + } + } catch { + // Client disconnected mid-write — the "close" handler will clean up + } + }, HEARTBEAT_INTERVAL_MS); + // Don't keep the process alive just for an open SSE stream + heartbeat.unref?.(); + // Clean up when the client disconnects res.on("close", () => { + clearInterval(heartbeat); const idx = clients.findIndex((c) => c.res === res); if (idx !== -1) clients.splice(idx, 1); });