Assist_Design/apps/bff/src/infra/realtime/realtime.service.ts

import { Injectable, Logger } from "@nestjs/common";
import type { MessageEvent } from "@nestjs/common";
import { Observable } from "rxjs";
import { getErrorMessage } from "@bff/core/utils/error.util.js";
import type {
  RealtimeEventEnvelope,
  RealtimePubSubMessage,
  RealtimeStreamOptions,
} from "./realtime.types.js";
import { RealtimePubSubService } from "./realtime.pubsub.js";

interface InternalObserver {
  next: (event: MessageEvent) => void;
  complete: () => void;
  error: (error: unknown) => void;
}

/**
 * Production-ready realtime event hub.
 *
 * - Subscriptions are in-memory per instance
 * - Publishes and receives via Redis Pub/Sub for multi-instance delivery
 * - Provides consistent "ready" + "heartbeat" conventions
 */
@Injectable()
export class RealtimeService {
  private readonly logger = new Logger(RealtimeService.name);
  private readonly observersByTopic = new Map<string, Set<InternalObserver>>();

  constructor(private readonly pubsub: RealtimePubSubService) {
    // Fan-in all Redis events and deliver to local subscribers
    this.pubsub.addHandler(msg => this.deliver(msg));
  }

  subscribe(topic: string, options: RealtimeStreamOptions = {}): Observable<MessageEvent> {
    const heartbeatMs = options.heartbeatMs ?? 30000;
    const readyEvent = options.readyEvent === undefined ? "stream.ready" : options.readyEvent;
    const heartbeatEvent =
      options.heartbeatEvent === undefined ? "stream.heartbeat" : options.heartbeatEvent;

    return new Observable<MessageEvent>(subscriber => {
      const wrappedObserver: InternalObserver = {
        next: value => subscriber.next(value),
        complete: () => subscriber.complete(),
        error: error => subscriber.error(error),
      };

      const set = this.observersByTopic.get(topic) ?? new Set<InternalObserver>();
      set.add(wrappedObserver);
      this.observersByTopic.set(topic, set);

      this.logger.debug("Realtime stream connected", { topic, listeners: set.size });

      if (readyEvent) {
        wrappedObserver.next(
          this.buildMessage(readyEvent, {
            timestamp: new Date().toISOString(),
          })
        );
      }

      const heartbeat =
        heartbeatMs > 0 && Boolean(heartbeatEvent)
          ? setInterval(() => {
              wrappedObserver.next(
                this.buildMessage(heartbeatEvent as string, {
                  timestamp: new Date().toISOString(),
                })
              );
            }, heartbeatMs)
          : null;

      return () => {
        if (heartbeat) {
          clearInterval(heartbeat);
        }

        const current = this.observersByTopic.get(topic);
        if (current) {
          current.delete(wrappedObserver);
          if (current.size === 0) {
            this.observersByTopic.delete(topic);
          }
        }

        this.logger.debug("Realtime stream disconnected", {
          topic,
          listeners: current?.size ?? 0,
        });
      };
    });
  }

  publish<TEvent extends string, TData>(topic: string, event: TEvent, data: TData): void {
    const message: RealtimePubSubMessage<TEvent, TData> = { topic, event, data };
    void this.pubsub.publish(message).catch(error => {
      this.logger.warn("Failed to publish realtime event", {
        topic,
        event,
        error: getErrorMessage(error),
      });
    });
  }

  private deliver(message: RealtimePubSubMessage): void {
    const set = this.observersByTopic.get(message.topic);
    if (!set || set.size === 0) {
      return;
    }

    const evt = this.buildMessage(message.event, message.data);
    set.forEach(observer => {
      try {
        observer.next(evt);
      } catch (error) {
        this.logger.warn("Failed to notify realtime listener", {
          topic: message.topic,
          error: getErrorMessage(error),
        });
      }
    });
  }

  private buildMessage<TEvent extends string>(event: TEvent, data: unknown): MessageEvent {
    return {
      // Always serialize explicitly so the browser EventSource receives valid JSON text.
      // This avoids environments where SSE payloads may be coerced to "[object Object]".
      data: JSON.stringify({ event, data } satisfies RealtimeEventEnvelope<TEvent, unknown>),
    } satisfies MessageEvent;
  }
}