MathLab/task-board/backend/sse.js

"use strict";

function safeJson(value) {
  try {
    return JSON.stringify(value);
  } catch {
    return "{}";
  }
}

class SseHub {
  constructor(options = {}) {
    this.clients = new Map();
    this.keepaliveMs = Number.isFinite(options.keepaliveMs) ? Math.max(5000, options.keepaliveMs) : 25000;
  }

  _setFor(conversationId) {
    if (!this.clients.has(conversationId)) {
      this.clients.set(conversationId, new Set());
    }
    return this.clients.get(conversationId);
  }

  subscribe(conversationId, req, res) {
    const key = conversationId || "*";
    const set = this._setFor(key);
    const client = { res, key };
    set.add(client);

    res.writeHead(200, {
      "Content-Type": "text/event-stream; charset=utf-8",
      "Cache-Control": "no-cache, no-transform",
      Connection: "keep-alive",
      "X-Accel-Buffering": "no",
    });
    res.write(": connected\n\n");

    const keepalive = setInterval(() => {
      try {
        res.write(": keepalive\n\n");
      } catch {
        this._unsubscribe(client, keepalive);
      }
    }, this.keepaliveMs);

    const cleanup = () => this._unsubscribe(client, keepalive);
    req.on("close", cleanup);
    req.on("error", cleanup);
    res.on("error", cleanup);
  }

  _unsubscribe(client, timer) {
    if (timer) clearInterval(timer);
    const set = this.clients.get(client.key);
    if (!set) return;
    set.delete(client);
    if (set.size === 0) {
      this.clients.delete(client.key);
    }
  }

  _broadcast(set, event, payload) {
    if (!set || set.size === 0) return;
    const data = safeJson(payload);
    for (const client of set) {
      try {
        client.res.write(`event: ${event}\n`);
        client.res.write(`data: ${data}\n\n`);
      } catch {
        // Ignore write failures; cleanup happens via close handler.
      }
    }
  }

  publishConversation(conversationId, payload) {
    const body = {
      conversationId,
      timestamp: new Date().toISOString(),
      ...payload,
    };
    this._broadcast(this.clients.get(conversationId), "conversation_update", body);
    this._broadcast(this.clients.get("*"), "conversation_update", body);
  }
}

module.exports = {
  SseHub,
};