diff --git a/packages/web/server/lib/notifications/DOCUMENTATION.md b/packages/web/server/lib/notifications/DOCUMENTATION.md index 8092c11c..01ff1f6f 100644 --- a/packages/web/server/lib/notifications/DOCUMENTATION.md +++ b/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 diff --git a/packages/web/server/lib/notifications/routes.js b/packages/web/server/lib/notifications/routes.js index 83b28f84..32ea30e4 100644 --- a/packages/web/server/lib/notifications/routes.js +++ b/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) => { diff --git a/packages/web/server/sse-routes.test.js b/packages/web/server/sse-routes.test.js index a37f051b..46551644 100644 --- a/packages/web/server/sse-routes.test.js +++ b/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, { - uiAuthController: { - ensureSessionToken: async () => 'ui-token', - }, - getUiSessionTokenFromRequest: () => 'ui-token', - getUiNotificationClients: () => clients, - writeSseEvent(res, payload) { - res.write(`data: ${JSON.stringify(payload)}\n\n`); - }, - }); + try { + registerNotificationRoutes(app, { + uiAuthController: { + ensureSessionToken: async () => 'ui-token', + }, + getUiSessionTokenFromRequest: () => 'ui-token', + getUiNotificationClients: () => clients, + writeSseEvent(res, payload) { + res.write(`data: ${JSON.stringify(payload)}\n\n`); + }, + }); - const handler = getRoute('GET', '/api/notifications/stream'); - const req = createMockRequest(); - const res = createMockResponse(); + const handler = getRoute('GET', '/api/notifications/stream'); + const req = createMockRequest(); + const res = createMockResponse(); - await handler(req, res); + await handler(req, res); - expect(res.statusCode).toBe(200); - expect(res.getHeader('content-type')).toContain('text/event-stream'); - expect(res.getHeader('cache-control')).toBe('no-cache, no-transform'); - expect(res.getHeader('connection')).toBe('keep-alive'); - expect(res.getHeader('x-accel-buffering')).toBe('no'); - expect(res.flushed).toBe(true); - expect(res.body).toContain('openchamber:notification-stream-ready'); - expect(clients.has(res)).toBe(true); + expect(res.statusCode).toBe(200); + expect(res.getHeader('content-type')).toContain('text/event-stream'); + expect(res.getHeader('cache-control')).toBe('no-cache, no-transform'); + expect(res.getHeader('connection')).toBe('keep-alive'); + expect(res.getHeader('x-accel-buffering')).toBe('no'); + expect(res.flushed).toBe(true); + expect(res.body).toContain('openchamber:notification-stream-ready'); + expect(clients.has(res)).toBe(true); + expect(vi.getTimerCount()).toBe(1); + expect(res.bodyFlushCount).toBe(1); - req.emit('close'); - expect(clients.has(res)).toBe(false); + vi.advanceTimersByTime(NOTIFICATION_SSE_HEARTBEAT_INTERVAL_MS); + 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', () => {