diff --git a/src/web/events/express-sse-adapter.ts b/src/web/events/express-sse-adapter.ts new file mode 100644 index 0000000..052f637 --- /dev/null +++ b/src/web/events/express-sse-adapter.ts @@ -0,0 +1,223 @@ +import type { RequestHandler, Response } from 'express'; +import { + systemEventClock, + type EventClock, +} from './event-dependencies.js'; +import type { RealtimeEvent } from './event-types.js'; +import type { + RealtimeEventHub, + RealtimeEventSubscription, +} from './realtime-event-hub.js'; + +const DEFAULT_HEARTBEAT_INTERVAL_MS = 15_000; +const DEFAULT_MAX_PENDING_FRAMES = 100; + +export interface RecurringTask { + cancel(): void; +} + +export interface RecurringTaskScheduler { + every(intervalMs: number, callback: () => void): RecurringTask; +} + +export const systemRecurringTaskScheduler: RecurringTaskScheduler = { + every(intervalMs, callback) { + const timer = setInterval(callback, intervalMs); + timer.unref(); + return { cancel: () => clearInterval(timer) }; + }, +}; + +export interface ExpressSseHandlerOptions { + hub: RealtimeEventHub; + scheduler?: RecurringTaskScheduler; + clock?: EventClock; + heartbeatIntervalMs?: number; + maxPendingFrames?: number; +} + +function isResponseClosed(response: Response): boolean { + return response.destroyed || response.writableEnded; +} + +function serializeEvent(event: RealtimeEvent): string { + const safeId = event.id.replace(/[\r\n\0]/g, ''); + return `id: ${safeId}\nevent: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`; +} + +/** + * Adapts the transport-agnostic realtime hub to an Express SSE endpoint. + * The returned handler owns only the lifetime of its individual connection. + */ +export function createExpressSseHandler( + options: ExpressSseHandlerOptions, +): RequestHandler { + const heartbeatIntervalMs = + options.heartbeatIntervalMs ?? DEFAULT_HEARTBEAT_INTERVAL_MS; + const maxPendingFrames = + options.maxPendingFrames ?? DEFAULT_MAX_PENDING_FRAMES; + if (!Number.isInteger(heartbeatIntervalMs) || heartbeatIntervalMs <= 0) { + throw new RangeError('heartbeatIntervalMs must be a positive integer'); + } + if (!Number.isInteger(maxPendingFrames) || maxPendingFrames <= 0) { + throw new RangeError('maxPendingFrames must be a positive integer'); + } + + const scheduler = options.scheduler ?? systemRecurringTaskScheduler; + const clock = options.clock ?? systemEventClock; + + return (request, response): void => { + if (isResponseClosed(response)) { + return; + } + + let active = true; + let waitingForDrain = false; + let subscription: RealtimeEventSubscription | undefined; + let heartbeat: RecurringTask | undefined; + const pendingFrames: string[] = []; + + const removeConnectionListeners = (): void => { + request.removeListener('close', handleConnectionClose); + response.removeListener('close', handleConnectionClose); + response.removeListener('drain', drainPendingFrames); + }; + + const close = (endResponse: boolean): void => { + if (!active) { + return; + } + active = false; + waitingForDrain = false; + pendingFrames.length = 0; + removeConnectionListeners(); + heartbeat?.cancel(); + subscription?.unsubscribe(); + if (endResponse && !isResponseClosed(response)) { + try { + response.end(); + } catch { + // The socket is already unusable; all owned resources are released. + } + } + }; + + const handleConnectionClose = (): void => close(false); + + const waitForDrain = (): void => { + if (active && !waitingForDrain) { + waitingForDrain = true; + response.once('drain', drainPendingFrames); + } + }; + + const writeFrame = (frame: string): boolean => { + if (!active || isResponseClosed(response)) { + close(false); + return false; + } + try { + if (!response.write(frame)) { + waitForDrain(); + } + return true; + } catch { + close(true); + return false; + } + }; + + const enqueueOrWrite = (frame: string): void => { + if (!active) { + return; + } + if (!waitingForDrain) { + writeFrame(frame); + return; + } + if (pendingFrames.length >= maxPendingFrames) { + close(true); + return; + } + pendingFrames.push(frame); + }; + + function drainPendingFrames(): void { + if (!active || isResponseClosed(response)) { + close(false); + return; + } + waitingForDrain = false; + while (active && !waitingForDrain && pendingFrames.length > 0) { + const frame = pendingFrames.shift(); + if (frame !== undefined) { + writeFrame(frame); + } + } + } + + request.once('close', handleConnectionClose); + response.once('close', handleConnectionClose); + + try { + response.statusCode = 200; + response.setHeader('Content-Type', 'text/event-stream; charset=utf-8'); + response.setHeader('Cache-Control', 'no-cache, no-transform'); + response.setHeader('Connection', 'keep-alive'); + response.setHeader('X-Accel-Buffering', 'no'); + response.flushHeaders(); + } catch { + close(true); + return; + } + + enqueueOrWrite(': connected\n\n'); + if (!active) { + return; + } + + try { + heartbeat = scheduler.every(heartbeatIntervalMs, () => { + if (!active || waitingForDrain) { + return; + } + try { + enqueueOrWrite(`: heartbeat ${clock.now().toISOString()}\n\n`); + } catch { + close(true); + } + }); + } catch { + close(true); + return; + } + if (!active) { + heartbeat.cancel(); + return; + } + + try { + subscription = options.hub.subscribe( + (event) => { + if (!active) { + return; + } + try { + enqueueOrWrite(serializeEvent(event)); + } catch { + close(true); + } + }, + { lastEventId: request.get('Last-Event-ID') }, + ); + } catch { + close(true); + return; + } + + // Replay is synchronous, so a failed replay can close before assignment. + if (!active) { + subscription.unsubscribe(); + } + }; +} diff --git a/tests/web/events/express-sse-adapter.test.ts b/tests/web/events/express-sse-adapter.test.ts new file mode 100644 index 0000000..d5e640c --- /dev/null +++ b/tests/web/events/express-sse-adapter.test.ts @@ -0,0 +1,296 @@ +import { EventEmitter } from 'node:events'; +import type { Request, Response } from 'express'; +import { describe, expect, it, vi } from 'vitest'; +import { + RealtimeEventHub, +} from '../../../src/web/events/index.js'; +import { + createExpressSseHandler, + type RecurringTask, + type RecurringTaskScheduler, +} from '../../../src/web/events/express-sse-adapter.js'; + +class FakeRequest extends EventEmitter { + constructor(private readonly lastEventId?: string) { + super(); + } + + get(name: string): string | undefined { + return name.toLowerCase() === 'last-event-id' + ? this.lastEventId + : undefined; + } +} + +class FakeResponse extends EventEmitter { + readonly headers = new Map(); + readonly frames: string[] = []; + readonly flushHeaders = vi.fn(); + readonly end = vi.fn(() => { + this.writableEnded = true; + return this; + }); + destroyed = false; + writableEnded = false; + writeResults: boolean[] = []; + throwOnWriteCall: number | undefined; + private writeCalls = 0; + + setHeader(name: string, value: string): this { + this.headers.set(name, value); + return this; + } + + write(frame: string): boolean { + this.writeCalls += 1; + if (this.writeCalls === this.throwOnWriteCall) { + throw new Error('socket write failed'); + } + this.frames.push(frame); + return this.writeResults.shift() ?? true; + } +} + +class FakeRecurringTask implements RecurringTask { + readonly cancel = vi.fn(); +} + +class FakeScheduler implements RecurringTaskScheduler { + readonly scheduled: Array<{ + intervalMs: number; + callback: () => void; + task: FakeRecurringTask; + }> = []; + + every(intervalMs: number, callback: () => void): RecurringTask { + const task = new FakeRecurringTask(); + this.scheduled.push({ intervalMs, callback, task }); + return task; + } +} + +function asRequest(request: FakeRequest): Request { + return request as unknown as Request; +} + +function asResponse(response: FakeResponse): Response { + return response as unknown as Response; +} + +function createHub(logger?: { warn: ReturnType }) { + let sequence = 0; + return new RealtimeEventHub({ + capacity: 10, + clock: { now: () => new Date('2026-08-08T12:00:00.000Z') }, + idGenerator: { next: () => `event-${++sequence}` }, + logger, + }); +} + +describe('createExpressSseHandler', () => { + it('sets streaming headers and resumes ordered event frames after Last-Event-ID', () => { + const hub = createHub(); + hub.publish('log.created', { level: 'info', message: 'already seen' }); + const second = hub.publish('bot.status.changed', { + status: 'running', + dryRun: false, + }); + const third = hub.publish('sync.started', { + syncId: 'sync-1', + kind: 'media', + }); + const scheduler = new FakeScheduler(); + const handler = createExpressSseHandler({ hub, scheduler }); + const request = new FakeRequest('event-1'); + const response = new FakeResponse(); + + handler(asRequest(request), asResponse(response), vi.fn()); + + expect(response.headers).toEqual( + new Map([ + ['Content-Type', 'text/event-stream; charset=utf-8'], + ['Cache-Control', 'no-cache, no-transform'], + ['Connection', 'keep-alive'], + ['X-Accel-Buffering', 'no'], + ]), + ); + expect(response.flushHeaders).toHaveBeenCalledOnce(); + expect(response.frames).toEqual([ + ': connected\n\n', + `id: event-2\nevent: bot.status.changed\ndata: ${JSON.stringify(second)}\n\n`, + `id: event-3\nevent: sync.started\ndata: ${JSON.stringify(third)}\n\n`, + ]); + }); + + it('replays the full available window when Last-Event-ID is absent', () => { + const hub = createHub(); + hub.publish('log.created', { level: 'info', message: 'one' }); + hub.publish('log.created', { level: 'warn', message: 'two' }); + const response = new FakeResponse(); + + createExpressSseHandler({ hub, scheduler: new FakeScheduler() })( + asRequest(new FakeRequest()), + asResponse(response), + vi.fn(), + ); + + expect(response.frames.map((frame) => frame.split('\n')[0])).toEqual([ + ': connected', + 'id: event-1', + 'id: event-2', + ]); + }); + + it('emits timestamped heartbeats through injected scheduling and clock dependencies', () => { + const hub = createHub(); + const scheduler = new FakeScheduler(); + const response = new FakeResponse(); + const handler = createExpressSseHandler({ + hub, + scheduler, + clock: { now: () => new Date('2026-08-08T12:34:56.789Z') }, + heartbeatIntervalMs: 5_000, + }); + handler(asRequest(new FakeRequest()), asResponse(response), vi.fn()); + + expect(scheduler.scheduled).toHaveLength(1); + expect(scheduler.scheduled[0]?.intervalMs).toBe(5_000); + scheduler.scheduled[0]?.callback(); + + expect(response.frames.at(-1)).toBe( + ': heartbeat 2026-08-08T12:34:56.789Z\n\n', + ); + }); + + it('unsubscribes and cancels its heartbeat when the request closes', () => { + const hub = createHub(); + const scheduler = new FakeScheduler(); + const response = new FakeResponse(); + const request = new FakeRequest(); + createExpressSseHandler({ hub, scheduler })( + asRequest(request), + asResponse(response), + vi.fn(), + ); + + request.emit('close'); + request.emit('close'); + hub.publish('log.created', { level: 'info', message: 'after close' }); + + expect(scheduler.scheduled[0]?.task.cancel).toHaveBeenCalledOnce(); + expect(response.frames).toEqual([': connected\n\n']); + }); + + it('queues live events during backpressure and flushes them in order on drain', () => { + const hub = createHub(); + const response = new FakeResponse(); + response.writeResults = [true, false, true]; + createExpressSseHandler({ hub, scheduler: new FakeScheduler() })( + asRequest(new FakeRequest()), + asResponse(response), + vi.fn(), + ); + + hub.publish('log.created', { level: 'info', message: 'fills buffer' }); + hub.publish('log.created', { level: 'info', message: 'queued' }); + expect(response.frames.map((frame) => frame.split('\n')[0])).toEqual([ + ': connected', + 'id: event-1', + ]); + + response.emit('drain'); + expect(response.frames.map((frame) => frame.split('\n')[0])).toEqual([ + ': connected', + 'id: event-1', + 'id: event-2', + ]); + }); + + it('closes a persistently backpressured response when its bounded queue fills', () => { + const hub = createHub(); + const scheduler = new FakeScheduler(); + const response = new FakeResponse(); + response.writeResults = [true, false]; + createExpressSseHandler({ + hub, + scheduler, + maxPendingFrames: 1, + })(asRequest(new FakeRequest()), asResponse(response), vi.fn()); + + hub.publish('log.created', { level: 'info', message: 'buffered by Node' }); + hub.publish('log.created', { level: 'info', message: 'adapter queue' }); + hub.publish('log.created', { level: 'info', message: 'overflow' }); + hub.publish('log.created', { level: 'info', message: 'not delivered' }); + + expect(response.end).toHaveBeenCalledOnce(); + expect(scheduler.scheduled[0]?.task.cancel).toHaveBeenCalledOnce(); + expect(response.frames.map((frame) => frame.split('\n')[0])).toEqual([ + ': connected', + 'id: event-1', + ]); + }); + + it('handles closed and throwing responses without leaking subscriber errors', () => { + const logger = { warn: vi.fn() }; + const closedHub = createHub(logger); + const closedResponse = new FakeResponse(); + closedResponse.destroyed = true; + createExpressSseHandler({ + hub: closedHub, + scheduler: new FakeScheduler(), + })(asRequest(new FakeRequest()), asResponse(closedResponse), vi.fn()); + closedHub.publish('log.created', { level: 'info', message: 'ignored' }); + + const throwingHub = createHub(logger); + const scheduler = new FakeScheduler(); + const throwingResponse = new FakeResponse(); + throwingResponse.throwOnWriteCall = 2; + createExpressSseHandler({ hub: throwingHub, scheduler })( + asRequest(new FakeRequest()), + asResponse(throwingResponse), + vi.fn(), + ); + throwingHub.publish('log.created', { level: 'error', message: 'throws' }); + throwingHub.publish('log.created', { level: 'info', message: 'ignored' }); + + expect(logger.warn).not.toHaveBeenCalled(); + expect(scheduler.scheduled[0]?.task.cancel).toHaveBeenCalledOnce(); + expect(throwingResponse.end).toHaveBeenCalledOnce(); + }); + + it('releases the connection if the event hub cannot accept a subscriber', () => { + const hub = createHub(); + hub.close(); + const scheduler = new FakeScheduler(); + const response = new FakeResponse(); + const handler = createExpressSseHandler({ hub, scheduler }); + + expect(() => + handler( + asRequest(new FakeRequest()), + asResponse(response), + vi.fn(), + ), + ).not.toThrow(); + expect(response.end).toHaveBeenCalledOnce(); + expect(scheduler.scheduled[0]?.task.cancel).toHaveBeenCalledOnce(); + }); + + it('validates adapter limits', () => { + const hub = createHub(); + expect(() => + createExpressSseHandler({ + hub, + scheduler: new FakeScheduler(), + heartbeatIntervalMs: 0, + }), + ).toThrow('heartbeatIntervalMs must be a positive integer'); + expect(() => + createExpressSseHandler({ + hub, + scheduler: new FakeScheduler(), + maxPendingFrames: 0, + }), + ).toThrow('maxPendingFrames must be a positive integer'); + }); +});