RedditBot/src/web/events/realtime-event-hub.ts

import {
  MonotonicEventIdGenerator,
  systemEventClock,
  type EventClock,
  type EventHubLogger,
  type EventIdGenerator,
} from './event-dependencies.js';
import {
  type RealtimeEvent,
  type RealtimeEventMap,
  type RealtimeEventType,
} from './event-types.js';
import { ReplayBuffer, type Sequenced } from './replay-buffer.js';

export type RealtimeEventListener = (event: RealtimeEvent) => void;

export interface SubscribeOptions {
  /** Value from the SSE `Last-Event-ID` request header. */
  lastEventId?: string;
}

export interface RealtimeEventSubscription {
  readonly closed: boolean;
  unsubscribe(): void;
}

export interface RealtimeEventHubOptions {
  capacity: number;
  clock?: EventClock;
  idGenerator?: EventIdGenerator;
  logger?: EventHubLogger;
}

interface SubscriberState {
  listener: RealtimeEventListener;
  closed: boolean;
  replaying: boolean;
  lastDeliveredSequence: number;
  queued: Sequenced<RealtimeEvent>[];
}

class HubSubscription implements RealtimeEventSubscription {
  constructor(
    private readonly state: SubscriberState,
    private readonly remove: () => void,
  ) {}

  get closed(): boolean {
    return this.state.closed;
  }

  unsubscribe(): void {
    if (this.state.closed) {
      return;
    }
    this.state.closed = true;
    this.remove();
  }
}
export class RealtimeEventHub {
  private readonly buffer: ReplayBuffer<RealtimeEvent>;
  private readonly clock: EventClock;
  private readonly idGenerator: EventIdGenerator;
  private readonly logger?: EventHubLogger;
  private readonly subscribers = new Set<SubscriberState>();
  private readonly dispatchQueue: Sequenced<RealtimeEvent>[] = [];
  private sequence = 0;
  private dispatching = false;
  private isClosed = false;

  constructor(options: RealtimeEventHubOptions) {
    this.buffer = new ReplayBuffer(options.capacity);
    this.clock = options.clock ?? systemEventClock;
    this.idGenerator = options.idGenerator ?? new MonotonicEventIdGenerator();
    this.logger = options.logger;
  }

  get closed(): boolean {
    return this.isClosed;
  }

  publish<TType extends RealtimeEventType>(
    type: TType,
    payload: RealtimeEventMap[TType],
  ): RealtimeEvent<TType> {
    this.assertOpen();
    const occurredAt = this.clock.now();
    if (Number.isNaN(occurredAt.getTime())) {
      throw new RangeError('clock returned an invalid date');
    }

    const id = this.idGenerator.next(occurredAt);
    if (id.length === 0) {
      throw new Error('idGenerator returned an empty event ID');
    }

    const event = Object.freeze({
      id,
      occurredAt: occurredAt.toISOString(),
      type,
      payload: Object.freeze({ ...payload }),
    }) as RealtimeEvent<TType>;
    const sequenced: Sequenced<RealtimeEvent> = {
      sequence: ++this.sequence,
      value: event,
    };
    this.buffer.push(sequenced);
    this.dispatchQueue.push(sequenced);
    this.drainDispatchQueue();
    return event;
  }

  subscribe(
    listener: RealtimeEventListener,
    options: SubscribeOptions = {},
  ): RealtimeEventSubscription {
    this.assertOpen();
    const state: SubscriberState = {
      listener,
      closed: false,
      replaying: true,
      lastDeliveredSequence: 0,
      queued: [],
    };
    this.subscribers.add(state);
    const subscription = new HubSubscription(state, () => {
      this.subscribers.delete(state);
      state.queued = [];
    });

    for (const event of this.buffer.after(options.lastEventId)) {
      this.deliver(state, event);
      if (state.closed) {
        return subscription;
      }
    }

    state.replaying = false;
    const queued = state.queued;
    state.queued = [];
    for (const event of queued) {
      this.deliver(state, event);
      if (state.closed) {
        break;
      }
    }
    return subscription;
  }

  recent(limit = this.buffer.capacity): readonly RealtimeEvent[] {
    if (!Number.isInteger(limit) || limit < 0) {
      throw new RangeError('limit must be a non-negative integer');
    }
    return limit === 0 ? [] : this.buffer.recent(limit);
  }

  close(): void {
    if (this.isClosed) {
      return;
    }
    this.isClosed = true;
    for (const state of this.subscribers) {
      state.closed = true;
      state.queued = [];
    }
    this.subscribers.clear();
    this.dispatchQueue.length = 0;
    this.buffer.clear();
  }

  private drainDispatchQueue(): void {
    if (this.dispatching) {
      return;
    }
    this.dispatching = true;
    try {
      let next: Sequenced<RealtimeEvent> | undefined;
      while ((next = this.dispatchQueue.shift()) !== undefined) {
        for (const state of [...this.subscribers]) {
          if (state.replaying) {
            state.queued.push(next);
          } else {
            this.deliver(state, next);
          }
        }
      }
    } finally {
      this.dispatching = false;
    }
  }

  private deliver(
    state: SubscriberState,
    event: Sequenced<RealtimeEvent>,
  ): void {
    if (state.closed || event.sequence <= state.lastDeliveredSequence) {
      return;
    }
    state.lastDeliveredSequence = event.sequence;
    try {
      state.listener(event.value);
    } catch (error: unknown) {
      state.closed = true;
      state.queued = [];
      this.subscribers.delete(state);
      try {
        this.logger?.warn(
          'Realtime event subscriber failed; subscription closed',
          { eventId: event.value.id, error },
        );
      } catch {
        // A diagnostic logger must not break event delivery to healthy clients.
      }
    }
  }

  private assertOpen(): void {
    if (this.isClosed) {
      throw new Error('RealtimeEventHub is closed');
    }
  }
}