Refactor config, payments, websocket, and routing

backend/services/notification/__tests__/websocket.test.ts

@@ -1,4 +1,4 @@
-import { WebSocketServer, SubscriptionEvent } from '../websocket';
+import { InMemorySubscriptionEventBus, WebSocketServer, SubscriptionEvent } from '../websocket';
 
 const makeEvent = (overrides: Partial<SubscriptionEvent> = {}): SubscriptionEvent => ({
   type: 'subscription.created',
@@ -16,6 +16,11 @@ describe('WebSocketServer', () => {
     server = new WebSocketServer();
   });
 
+  afterEach(() => {
+    server.shutdown();
+    jest.useRealTimers();
+  });
+
   // ── Connection / presence ─────────────────────────────────────────────────
 
   it('connects a client and tracks presence', () => {
@@ -130,4 +135,72 @@ describe('WebSocketServer', () => {
     server.broadcast(makeEvent());
     expect(handler).toHaveBeenCalledWith(expect.objectContaining({ delivered: 1 }));
   });
+
+  it('routes broadcasts through an injected event bus', () => {
+    const eventBus = new InMemorySubscriptionEventBus();
+    server.shutdown();
+    server = new WebSocketServer({}, eventBus);
+
+    const published = jest.fn();
+    eventBus.on('subscription.event.published', published);
+
+    const send = jest.fn();
+    server.connect('c1', 'user-1', send);
+    const delivered = server.broadcast(makeEvent({ userId: 'user-1' }));
+
+    expect(delivered).toBe(1);
+    expect(send).toHaveBeenCalledTimes(1);
+    expect(published).toHaveBeenCalledWith(
+      expect.objectContaining({ matchedClients: 1 })
+    );
+  });
+
+  it('can consume events published by another producer on the same bus', () => {
+    const eventBus = new InMemorySubscriptionEventBus();
+    server.shutdown();
+    server = new WebSocketServer({}, eventBus);
+    const send = jest.fn();
+    server.connect('c1', 'user-1', send, { userId: 'user-1' });
+
+    const delivered = eventBus.publish(makeEvent({ userId: 'user-1' }));
+
+    expect(delivered).toBe(1);
+    expect(send).toHaveBeenCalledTimes(1);
+  });
+
+  it('batches messages when batchIntervalMs is configured', () => {
+    jest.useFakeTimers();
+    server.shutdown();
+    server = new WebSocketServer({ batchIntervalMs: 50, heartbeatIntervalMs: 0 });
+    const send = jest.fn();
+    server.connect('c1', 'user-1', send);
+
+    server.broadcast(makeEvent({ subscriptionId: 'sub-1' }));
+    server.broadcast(makeEvent({ subscriptionId: 'sub-2' }));
+
+    expect(send).not.toHaveBeenCalled();
+    jest.advanceTimersByTime(50);
+    expect(send).toHaveBeenCalledTimes(2);
+    expect(server.getMetrics().batchesFlushed).toBe(1);
+  });
+
+  it('drops the oldest queued event when a batched client exceeds queue capacity', () => {
+    jest.useFakeTimers();
+    server.shutdown();
+    server = new WebSocketServer({
+      batchIntervalMs: 50,
+      heartbeatIntervalMs: 0,
+      maxQueueSize: 1,
+    });
+    const send = jest.fn();
+    server.connect('c1', 'user-1', send);
+
+    server.broadcast(makeEvent({ subscriptionId: 'old' }));
+    server.broadcast(makeEvent({ subscriptionId: 'new' }));
+    jest.advanceTimersByTime(50);
+
+    expect(send).toHaveBeenCalledTimes(1);
+    expect(send).toHaveBeenCalledWith(expect.objectContaining({ subscriptionId: 'new' }));
+    expect(server.getMetrics().eventsDropped).toBe(1);
+  });
 });

backend/services/notification/websocket.ts

@@ -75,7 +75,7 @@ export interface WebSocketServerConfig {
    * Batch flush interval in milliseconds.
    * Messages accumulate and are sent together at each interval.
    * 0 = disabled (send immediately, legacy behaviour).
-   * Default: 50 ms.
+   * Default: 0 ms.
    */
   batchIntervalMs?: number;
   /**
@@ -97,6 +97,35 @@ export interface WebSocketServerConfig {
   pingTimeoutMs?: number;
 }
 
+export type SubscriptionEventHandler = (event: SubscriptionEvent) => number;
+
+export interface SubscriptionEventBus {
+  publish(event: SubscriptionEvent): number;
+  subscribe(handler: SubscriptionEventHandler): () => void;
+}
+
+export class InMemorySubscriptionEventBus extends EventEmitter implements SubscriptionEventBus {
+  private readonly handlers = new Set<SubscriptionEventHandler>();
+
+  publish(event: SubscriptionEvent): number {
+    let matchedClients = 0;
+    for (const handler of this.handlers) {
+      matchedClients += handler(event);
+    }
+    this.emit('subscription.event.published', { event, matchedClients });
+    return matchedClients;
+  }
+
+  subscribe(handler: SubscriptionEventHandler): () => void {
+    this.handlers.add(handler);
+    this.emit('subscription.handler.registered', { handlerCount: this.handlers.size });
+    return () => {
+      this.handlers.delete(handler);
+      this.emit('subscription.handler.removed', { handlerCount: this.handlers.size });
+    };
+  }
+}
+
 // ---------------------------------------------------------------------------
 // Internal state per connected client
 // ---------------------------------------------------------------------------
@@ -119,6 +148,8 @@ interface ClientState {
 
 export class WebSocketServer extends EventEmitter {
   private readonly cfg: Required<WebSocketServerConfig>;
+  private readonly eventBus: SubscriptionEventBus;
+  private readonly unsubscribeFromEventBus: () => void;
 
   /** clientId → state */
   private clients: Map<string, ClientState> = new Map();
@@ -149,16 +180,23 @@ export class WebSocketServer extends EventEmitter {
   private deliveryTimestamps: number[] = [];
   private totalBatchItems = 0;
 
-  constructor(config: WebSocketServerConfig = {}) {
+  constructor(
+    config: WebSocketServerConfig = {},
+    eventBus: SubscriptionEventBus = new InMemorySubscriptionEventBus()
+  ) {
     super();
     this.cfg = {
       maxConnectionsPerUser: 5,
-      batchIntervalMs: 50,
+      batchIntervalMs: 0,
       maxQueueSize: 100,
       heartbeatIntervalMs: 30_000,
       pingTimeoutMs: 10_000,
       ...config,
     };
+    this.eventBus = eventBus;
+    this.unsubscribeFromEventBus = this.eventBus.subscribe((event) =>
+      this._dispatchSubscriptionEvent(event)
+    );
     this._startBatchFlush();
     this._startHeartbeat();
   }
@@ -279,8 +317,15 @@ export class WebSocketServer extends EventEmitter {
    *
    * When `batchIntervalMs > 0`, events accumulate until the next flush.
    * When `batchIntervalMs === 0`, events are sent immediately (legacy).
-   */
+  */
   broadcast(event: SubscriptionEvent): number {
+    const queued = this.eventBus.publish(event);
+
+    this.emit('broadcast', { event, queued, delivered: queued });
+    return queued;
+  }
+
+  private _dispatchSubscriptionEvent(event: SubscriptionEvent): number {
     this.metrics.eventsPublished++;
     let queued = 0;
 
@@ -304,7 +349,7 @@ export class WebSocketServer extends EventEmitter {
       queued++;
     }
 
-    this.emit('broadcast', { event, queued });
+    this.emit('eventQueued', { event, queued });
     return queued;
   }
 
@@ -332,6 +377,7 @@ export class WebSocketServer extends EventEmitter {
   shutdown(): void {
     if (this.flushTimer) clearInterval(this.flushTimer);
     if (this.heartbeatTimer) clearInterval(this.heartbeatTimer);
+    this.unsubscribeFromEventBus();
     this._flushAll();
     // Disconnect all clients
     for (const clientId of [...this.clients.keys()]) {
@@ -427,4 +473,6 @@ export class WebSocketServer extends EventEmitter {
   }
 }
 
-export const webSocketServer = new WebSocketServer();
+export const webSocketServer = new WebSocketServer(
+  process.env.NODE_ENV === 'test' ? { heartbeatIntervalMs: 0 } : {}
+);