Keep notification SSE stream alive behind proxies (#1516)

* fix: keep notification SSE stream alive

* Fix PR comments

* fix: cover notification stream error cleanup

---------

Co-authored-by: Konstantin Zolin <zolin_ka@vk.com>
Co-authored-by: Bohdan Triapitsyn <artmore@protonmail.com>

packages/web/server/lib/notifications/DOCUMENTATION.md

@@ -26,6 +26,7 @@ This module provides notification message preparation utilities for the web serv
   - `DELETE /api/push/subscribe`
   - `POST /api/push/visibility`
   - `GET /api/push/visibility`
+  - `GET /api/notifications/stream`
   - `GET /api/session-activity`
   - `GET /api/sessions/snapshot`
   - `GET /api/sessions/status`
@@ -86,6 +87,7 @@ This module provides notification message preparation utilities for the web serv
 
 ### Default values
 - `DEFAULT_NOTIFICATION_MESSAGE_MAX_LENGTH`: 250 (default max length for notification text).
+- `NOTIFICATION_SSE_HEARTBEAT_INTERVAL_MS`: 20000 (notification SSE comment heartbeat interval).
 
 ## Settings object format
 

packages/web/server/lib/notifications/routes.js

@@ -22,6 +22,8 @@ const parsePushUnsubscribeBody = (body) => {
   return { endpoint: endpoint.trim() };
 };
 
+export const NOTIFICATION_SSE_HEARTBEAT_INTERVAL_MS = 20_000;
+
 export const registerNotificationRoutes = (app, dependencies) => {
   const {
     uiAuthController,
@@ -180,17 +182,50 @@ export const registerNotificationRoutes = (app, dependencies) => {
     const clients = getUiNotificationClients();
     clients.add(res);
 
+    let closed = false;
+    let heartbeatTimer = null;
+
+    const cleanup = () => {
+      if (closed) {
+        return;
+      }
+      closed = true;
+      if (heartbeatTimer) {
+        clearInterval(heartbeatTimer);
+        heartbeatTimer = null;
+      }
+      clients.delete(res);
+    };
+
+    req.on('close', cleanup);
+    res.on('error', cleanup);
+
+    const flushSse = () => {
+      res.flush?.();
+    };
+
+    heartbeatTimer = setInterval(() => {
+      if (closed || res.writableEnded || res.destroyed) {
+        cleanup();
+        return;
+      }
+      try {
+        res.write(':heartbeat\n\n');
+        flushSse();
+      } catch {
+        cleanup();
+      }
+    }, NOTIFICATION_SSE_HEARTBEAT_INTERVAL_MS);
+
     try {
       writeSseEvent(res, {
         type: 'openchamber:notification-stream-ready',
         properties: { uiToken },
       });
+      flushSse();
     } catch {
+      cleanup();
     }
-
-    req.on('close', () => {
-      clients.delete(res);
-    });
   });
 
   app.get('/api/session-activity', (_req, res) => {

packages/web/server/sse-routes.test.js

@@ -1,6 +1,6 @@
-import { describe, expect, it } from 'bun:test';
+import { describe, expect, it, vi } from 'bun:test';
 
-import { registerNotificationRoutes } from './lib/notifications/routes.js';
+import { NOTIFICATION_SSE_HEARTBEAT_INTERVAL_MS, registerNotificationRoutes } from './lib/notifications/routes.js';
 import { registerScheduledTaskRoutes } from './lib/scheduled-tasks/routes.js';
 
 const createRouteRegistry = () => {
@@ -47,11 +47,23 @@ const createMockRequest = () => {
 
 const createMockResponse = () => {
   const headers = new Map();
+  const listeners = new Map();
   let statusCode = 200;
   let body = '';
   let flushed = false;
+  let bodyFlushCount = 0;
 
   return {
+    on(event, handler) {
+      listeners.set(event, handler);
+      return this;
+    },
+    emit(event) {
+      const handler = listeners.get(event);
+      if (typeof handler === 'function') {
+        handler();
+      }
+    },
     setHeader(name, value) {
       headers.set(name.toLowerCase(), value);
     },
@@ -61,6 +73,9 @@ const createMockResponse = () => {
     flushHeaders() {
       flushed = true;
     },
+    flush() {
+      bodyFlushCount += 1;
+    },
     write(chunk) {
       body += String(chunk);
       return true;
@@ -82,42 +97,61 @@ const createMockResponse = () => {
     get flushed() {
       return flushed;
     },
+    get bodyFlushCount() {
+      return bodyFlushCount;
+    },
   };
 };
 
 describe('local SSE routes', () => {
   it('serves notification SSE with nginx-safe headers', async () => {
+    vi.useFakeTimers();
     const { app, getRoute } = createRouteRegistry();
     const clients = new Set();
 
-    registerNotificationRoutes(app, {
+    try {
-      uiAuthController: {
+      registerNotificationRoutes(app, {
-        ensureSessionToken: async () => 'ui-token',
+        uiAuthController: {
-      },
+          ensureSessionToken: async () => 'ui-token',
-      getUiSessionTokenFromRequest: () => 'ui-token',
+        },
-      getUiNotificationClients: () => clients,
+        getUiSessionTokenFromRequest: () => 'ui-token',
-      writeSseEvent(res, payload) {
+        getUiNotificationClients: () => clients,
-        res.write(`data: ${JSON.stringify(payload)}\n\n`);
+        writeSseEvent(res, payload) {
-      },
+          res.write(`data: ${JSON.stringify(payload)}\n\n`);
-    });
+        },
+      });
 
-    const handler = getRoute('GET', '/api/notifications/stream');
+      const handler = getRoute('GET', '/api/notifications/stream');
-    const req = createMockRequest();
+      const req = createMockRequest();
-    const res = createMockResponse();
+      const res = createMockResponse();
 
-    await handler(req, res);
+      await handler(req, res);
 
-    expect(res.statusCode).toBe(200);
+      expect(res.statusCode).toBe(200);
-    expect(res.getHeader('content-type')).toContain('text/event-stream');
+      expect(res.getHeader('content-type')).toContain('text/event-stream');
-    expect(res.getHeader('cache-control')).toBe('no-cache, no-transform');
+      expect(res.getHeader('cache-control')).toBe('no-cache, no-transform');
-    expect(res.getHeader('connection')).toBe('keep-alive');
+      expect(res.getHeader('connection')).toBe('keep-alive');
-    expect(res.getHeader('x-accel-buffering')).toBe('no');
+      expect(res.getHeader('x-accel-buffering')).toBe('no');
-    expect(res.flushed).toBe(true);
+      expect(res.flushed).toBe(true);
-    expect(res.body).toContain('openchamber:notification-stream-ready');
+      expect(res.body).toContain('openchamber:notification-stream-ready');
-    expect(clients.has(res)).toBe(true);
+      expect(clients.has(res)).toBe(true);
+      expect(vi.getTimerCount()).toBe(1);
+      expect(res.bodyFlushCount).toBe(1);
 
-    req.emit('close');
+      vi.advanceTimersByTime(NOTIFICATION_SSE_HEARTBEAT_INTERVAL_MS);
-    expect(clients.has(res)).toBe(false);
+      expect(res.body).toContain(':heartbeat\n\n');
+      expect(res.bodyFlushCount).toBe(2);
+
+      res.emit('error');
+      expect(clients.has(res)).toBe(false);
+      expect(vi.getTimerCount()).toBe(0);
+
+      const bodyAfterClose = res.body;
+      vi.advanceTimersByTime(NOTIFICATION_SSE_HEARTBEAT_INTERVAL_MS);
+      expect(res.body).toBe(bodyAfterClose);
+    } finally {
+      vi.useRealTimers();
+    }
   });
 
   it('serves OpenChamber SSE with nginx-safe headers', () => {