feat(web): add Express SSE event adapter

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

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<string, string>();
+  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<typeof vi.fn> }) {
+  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');
+  });
+});