boocode/apps/server/src/services/broker.ts

export type Frame = Record<string, unknown> & { type: string };
export type Listener = (frame: Frame) => void;

export interface Broker {
  publish(sessionId: string, frame: Frame): void;
  subscribe(sessionId: string, listener: Listener): () => void;
  publishUser(user: string, frame: Frame): void;
  subscribeUser(user: string, listener: Listener): () => void;
}

export function createBroker(): Broker {
  const topics = new Map<string, Set<Listener>>();
  const userTopics = new Map<string, Set<Listener>>();

  function publishTo(map: Map<string, Set<Listener>>, key: string, frame: Frame): void {
    const set = map.get(key);
    if (!set) return;
    for (const listener of set) {
      try {
        listener(frame);
      } catch {
        // ignore listener errors so one bad subscriber doesn't break the rest
      }
    }
  }

  function subscribeTo(map: Map<string, Set<Listener>>, key: string, listener: Listener): () => void {
    let set = map.get(key);
    if (!set) {
      set = new Set();
      map.set(key, set);
    }
    set.add(listener);
    return () => {
      const s = map.get(key);
      if (!s) return;
      s.delete(listener);
      if (s.size === 0) map.delete(key);
    };
  }

  return {
    publish(sessionId, frame) {
      publishTo(topics, sessionId, frame);
    },
    subscribe(sessionId, listener) {
      return subscribeTo(topics, sessionId, listener);
    },
    publishUser(user, frame) {
      publishTo(userTopics, user, frame);
    },
    subscribeUser(user, listener) {
      return subscribeTo(userTopics, user, listener);
    },
  };
}