Spanglish/backend/src/lib/stores/pubsub.ts

// Pub/Sub abstraction with two implementations:
// - memory: in-process EventEmitter (single instance only)
// - redis: PUBLISH / SUBSCRIBE so a message published on one instance reaches
//   subscribers on every instance
//
// Messages are JSON-serialized. Selection happens once based on REDIS_URL.

import { EventEmitter } from 'events';
import { getRedis, getSubscriber, isRedisEnabled } from '../redis.js';

export type PubSubHandler = (message: any) => void;

export interface PubSub {
  readonly backend: 'memory' | 'redis';
  publish(channel: string, message: any): Promise<void>;
  // Returns an unsubscribe function for this specific handler.
  subscribe(channel: string, handler: PubSubHandler): Promise<() => void>;
}

// ==================== Memory implementation ====================

class MemoryPubSub implements PubSub {
  readonly backend = 'memory' as const;
  private emitter = new EventEmitter();

  constructor() {
    // SSE fan-out can attach many listeners to the same channel; lift the cap.
    this.emitter.setMaxListeners(0);
  }

  async publish(channel: string, message: any): Promise<void> {
    this.emitter.emit(channel, message);
  }

  async subscribe(channel: string, handler: PubSubHandler): Promise<() => void> {
    this.emitter.on(channel, handler);
    return () => this.emitter.off(channel, handler);
  }
}

// ==================== Redis implementation ====================

class RedisPubSub implements PubSub {
  readonly backend = 'redis' as const;
  // Per-channel handler sets so a single Redis subscription fans out locally.
  private handlers = new Map<string, Set<PubSubHandler>>();
  private wired = false;

  private ensureWired(): void {
    if (this.wired) return;
    const sub = getSubscriber();
    if (!sub) return;
    this.wired = true;
    sub.on('message', (channel: string, payload: string) => {
      const set = this.handlers.get(channel);
      if (!set || set.size === 0) return;
      let parsed: any = payload;
      try {
        parsed = JSON.parse(payload);
      } catch {
        // Leave as raw string if it was not JSON.
      }
      for (const handler of set) {
        try {
          handler(parsed);
        } catch (err: any) {
          console.error('[pubsub] handler error:', err?.message || err);
        }
      }
    });
  }

  async publish(channel: string, message: any): Promise<void> {
    const redis = getRedis();
    if (!redis) return;
    try {
      await redis.publish(channel, JSON.stringify(message));
    } catch (err: any) {
      console.error('[pubsub] publish error:', err?.message || err);
    }
  }

  async subscribe(channel: string, handler: PubSubHandler): Promise<() => void> {
    this.ensureWired();
    const sub = getSubscriber();
    if (!sub) return () => undefined;

    let set = this.handlers.get(channel);
    if (!set) {
      set = new Set();
      this.handlers.set(channel, set);
      try {
        await sub.subscribe(channel);
      } catch (err: any) {
        console.error('[pubsub] subscribe error:', err?.message || err);
      }
    }
    set.add(handler);

    return () => {
      const current = this.handlers.get(channel);
      if (!current) return;
      current.delete(handler);
      if (current.size === 0) {
        this.handlers.delete(channel);
        sub.unsubscribe(channel).catch(() => undefined);
      }
    };
  }
}

// ==================== Selection ====================

let instance: PubSub | null = null;

export function getPubSub(): PubSub {
  if (!instance) {
    instance = isRedisEnabled() ? new RedisPubSub() : new MemoryPubSub();
  }
  return instance;
}