RedditBot/tests/web/events/express-sse-adapter.test.ts

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('starts at the live edge 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).toEqual([': connected\n\n']);

    hub.publish('log.created', { level: 'info', message: 'live' });
    expect(response.frames.map((frame) => frame.split('\n')[0])).toEqual([
      ': connected',
      'id: event-3',
    ]);
  });

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