JX-BPOS/apps/api/src/modules/trace/internal/usecase/trace-service.ts

/**
 * trace 用例 —— **手动打点的全链观测**(编码规范 §5.3;`@mastra` 只准 adapter/app §6.2)。
 *
 * 我们的执行不经 Mastra 图(自持流程运行时 + 自持范式引擎),所以 Mastra 自带的
 * 自动 tracing 不会产出任何 span。这里把它的**观测组件**当作产 span 的工具用:
 * `Observability` 实例出 span(标准形状:traceId / parentSpanId / input / output /
 * errorInfo / 起止时刻),我们在执行链路的关键点手动开合。
 *
 * 落库不用 Mastra 的 storage —— exporter 那条路是**批量异步**的(缓冲 + 定时冲刷),
 * 而测试台要的是「点一下就看得见」。所以 span 一开就落一行、一收就补一行,
 * 两笔写都在 usecase 里同步做:**观测面不该比被观测的事晚**。
 *
 * Mastra 侧只用来生成 id 与保证 span 形状 —— 换掉它只需改 `spanFactory`。
 */
import type { SpanHandle, SpanKind, TraceService, TraceSpan } from '../../contract.ts';
import type { Clock, TraceStore } from '../../port.ts';

/** span 工厂:出 id 与父子关系 —— 由 app 层用 Mastra 的 Observability 供给 */
export type SpanFactory = {
  root(name: string): { spanId: string; traceId: string };
  child(parent: { traceId: string; spanId: string }, name: string): { spanId: string; traceId: string };
};

export type TraceServiceDeps = {
  readonly store: TraceStore;
  readonly clock: Clock;
  readonly spans: SpanFactory;
};

export function createTraceService(deps: TraceServiceDeps): TraceService {
  /** 落一行「开」;失败不掀翻被观测的事 —— 观测坏了不该让业务停摆 */
  const open = async (span: TraceSpan): Promise<void> => {
    try {
      await deps.store.insert(span);
    } catch {
      /* 观测尽力而为:trace 写不进去也不能让流程跑不动 */
    }
  };

  const close = async (
    id: string,
    patch: { output?: Record<string, unknown>; error?: { message: string }; attributes?: Record<string, unknown> },
  ): Promise<void> => {
    try {
      await deps.store.finish(id, { ...patch, endedAt: deps.clock.now().toISOString() });
    } catch {
      /* 同上 */
    }
  };

  const handleOf = (
    ids: { spanId: string; traceId: string },
    scope: { workflowId?: string; nodeId?: string; taskId?: string },
  ): SpanHandle => ({
    id: ids.spanId,
    traceId: ids.traceId,

    child(input) {
      const born = deps.spans.child({ traceId: ids.traceId, spanId: ids.spanId }, input.name);
      // 归属向下继承:子 span 没说自己属于谁,就跟父亲同一条流、同一个节点
      const childScope = {
        ...scope,
        ...(input.nodeId === undefined ? {} : { nodeId: input.nodeId }),
        ...(input.taskId === undefined ? {} : { taskId: input.taskId }),
      };
      void open({
        id: born.spanId,
        traceId: born.traceId,
        parentId: ids.spanId,
        name: input.name,
        kind: input.kind,
        ...childScope,
        ...(input.input === undefined ? {} : { input: input.input }),
        ...(input.attributes === undefined ? {} : { attributes: input.attributes }),
        startedAt: deps.clock.now().toISOString(),
      });
      return handleOf(born, childScope);
    },

    async end(out) {
      await close(ids.spanId, { ...(out?.output === undefined ? {} : { output: out.output }), ...(out?.attributes === undefined ? {} : { attributes: out.attributes }) });
    },

    async fail(error) {
      await close(ids.spanId, { error: { message: error instanceof Error ? error.message : String(error) } });
    },
  });

  /*
   * 作用域索引:workflowId → 工作流 span,taskId → 节点 span。
   * 打点散在装配层十几处,让每处自己拎 handle 传来传去,一处漏传整棵树就断了 ——
   * 所以由服务记着「这一单现在挂在哪儿」,调用方只报 id。
   * 进程内存活即可:trace 已经落库了,这张表只是为了拼父子。
   */
  const workflowSpans = new Map<string, SpanHandle>();
  const nodeSpans = new Map<string, SpanHandle>();

  const service: TraceService = {
    beginWorkflow({ workflowId, subject, input }) {
      const span = service.start({
        name: `流程:${subject}`,
        kind: 'workflow',
        workflowId,
        ...(input === undefined ? {} : { input }),
      });
      workflowSpans.set(workflowId, span);
    },

    beginNode({ workflowId, nodeId, label, taskId, approvalId, assigneeId, input }) {
      // 派给谁、卡了哪张审批单 —— 开 span 那一刻就写清「这一步在谁手上」
      const shape = {
        name: `${nodeId} ${label}`,
        kind: 'node' as const,
        nodeId,
        ...(taskId === undefined ? {} : { taskId }),
        ...(input === undefined ? {} : { input }),
        attributes: {
          ...(assigneeId === undefined ? {} : { assigneeId }),
          ...(approvalId === undefined ? {} : { approvalId }),
        },
      };
      // 工作流 span 不在手上(进程重启过)就另起一棵 —— 观测尽力而为,不让业务停摆
      const parent = workflowSpans.get(workflowId);
      const span = parent === undefined ? service.start({ ...shape, workflowId }) : parent.child(shape);
      nodeSpans.set(taskId ?? `${workflowId}:${nodeId}`, span);
    },

    beginAction({ taskId, name, kind, input }) {
      const parent = nodeSpans.get(taskId);
      if (parent === undefined) return undefined;
      return parent.child({ name, kind, taskId, ...(input === undefined ? {} : { input }) });
    },

    async endNode({ taskId, workflowId, nodeId, output }) {
      const key = taskId ?? `${workflowId}:${nodeId}`;
      const span = nodeSpans.get(key);
      if (span === undefined) return;
      nodeSpans.delete(key);
      await span.end(output === undefined ? undefined : { output });
    },

    async endWorkflow({ workflowId, state }) {
      const span = workflowSpans.get(workflowId);
      if (span === undefined) return;
      workflowSpans.delete(workflowId);
      await span.end({ output: { state } });
    },

    start(input) {
      const born =
        input.parent === undefined ? deps.spans.root(input.name) : deps.spans.child(input.parent, input.name);
      const scope = {
        ...(input.workflowId === undefined ? {} : { workflowId: input.workflowId }),
        ...(input.nodeId === undefined ? {} : { nodeId: input.nodeId }),
        ...(input.taskId === undefined ? {} : { taskId: input.taskId }),
      };
      void open({
        id: born.spanId,
        traceId: born.traceId,
        ...(input.parent === undefined ? {} : { parentId: input.parent.spanId }),
        name: input.name,
        kind: input.kind as SpanKind,
        ...scope,
        ...(input.input === undefined ? {} : { input: input.input }),
        ...(input.attributes === undefined ? {} : { attributes: input.attributes }),
        startedAt: deps.clock.now().toISOString(),
      });
      return handleOf(born, scope);
    },

    async ofWorkflow(workflowId) {
      return deps.store.listByWorkflow(workflowId);
    },

    async ofTask(taskId) {
      return deps.store.listByTask(taskId);
    },

    async flush() {
      /* 同步落库,没有缓冲要冲 —— 留这个口子是给将来换成批量 exporter 用的 */
    },
  };

  return service;
}