feat(web): add typed realtime event core

src/web/events/event-dependencies.ts

@@ -0,0 +1,35 @@
+export interface EventClock {
+  now(): Date;
+}
+
+export interface EventIdGenerator {
+  next(occurredAt: Date): string;
+}
+
+/** The subset of Winston's logger contract used by the event hub. */
+export interface EventHubLogger {
+  warn(message: string, metadata?: Record<string, unknown>): unknown;
+}
+
+export const systemEventClock: EventClock = {
+  now: () => new Date(),
+};
+
+/**
+ * Generates lexically sortable IDs and remains monotonic if the wall clock
+ * moves backwards. The per-process sequence disambiguates events in one ms.
+ */
+export class MonotonicEventIdGenerator implements EventIdGenerator {
+  private lastTimestamp = 0;
+  private sequence = 0;
+
+  next(occurredAt: Date): string {
+    const timestamp = Math.max(occurredAt.getTime(), this.lastTimestamp);
+    this.sequence = timestamp === this.lastTimestamp ? this.sequence + 1 : 0;
+    this.lastTimestamp = timestamp;
+
+    return `${timestamp.toString().padStart(13, '0')}-${this.sequence
+      .toString()
+      .padStart(6, '0')}`;
+  }
+}

src/web/events/event-types.ts

@@ -0,0 +1,101 @@
+export type BotRuntimeStatus = 'starting' | 'running' | 'stopping' | 'stopped';
+
+export type ConversationStatus =
+  | 'new'
+  | 'awaiting-reply'
+  | 'replied'
+  | 'closed';
+
+export type SyncKind = 'media' | 'messages' | 'conversations' | 'full';
+
+export type RedditSessionStatus =
+  | 'unknown'
+  | 'authenticating'
+  | 'authenticated'
+  | 'expired'
+  | 'logged-out'
+  | 'error';
+
+export interface RealtimeEventMap {
+  'bot.status.changed': {
+    status: BotRuntimeStatus;
+    dryRun: boolean;
+    reason?: string;
+  };
+  'message.sent': {
+    messageId: string;
+    recipient: string;
+    conversationId?: string;
+  };
+  'message.failed': {
+    recipient: string;
+    reason: string;
+    messageId?: string;
+    conversationId?: string;
+    code?: string;
+    retryable?: boolean;
+  };
+  'conversation.updated': {
+    conversationId: string;
+    status: ConversationStatus;
+    unreadCount?: number;
+  };
+  'reply.received': {
+    conversationId: string;
+    replyId: string;
+    sender: string;
+    preview?: string;
+  };
+  'sync.started': {
+    syncId: string;
+    kind: SyncKind;
+  };
+  'sync.progress': {
+    syncId: string;
+    kind: SyncKind;
+    completed: number;
+    total?: number;
+    message?: string;
+  };
+  'sync.completed': {
+    syncId: string;
+    kind: SyncKind;
+    processed: number;
+  };
+  'sync.failed': {
+    syncId: string;
+    kind: SyncKind;
+    reason: string;
+    code?: string;
+  };
+  'rate-limit.changed': {
+    limited: boolean;
+    remaining?: number;
+    resetsAt?: string;
+    reason?: string;
+  };
+  'reddit.session.changed': {
+    status: RedditSessionStatus;
+    username?: string;
+    reason?: string;
+  };
+  'log.created': {
+    level: 'debug' | 'info' | 'warn' | 'error';
+    message: string;
+    data?: unknown;
+  };
+}
+
+export type RealtimeEventType = keyof RealtimeEventMap;
+
+/** A distributive type so the default is a discriminated union. */
+export type RealtimeEvent<
+  TType extends RealtimeEventType = RealtimeEventType,
+> = TType extends RealtimeEventType
+  ? Readonly<{
+      id: string;
+      occurredAt: string;
+      type: TType;
+      payload: Readonly<RealtimeEventMap[TType]>;
+    }>
+  : never;

src/web/events/index.ts

@@ -0,0 +1,23 @@
+export {
+  MonotonicEventIdGenerator,
+  systemEventClock,
+  type EventClock,
+  type EventHubLogger,
+  type EventIdGenerator,
+} from './event-dependencies.js';
+export {
+  type BotRuntimeStatus,
+  type ConversationStatus,
+  type RealtimeEvent,
+  type RealtimeEventMap,
+  type RealtimeEventType,
+  type RedditSessionStatus,
+  type SyncKind,
+} from './event-types.js';
+export {
+  RealtimeEventHub,
+  type RealtimeEventHubOptions,
+  type RealtimeEventListener,
+  type RealtimeEventSubscription,
+  type SubscribeOptions,
+} from './realtime-event-hub.js';

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

@@ -0,0 +1,222 @@
+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');
+    }
+  }
+}

src/web/events/replay-buffer.ts

@@ -0,0 +1,42 @@
+export interface Sequenced<TValue> {
+  readonly sequence: number;
+  readonly value: TValue;
+}
+
+export class ReplayBuffer<TValue extends { readonly id: string }> {
+  private values: Sequenced<TValue>[] = [];
+
+  constructor(readonly capacity: number) {
+    if (!Number.isInteger(capacity) || capacity <= 0) {
+      throw new RangeError('capacity must be a positive integer');
+    }
+  }
+
+  push(value: Sequenced<TValue>): void {
+    this.values.push(value);
+    if (this.values.length > this.capacity) {
+      this.values.shift();
+    }
+  }
+
+  after(lastEventId?: string): readonly Sequenced<TValue>[] {
+    if (lastEventId === undefined || lastEventId.length === 0) {
+      return [...this.values];
+    }
+
+    const matchingIndex = this.values.findIndex(
+      ({ value }) => value.id === lastEventId,
+    );
+    return matchingIndex === -1
+      ? [...this.values]
+      : this.values.slice(matchingIndex + 1);
+  }
+
+  recent(limit: number): readonly TValue[] {
+    return this.values.slice(-limit).map(({ value }) => value);
+  }
+
+  clear(): void {
+    this.values = [];
+  }
+}