spb/market-atlas

1import { EventEmitter } from "node:events";2import type { CanonicalQuote, ConnectorHealth, MarketEvent, Observation, RawObservation } from "@market-atlas/market-model";34/**5 * Internal market bus. In-process today (modular monolith); the topic names mirror the6 * spec so that a NATS/Redpanda transport can replace this class without touching producers.7 */8export interface BusTopics {9  "raw.observation": RawObservation;10  "normalized.observation": Observation;11  "canonical.quote": CanonicalQuote;12  "market.event": MarketEvent;13  "connector.health": ConnectorHealth;14  "market.state": { exchangeId: string; state: string; at: number };15}1617type Handler<T> = (msg: T) => void;1819export class MarketBus {20  private emitter = new EventEmitter({ captureRejections: false });21  readonly counters: Record<string, number> = {};2223  constructor() {24    this.emitter.setMaxListeners(200);25  }2627  publish<K extends keyof BusTopics>(topic: K, msg: BusTopics[K]): void {28    this.counters[topic] = (this.counters[topic] ?? 0) + 1;29    this.emitter.emit(topic, msg);30  }3132  subscribe<K extends keyof BusTopics>(topic: K, handler: Handler<BusTopics[K]>): () => void {33    this.emitter.on(topic, handler);34    return () => this.emitter.off(topic, handler);35  }36}3738export const bus = new MarketBus();39