宜博 / lemonboss

/**
 * 图执行引擎 · graphRunSse —— 按 run_id 的 SSE 广播 + 审计 timeline + 状态转换三合一
 *
 * 形态照抄 server/services/devloop/sse-broker.js（每通道一个广播集合、emit 同时落盘 + 广播、
 * 25s 心跳、Last-Event-ID 断线重放），但补上 DevLoop 的那个缺口：
 *
 *   DevLoop 里 timeline-logger 的审计事件和 sse-broker 的实时推送是两条独立的轨，
 *   loop-controller 往 timeline 追加一行不会顺带 emit SSE。本引擎明确要求
 *   **状态转换、timeline 写入、SSE 广播三者在同一次原子操作里一起发生**（设计文档 §三.3 末段）。
 *   所以对外只暴露组合函数 transitionNode / transitionRun / emitTransition，
 *   调用方（graph-engine.js）调一次，三件事一起做完；不提供「只改状态不发事件」的路径。
 *
 * 通道粒度：**按 root run 建通道、按 run_id 过滤投递**。
 *   一棵 run 树（root → org node 子 run → 嵌套子图）共用一份 stream.log 和一条 id 序列，
 *   这样订阅 root 的控制台能看到整棵树的事件，订阅某个子 run 的下钻视图只收自己那一份，
 *   而两者的 Last-Event-ID 重放走同一条序列、不会错位。
 *
 * 落盘位置（设计文档 §四.A.5 的 run 目录）：
 *   data/companies/{companyId}/runs/{rootRunId}/stream.log      SSE 帧，供断线重放
 *   data/companies/{companyId}/runs/{rootRunId}/timeline.ndjson 审计事件，append-only
 *
 * 设计：docs/todo/system-design/27-p3-p4-work-graph-console.md §三.3 / §四.A.5 / §五 T5
 */

const fs = require('fs');
const path = require('path');

const workGraphDb = require('./work-graph-db');

const HEARTBEAT_MS = 25_000;
const SUMMARY_MAX = 280; // 与 devloop/timeline-logger 一致

const ROOT_DATA_DIR = path.resolve(__dirname, '../../data');
const RUNS_ROOT = process.env.GRAPH_RUNS_DIR || null; // 覆盖时直接以它为 {rootRunId} 的父目录

// rootRunId → Set<{ res, runId, heartbeatTimer }>；runId 是订阅者关心的那一层
const subscriptions = new Map();
// rootRunId → 已用到的最大事件 id（每棵 run 树一条序列）
const counters = new Map();
// runId → { rootRunId, companyId }，避免每条事件都查一次库
const runMeta = new Map();

// ── 路径 ────────────────────────────────────────────────────────────

/**
 * run 所属的 { rootRunId, companyId }；查不到库时退化成「自己就是 root」。
 *
 * ⚠️ **只缓存查得到的权威结果，退化值一律不落缓存**（R4-F2）。
 * 缓存退化值会把一个错误状态永久固化：此后这个 run 的所有事件都写到
 * data/companies/_unknown/runs/{runId}/ 并广播到错误的 root 频道，
 * 订阅真 root 的控制台从此静默收不到它——而 graph_runs 那一行明明已经建好了。
 * 省下的那次 SELECT 换来一个没有任何失效路径的错误缓存，不划算。
 */
function resolveMeta(runId) {
  const hit = runMeta.get(runId);
  if (hit) return hit;
  try {
    const run = workGraphDb.getRun(runId);
    if (run) {
      const meta = { rootRunId: run.root_run_id || runId, companyId: run.company_id };
      runMeta.set(runId, meta);
      return meta;
    }
  } catch {
    // 库还没建好（如纯单测场景）时不阻塞广播，按退化值走
  }
  // 退化值：每次重算，下一次调用会重新试一遍库，行建好之后自动恢复
  return { rootRunId: runId, companyId: '_unknown' };
}

/** data/companies/{cid}/runs/{rootRunId}/ */
function runDir(runId) {
  const { rootRunId, companyId } = resolveMeta(runId);
  if (RUNS_ROOT) return path.join(RUNS_ROOT, rootRunId);
  return path.join(ROOT_DATA_DIR, 'companies', companyId, 'runs', rootRunId);
}

function streamLogPath(runId) {
  return path.join(runDir(runId), 'stream.log');
}

function timelinePath(runId) {
  return path.join(runDir(runId), 'timeline.ndjson');
}

// ── 事件 id 序列 ─────────────────────────────────────────────────────

function nextEventId(rootRunId, logPath) {
  if (!counters.has(rootRunId)) {
    // 进程重启后从 stream.log 现有最大 id 续号，避免客户端 Last-Event-ID 判断错乱
    let max = 0;
    if (fs.existsSync(logPath)) {
      const matches = fs.readFileSync(logPath, 'utf8').match(/^id: (\d+)$/gm) || [];
      for (const m of matches) {
        const n = parseInt(m.slice(4), 10);
        if (n > max) max = n;
      }
    }
    counters.set(rootRunId, max);
  }
  const id = counters.get(rootRunId) + 1;
  counters.set(rootRunId, id);
  return id;
}

