fix(server): add 30s SSE heartbeat to keep /api/events alive

server/package.json

@@ -65,6 +65,9 @@
     "roots": [
       "<rootDir>/src"
     ],
+    "moduleNameMapper": {
+      "^@server/(.*)$": "<rootDir>/src/$1"
+    },
     "testMatch": [
       "**/*.test.ts"
     ],

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 ..." } });

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<Response> & { 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();
+  });
+});

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);
   });