diff --git a/src/web/events/event-dependencies.ts b/src/web/events/event-dependencies.ts new file mode 100644 index 0000000..97c8429 --- /dev/null +++ b/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): 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')}`; + } +} diff --git a/src/web/events/event-types.ts b/src/web/events/event-types.ts new file mode 100644 index 0000000..e8c6da1 --- /dev/null +++ b/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; + }> + : never; diff --git a/src/web/events/index.ts b/src/web/events/index.ts new file mode 100644 index 0000000..c41cc1e --- /dev/null +++ b/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'; diff --git a/src/web/events/realtime-event-hub.ts b/src/web/events/realtime-event-hub.ts new file mode 100644 index 0000000..89462fb --- /dev/null +++ b/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[]; +} + +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; + private readonly clock: EventClock; + private readonly idGenerator: EventIdGenerator; + private readonly logger?: EventHubLogger; + private readonly subscribers = new Set(); + private readonly dispatchQueue: Sequenced[] = []; + 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( + type: TType, + payload: RealtimeEventMap[TType], + ): RealtimeEvent { + 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; + const sequenced: Sequenced = { + 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 | 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, + ): 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'); + } + } +} diff --git a/src/web/events/replay-buffer.ts b/src/web/events/replay-buffer.ts new file mode 100644 index 0000000..fadedf4 --- /dev/null +++ b/src/web/events/replay-buffer.ts @@ -0,0 +1,42 @@ +export interface Sequenced { + readonly sequence: number; + readonly value: TValue; +} + +export class ReplayBuffer { + private values: Sequenced[] = []; + + constructor(readonly capacity: number) { + if (!Number.isInteger(capacity) || capacity <= 0) { + throw new RangeError('capacity must be a positive integer'); + } + } + + push(value: Sequenced): void { + this.values.push(value); + if (this.values.length > this.capacity) { + this.values.shift(); + } + } + + after(lastEventId?: string): readonly Sequenced[] { + 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 = []; + } +} diff --git a/tests/web/events/realtime-event-hub.test.ts b/tests/web/events/realtime-event-hub.test.ts new file mode 100644 index 0000000..2080bf7 --- /dev/null +++ b/tests/web/events/realtime-event-hub.test.ts @@ -0,0 +1,256 @@ +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().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()); + }); +});