// ── 订阅 ────────────────────────────────────────────────────────────

/**
 * 把 HTTP res 注册为 runId 这一层的订阅者。调用方应已 writeHead(200, SSE headers)。
 * @param {string} runId - 订阅哪个 run；传 root run id 则收整棵 run 树的事件
 * @param {import('http').ServerResponse} res
 * @param {{ replayFromOffset?: string|number }} opts - replayFromOffset 取自 Last-Event-ID
 */
function subscribe(runId, res, opts = {}) {
  const { rootRunId } = resolveMeta(runId);

  if (opts.replayFromOffset != null) {
    replayTo(res, runId, parseInt(opts.replayFromOffset, 10) || 0);
  }

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

  if (!subscriptions.has(rootRunId)) subscriptions.set(rootRunId, new Set());
  const entry = { res, runId, heartbeatTimer };
  subscriptions.get(rootRunId).add(entry);

  res.on('close', () => {
    clearInterval(heartbeatTimer);
    subscriptions.get(rootRunId)?.delete(entry);
    if (subscriptions.get(rootRunId)?.size === 0) subscriptions.delete(rootRunId);
  });

  return entry;
}

/** 订阅 root 的看整棵树，订阅子 run 的只看自己那一层 */
function wants(entry, eventRunId) {
  const { rootRunId } = resolveMeta(entry.runId);
  return entry.runId === rootRunId || entry.runId === eventRunId;
}

function replayTo(res, runId, fromId) {
  const p = streamLogPath(runId);
  if (!fs.existsSync(p)) return;
  const { rootRunId } = resolveMeta(runId);

  // stream.log 每帧形如 `id: N\nevent: X\ndata: {...}\n\n`
  for (const block of fs.readFileSync(p, 'utf8').split('\n\n').filter(Boolean)) {
    const idMatch = block.match(/^id: (\d+)/m);
    if (!idMatch || parseInt(idMatch[1], 10) <= fromId) continue;
    // 同一份 log 混着整棵树的事件，重放时按订阅层级过滤
    if (runId !== rootRunId) {
      const dataMatch = block.match(/^data: (.*)$/m);
      let evRunId = null;
      try { evRunId = JSON.parse(dataMatch?.[1] || '{}').run_id; } catch { /* 坏行跳过过滤 */ }
      if (evRunId && evRunId !== runId) continue;
    }
    try { res.write(block + '\n\n'); } catch { return; }
  }
}

// ── 落盘 + 广播 ──────────────────────────────────────────────────────

function truncate(s, n) {
  const str = typeof s === 'string' ? s : String(s);
  return str.length <= n ? str : str.slice(0, n - 1) + '…';
}

/**
 * 广播一条事件 + 落盘 stream.log（不写 timeline；审计轨见 appendTimeline / emitTransition）。
 * @param {string} runId - 事件属于哪个 run
 * @param {string} event - 事件名，如 'node.status' / 'run.status' / 'loop.registered'
 * @param {object} data  - 会自动补 run_id / root_run_id / ts 三个字段
 */
function emit(runId, event, data) {
  const { rootRunId } = resolveMeta(runId);
  const p = streamLogPath(runId);
  const id = nextEventId(rootRunId, p);
  const payload = {
    ...(data == null ? {} : data),
    run_id: runId,
    root_run_id: rootRunId,
    ts: new Date().toISOString(),
  };
  const line = `id: ${id}\nevent: ${event}\ndata: ${JSON.stringify(payload)}\n\n`;

  fs.mkdirSync(path.dirname(p), { recursive: true });
  fs.appendFileSync(p, line);

  for (const entry of subscriptions.get(rootRunId) || []) {
    if (!wants(entry, runId)) continue;
    try { entry.res.write(line); } catch { /* ignore broken pipe */ }
  }

  return { id, event, data: payload };
}

/**
 * 只写审计轨（append-only NDJSON）。正常路径请用 emitTransition —— 它会连 SSE 一起发。
 * @param {string} runId
 * @param {{ source?: string, kind: string, summary: string, ref?: string, payload?: object }} ev
 */
function appendTimeline(runId, ev) {
  if (!ev || !ev.kind || !ev.summary) {
    throw new Error('graphRunSse.appendTimeline: kind/summary required');
  }
  const { rootRunId } = resolveMeta(runId);
  const event = {
    ts: new Date().toISOString(),
    run_id: runId,
    root_run_id: rootRunId,
    source: ev.source || 'graph-engine',
    kind: ev.kind,
    ref: ev.ref || null,
    summary: truncate(ev.summary, SUMMARY_MAX),
  };
  if (ev.payload !== undefined) event.payload = ev.payload;

  const p = timelinePath(runId);
  fs.mkdirSync(path.dirname(p), { recursive: true });
  fs.appendFileSync(p, JSON.stringify(event) + '\n');
  return event;
}

// ── 三合一：状态转换 + timeline + SSE ────────────────────────────────

