RedditBot/tests/web/events/realtime-event-hub.test.ts

import { describe, expect, expectTypeOf, it, vi } from 'vitest';
import {
  MonotonicEventIdGenerator,
  RealtimeEventHub,
  type RealtimeEvent,
  type RealtimeEventMap,
} from '../../../src/web/events/index.js';

function createDependencies() {
  let timestamp = Date.parse('2026-08-08T12:00:00.000Z');
  let sequence = 0;

  return {
    clock: {
      now: () => new Date(timestamp++),
    },
    idGenerator: {
      next: () => `event-${++sequence}`,
    },
  };
}

describe('RealtimeEventHub', () => {
  it('creates typed envelopes with stable IDs and timestamps', () => {
    const hub = new RealtimeEventHub({
      capacity: 10,
      ...createDependencies(),
    });

    const event = hub.publish('bot.status.changed', {
      status: 'running',
      dryRun: false,
    });

    expect(event).toStrictEqual({
      id: 'event-1',
      occurredAt: '2026-08-08T12:00:00.000Z',
      type: 'bot.status.changed',
      payload: { status: 'running', dryRun: false },
    });
    expectTypeOf(event).toEqualTypeOf<
      RealtimeEvent<'bot.status.changed'>
    >();
  });

  it('supports every event family through the event map', () => {
    expectTypeOf<keyof RealtimeEventMap>().toEqualTypeOf<
      | 'bot.status.changed'
      | 'message.sent'
      | 'message.failed'
      | 'conversation.updated'
      | 'reply.received'
      | 'sync.started'
      | 'sync.progress'
      | 'sync.completed'
      | 'sync.failed'
      | 'rate-limit.changed'
      | 'reddit.session.changed'
      | 'log.created'
    >();
  });

  it('fans out live events and stops delivery after unsubscribe', () => {
    const hub = new RealtimeEventHub({
      capacity: 10,
      ...createDependencies(),
    });
    const listener = vi.fn();
    const subscription = hub.subscribe(listener);

    hub.publish('message.sent', {
      messageId: 'message-1',
      recipient: 'redditor',
    });
    subscription.unsubscribe();
    subscription.unsubscribe();
    hub.publish('conversation.updated', {
      conversationId: 'conversation-1',
      status: 'awaiting-reply',
    });

    expect(listener).toHaveBeenCalledTimes(1);
    expect(listener.mock.calls[0]?.[0]).toMatchObject({
      id: 'event-1',
      type: 'message.sent',
    });
    expect(subscription.closed).toBe(true);
  });

  it('retains only the configured replay capacity', () => {
    const hub = new RealtimeEventHub({
      capacity: 2,
      ...createDependencies(),
    });

    hub.publish('log.created', { level: 'info', message: 'one' });
    hub.publish('log.created', { level: 'info', message: 'two' });
    hub.publish('log.created', { level: 'info', message: 'three' });

    expect(hub.recent()).toEqual([
      expect.objectContaining({ id: 'event-2' }),
      expect.objectContaining({ id: 'event-3' }),
    ]);
    expect(hub.recent(1)).toEqual([
      expect.objectContaining({ id: 'event-3' }),
    ]);
  });

  it('resumes strictly after Last-Event-ID and replays the available window for an unknown ID', () => {
    const hub = new RealtimeEventHub({
      capacity: 3,
      ...createDependencies(),
    });
    hub.publish('log.created', { level: 'info', message: 'one' });
    hub.publish('log.created', { level: 'info', message: 'two' });
    hub.publish('log.created', { level: 'info', message: 'three' });

    const resumed = vi.fn();
    hub.subscribe(resumed, { lastEventId: 'event-1' });
    expect(resumed.mock.calls.map(([event]) => event.id)).toEqual([
      'event-2',
      'event-3',
    ]);

    const boundedHub = new RealtimeEventHub({
      capacity: 2,
      ...createDependencies(),
    });
    boundedHub.publish('log.created', { level: 'info', message: 'evicted' });
    boundedHub.publish('log.created', { level: 'info', message: 'available one' });
    boundedHub.publish('log.created', { level: 'info', message: 'available two' });
    const stale = vi.fn();
    boundedHub.subscribe(stale, { lastEventId: 'event-1' });
    expect(stale.mock.calls.map(([event]) => event.id)).toEqual([
      'event-2',
      'event-3',
    ]);
  });

  it('preserves order and prevents duplicate delivery when publish occurs during replay', () => {
    const hub = new RealtimeEventHub({
      capacity: 10,
      ...createDependencies(),
    });
    hub.publish('log.created', { level: 'info', message: 'one' });
    hub.publish('log.created', { level: 'info', message: 'two' });
    const received: string[] = [];

    hub.subscribe((event) => {
      received.push(event.id);
      if (event.id === 'event-1') {
        hub.publish('log.created', { level: 'info', message: 'during replay' });
      }
    });

    expect(received).toEqual(['event-1', 'event-2', 'event-3']);
  });

  it('keeps every subscriber in order across re-entrant publishes', () => {
    const hub = new RealtimeEventHub({
      capacity: 10,
      ...createDependencies(),
    });
    const firstReceived: string[] = [];
    const secondReceived: string[] = [];
    hub.subscribe((event) => {
      firstReceived.push(event.id);
      if (event.id === 'event-1') {
        hub.publish('log.created', { level: 'info', message: 'nested' });
      }
    });
    hub.subscribe((event) => secondReceived.push(event.id));

    hub.publish('log.created', { level: 'info', message: 'outer' });

    expect(firstReceived).toEqual(['event-1', 'event-2']);
    expect(secondReceived).toEqual(['event-1', 'event-2']);
  });

  it('isolates failed subscribers and reports failures through an injected logger', () => {
    const logger = { warn: vi.fn() };
    const hub = new RealtimeEventHub({
      capacity: 10,
      logger,
      ...createDependencies(),
    });
    const healthyListener = vi.fn();
    const failedSubscription = hub.subscribe(() => {
      throw new Error('client disconnected');
    });
    hub.subscribe(healthyListener);

    hub.publish('reddit.session.changed', { status: 'authenticated' });
    hub.publish('reddit.session.changed', { status: 'expired' });

    expect(failedSubscription.closed).toBe(true);
    expect(healthyListener).toHaveBeenCalledTimes(2);
    expect(logger.warn).toHaveBeenCalledOnce();
    expect(logger.warn).toHaveBeenCalledWith(
      'Realtime event subscriber failed; subscription closed',
      expect.objectContaining({ eventId: 'event-1', error: expect.any(Error) }),
    );
  });

  it('closes all subscriptions cleanly and rejects use after close', () => {
    const hub = new RealtimeEventHub({
      capacity: 10,
      ...createDependencies(),
    });
    const first = hub.subscribe(vi.fn());
    const second = hub.subscribe(vi.fn());

    hub.close();
    hub.close();

    expect(first.closed).toBe(true);
    expect(second.closed).toBe(true);
    expect(hub.closed).toBe(true);
    expect(() =>
      hub.publish('sync.started', { syncId: 'sync-1', kind: 'media' }),
    ).toThrow('RealtimeEventHub is closed');
    expect(() => hub.subscribe(vi.fn())).toThrow(
      'RealtimeEventHub is closed',
    );
  });

  it('validates construction and recent limits', () => {
    expect(
      () =>
        new RealtimeEventHub({
          capacity: 0,
          ...createDependencies(),
        }),
    ).toThrow('capacity must be a positive integer');

    const hub = new RealtimeEventHub({
      capacity: 1,
      ...createDependencies(),
    });
    expect(() => hub.recent(-1)).toThrow(
      'limit must be a non-negative integer',
    );
  });
});

describe('MonotonicEventIdGenerator', () => {
  it('generates unique sortable IDs when timestamps repeat or move backwards', () => {
    const generator = new MonotonicEventIdGenerator();
    const first = generator.next(new Date('2026-08-08T12:00:00.001Z'));
    const second = generator.next(new Date('2026-08-08T12:00:00.001Z'));
    const third = generator.next(new Date('2026-08-08T12:00:00.000Z'));

    expect(new Set([first, second, third])).toHaveLength(3);
    expect([first, second, third]).toEqual([...[first, second, third]].sort());
  });
});