宜博 / lemonboss

/**
 * DevLoop · sseBroker
 *
 * 每项目一个广播通道，多订阅者。
 * - subscribe(pid, res, opts) 把 HTTP res 注册为订阅者
 * - emit(pid, event, data) 写一行到 stream.log + 广播给所有订阅者
 * - 心跳：每 25s 一条 `:keepalive\n\n` 注释
 * - 断线重连：客户端发 Last-Event-ID header，replay 从 stream.log 读 > 该 id 的行
 *
 * 规格：docs/prd/newfunction/_spec/sdd-control.md · C-11
 *      docs/prd/newfunction/_spec/api-sse.md
 *      docs/prd/newfunction/_spec/sdd-process.md · P-5
 */

const fs = require('fs');
const path = require('path');
const pathResolver = require('./path-resolver');

const HEARTBEAT_MS = 25_000;

// pid → Set<{ res, heartbeatTimer }>
const subscriptions = new Map();

// 单调递增的事件 id（跨进程不严格保证；崩溃重启从 stream.log 读最大值续）
let eventCounter = 0;

function _streamLogPath(pid) {
  return pathResolver.resolveStreamLog(pid);
}

function _initCounterFromLog(pid) {
  // 启动时如果 stream.log 已存在，找最大 id 续号
  const p = _streamLogPath(pid);
  if (!fs.existsSync(p)) return;
  const raw = fs.readFileSync(p, 'utf8');
  const matches = raw.match(/^id: (\d+)$/gm) || [];
  for (const m of matches) {
    const n = parseInt(m.slice(4), 10);
    if (n > eventCounter) eventCounter = n;
  }
}

function _ensureCounterInit(pid) {
  // 全局单调即可；不严格 per-pid（避免每次启动都全扫）
  if (eventCounter === 0) _initCounterFromLog(pid);
}

/**
 * @param {string} pid
 * @param {http.ServerResponse} res - 调用方应已 writeHead(200, SSE headers)
 * @param {object} opts - { replayFromOffset?: string | number (Last-Event-ID) }
 */
function subscribe(pid, res, opts = {}) {
  _ensureCounterInit(pid);

  // Replay：从 stream.log 读 > replayFromOffset 的行
  if (opts.replayFromOffset != null) {
    const fromId = parseInt(opts.replayFromOffset, 10) || 0;
    _replayTo(res, pid, fromId);
  }

  // 心跳
  const heartbeat = setInterval(() => {
    try { res.write(':keepalive\n\n'); } catch { /* ignore broken pipe */ }
  }, HEARTBEAT_MS);

  // 加订阅
  if (!subscriptions.has(pid)) subscriptions.set(pid, new Set());
  const entry = { res, heartbeatTimer: heartbeat };
  subscriptions.get(pid).add(entry);

  // 客户端断开
  res.on('close', () => {
    clearInterval(heartbeat);
    subscriptions.get(pid)?.delete(entry);
    if (subscriptions.get(pid)?.size === 0) subscriptions.delete(pid);
  });
}

function _replayTo(res, pid, fromId) {
  const p = _streamLogPath(pid);
  if (!fs.existsSync(p)) return;

  const raw = fs.readFileSync(p, 'utf8');
  // stream.log 格式为 SSE 帧：`id: N\nevent: X\ndata: {...}\n\n`
  const blocks = raw.split('\n\n').filter(Boolean);
  for (const block of blocks) {
    const idMatch = block.match(/^id: (\d+)/m);
    if (!idMatch) continue;
    const id = parseInt(idMatch[1], 10);
    if (id <= fromId) continue;
    try { res.write(block + '\n\n'); } catch { return; }
  }
}

/**
 * 广播一条事件 + 落盘 stream.log。
 * @param {string} pid
 * @param {string} event - 事件名，如 "unit.state" / "bug.opened"
 * @param {any} data
 */
function emit(pid, event, data) {
  _ensureCounterInit(pid);
  const id = ++eventCounter;
  const payload = JSON.stringify(data == null ? {} : data);
  const line = `id: ${id}\nevent: ${event}\ndata: ${payload}\n\n`;

  // 1. 落盘
  const p = _streamLogPath(pid);
  fs.mkdirSync(path.dirname(p), { recursive: true });
  fs.appendFileSync(p, line);

  // 2. 广播
  const subs = subscriptions.get(pid);
  if (subs) {
    for (const entry of subs) {
      try { entry.res.write(line); } catch { /* ignore broken pipe */ }
    }
  }

  return { id, event };
}

/** 项目结束：发 close + 关所有订阅。 */
function close(pid) {
  const subs = subscriptions.get(pid);
  if (!subs) return;
  for (const entry of subs) {
    try { entry.res.write('event: close\ndata: {}\n\n'); } catch { /* ignore */ }
    try { entry.res.end(); } catch { /* ignore */ }
    clearInterval(entry.heartbeatTimer);
  }
  subscriptions.delete(pid);
}

/** 测试用 */
function _resetForTesting() {
  for (const subs of subscriptions.values()) {
    for (const entry of subs) clearInterval(entry.heartbeatTimer);
  }
  subscriptions.clear();
  eventCounter = 0;
}

function _activeSubscribers(pid) {
  return subscriptions.get(pid)?.size || 0;
}

module.exports = {
  subscribe,
  emit,
  close,
  HEARTBEAT_MS,
  _resetForTesting,
  _activeSubscribers,
};