/**
 * 原子三合一。整个函数体是同步的、中间没有 await，所以在 Node 单线程里
 * 「状态改了但事件没发」只可能因为进程崩溃，不会因为并发交错。
 *
 * apply() 必须是条件更新（`UPDATE ... WHERE status = @from`）并返回受影响行数：
 * 返回 0 表示这次转换已被别人抢先做掉（§四.A.6 幂等语义），此时**不写 timeline、不广播**，
 * 返回 null 让调用方据此判断「不是我推进的」。
 *
 * @param {string} runId
 * @param {object} ev
 * @param {string} ev.event   - SSE 事件名
 * @param {string} [ev.kind]  - timeline kind，默认取 ev.event
 * @param {string} ev.summary - 人类可读一句话
 * @param {object} [ev.data]  - 同时进 SSE data 和 timeline payload
 * @param {string} [ev.ref]   - timeline ref（如 node run id / edge id）
 * @param {string} [ev.source]
 * @param {() => number} apply - 条件更新，返回受影响行数
 * @returns {{ changes: number, event: object, timeline: object } | null}
 */
function emitTransition(runId, ev, apply) {
  const changes = typeof apply === 'function' ? apply() : 1;
  if (!changes) return null; // 幂等 no-op：状态没变，就不该有事件

  const timeline = appendTimeline(runId, {
    source: ev.source,
    kind: ev.kind || ev.event,
    ref: ev.ref,
    summary: ev.summary,
    payload: ev.data,
  });
  const sent = emit(runId, ev.event, ev.data);
  return { changes, event: sent, timeline };
}

/**
 * 节点状态转换（graph_run_nodes）+ timeline + SSE，一次调用做完。
 * @param {string} runId
 * @param {string} nodeRunId
 * @param {string|null} from - null = 用当前状态做期望值
 * @param {string} to
 * @param {object} [patch] - 透传给 workGraphDb.updateNodeStatus 的补丁列
 * @param {object} [extra] - { nodeName?, failReason?, attempt?, summary?, ... } 附在事件里
 */
function transitionNode(runId, nodeRunId, from, to, patch = {}, extra = {}) {
  const data = {
    node_run_id: nodeRunId,
    from: from ?? null,
    to,
    ...patch,
    ...extra,
  };
  return emitTransition(
    runId,
    {
      event: 'node.status',
      ref: nodeRunId,
      summary: extra.summary || `node ${extra.nodeName || nodeRunId}: ${from ?? '*'} → ${to}`,
      data,
    },
    () => workGraphDb.updateNodeStatus(nodeRunId, from, to, patch),
  );
}

/**
 * run 状态转换（graph_runs）+ timeline + SSE，一次调用做完。
 * 语义与 transitionNode 完全一致。
 */
function transitionRun(runId, from, to, patch = {}, extra = {}) {
  const data = { from: from ?? null, to, ...patch, ...extra };
  return emitTransition(
    runId,
    {
      event: 'run.status',
      ref: runId,
      summary: extra.summary || `run ${runId}: ${from ?? '*'} → ${to}`,
      data,
    },
    () => workGraphDb.updateRunStatus(runId, from, to, patch),
  );
}

/**
 * 非状态机类事件的三合一入口（回退备案 / 预算扣减 / agent 调用起止 …）：
 * 没有条件更新要做时 apply 传 null，事件照样同时进 timeline 和 SSE。
 */
function emitEvent(runId, event, summary, data = {}, opts = {}) {
  return emitTransition(runId, { event, summary, data, ref: opts.ref, kind: opts.kind, source: opts.source }, null);
}

// ── 收尾 ────────────────────────────────────────────────────────────

/**
 * run 跑完：给**订阅这一层**的连接发 close 并关掉它们。
 * 子 run 结束不会掐断订阅 root 的控制台连接（它还要继续看整棵树）；
 * 只有 close(rootRunId) 才会清空整个通道。
 */
function close(runId) {
  const { rootRunId } = resolveMeta(runId);
  const subs = subscriptions.get(rootRunId);
  if (!subs) return;
  for (const entry of subs) {
    if (runId !== rootRunId && entry.runId !== runId) continue;
    try { entry.res.write(`event: close\ndata: ${JSON.stringify({ run_id: runId })}\n\n`); } catch { /* ignore */ }
    try { entry.res.end(); } catch { /* ignore */ }
    clearInterval(entry.heartbeatTimer);
    subs.delete(entry);
  }
  if (subs.size === 0) subscriptions.delete(rootRunId);
}

function _resetForTesting() {
  for (const subs of subscriptions.values()) {
    for (const entry of subs) clearInterval(entry.heartbeatTimer);
  }
  subscriptions.clear();
  counters.clear();
  runMeta.clear();
}

function _activeSubscribers(runId) {
  return subscriptions.get(resolveMeta(runId).rootRunId)?.size || 0;
}

module.exports = {
  // 订阅 / 广播
  subscribe,
  emit,
  close,

  // 审计
  appendTimeline,

  // 三合一（graph-engine 应只用这三个）
  emitTransition,
  transitionNode,
  transitionRun,
  emitEvent,

  // 路径
  runDir,
  streamLogPath,
  timelinePath,

  HEARTBEAT_MS,
  _resetForTesting,
  _activeSubscribers,
};