agentic-dev/packages/core-realtime/src/in-memory-realtime-broadcaster.ts

import type { z } from "zod";
import type { IRealtimeBroadcaster } from "./realtime-broadcaster.interface";
import type { RealtimeChannelDescriptor } from "./realtime-channel";

type Listener<T> = (payload: T) => Promise<void> | void;

export class InMemoryRealtimeBroadcaster implements IRealtimeBroadcaster {
  private readonly listeners = new Map<string, Listener<unknown>[]>();

  async broadcast<T>(
    descriptor: RealtimeChannelDescriptor<string, z.ZodType<T>>,
    payload: T,
  ): Promise<void> {
    descriptor.schema.parse(payload);
    const arr = this.listeners.get(descriptor.name) ?? [];
    for (const l of arr) await l(payload);
  }

  // Test-friendly: lets unit tests subscribe directly without a Socket.IO server.
  subscribe<T>(
    descriptor: RealtimeChannelDescriptor<string, z.ZodType<T>>,
    listener: Listener<T>,
  ): void {
    const arr = this.listeners.get(descriptor.name) ?? [];
    arr.push(listener as Listener<unknown>);
    this.listeners.set(descriptor.name, arr);
  }
}