BPOS-MVP/apps/api/src/modules/process/internal/usecase/process-service.ts

/**
 * process 用例 —— 流程引擎:确定性流转,**零 LLM、零临场**(编码规范 §5.3;语义出处 OS 03 / 04)。
 *
 * 起单 · 调度节点 · 审批点卡工位 · 网关分合 · 每次流转记入事件。
 */
import type { ProcessNode, ProcessPackage, StaffId, TerminalState, WorkflowState } from '@contracts';
import type { ProcessService, StartWorkflowInput } from '../../contract.ts';
import { CallInterfaceViolationError, NotPublishableError, PositionVacantError } from '../../contract.ts';
import type {
  ActivityDirectory,
  ApprovalGateway,
  Clock,
  FlowEventPublisher,
  IdGenerator,
  OrgDirectory,
  ProcessStore,
  ReceiptNotifier,
  SchemaValidator,
  Sfeel,
  TaskGateway,
  WorkflowRecord,
} from '../../port.ts';
import { checkClosure } from '../domain/closure.ts';
import { aggregateWorkflowState, isDigested, nextInstanceNumber, resolveOwner } from '../domain/owner-resolution.ts';
import type { NodeInstance } from '../domain/owner-resolution.ts';
import { assemble, dispatchOf, joinReady, rejectRouteOf, selectEdges, startNodeOf } from '../domain/routing.ts';

export type ProcessServiceDeps = {
  readonly store: ProcessStore;
  readonly activities: ActivityDirectory;
  readonly org: OrgDirectory;
  readonly tasks: TaskGateway;
  readonly approvals: ApprovalGateway;
  readonly sfeel: Sfeel;
  readonly schemas: SchemaValidator;
  readonly events: FlowEventPublisher;
  readonly receipts: ReceiptNotifier;
  readonly clock: Clock;
  readonly ids: IdGenerator;
};

export function createProcessService(deps: ProcessServiceDeps): ProcessService {
  async function requirePackage(id: string, version?: number): Promise<ProcessPackage> {
    const pkg = await deps.store.getPackage(id, version);
    if (pkg === undefined) throw new Error(`流程包不存在:${id}`);
    return pkg;
  }

  /**
   * 起单取版:未指定版本时取**最新已发布** —— 草稿不入选。
   * 若按「最新版本」取,治理台一存草稿,起单立刻被没发布的定义顶住;
   * 草稿是编排中的半成品,运行时永远不该看见它。
   */
  async function latestPublished(id: string): Promise<ProcessPackage> {
    const all = await deps.store.listPackages();
    const hit = all
      .filter((pkg) => pkg.id === id && pkg.state === 'published')
      .sort((a, b) => b.version - a.version)[0];
    if (hit === undefined) throw new Error(`流程包无已发布版本,不得起单:${id}`);
    return hit;
  }

  async function requireWorkflow(id: string): Promise<WorkflowRecord> {
    const record = await deps.store.getWorkflow(id);
    if (record === undefined) throw new Error(`工作流不存在:${id}`);
    return record;
  }

  function scopeOf(record: WorkflowRecord) {
    return { ports: record.ports, vars: record.vars, initiation: record.initiation };
  }

  /** 起一个节点:活动 → 起单;审批点 → 卡工位;终点 → 收束 */
  async function enter(
    record: WorkflowRecord,
    pkg: ProcessPackage,
    node: ProcessNode,
    payload: Record<string, unknown>,
  ): Promise<void> {
    const dispatch = dispatchOf(node, payload);

    if (dispatch === undefined) {
      // 网关本身不是责任单元:直接继续流转
      await advanceFrom(record, pkg, node.id, payload);
      return;
    }

    const instances = await deps.store.listInstances(record.id);
    const instance = nextInstanceNumber(instances, node.id);

    if (dispatch.kind === 'finish') {
      await deps.store.insertInstance(record.id, {
        nodeId: node.id,
        instance,
        running: false,
        waiting: false,
        activated: true,
        terminal: 'released',
      });
      const finished = await deps.store.updateWorkflow(record.id, {
        state: 'completed',
        finishedAt: deps.clock.now().toISOString(),
        ports: { ...record.ports, __output: payload },
      });
      await deps.events.publish({ workflowId: record.id, kind: 'workflow-completed', detail: payload });
      // 凡单必有来源 —— 登记人收消息回执
      if (record.source.by !== undefined) {
        await deps.receipts.notify({
          to: record.source.by as StaffId,
          workflowId: record.id,
          subject: record.subject,
          state: finished.state,
        });
      }
      return;
    }

    // 负责人 / 审批人:编排声明岗位,起单时确定性解析;结果与依据留痕
    const spec = dispatch.kind === 'start-task' ? dispatch.node.owner : dispatch.node.approver;
    const candidates = await deps.org.candidates(spec.position);
    /*
     * 绑定先于策略:工作组说了这个岗位是谁,策略就无从发问(ADR-011)。
     * 在岗与否在这里核 —— 绑定是**跨岗位**指派,拿该岗位的候选集核不了它。
     */
    const boundTo = record.workGroup?.[spec.position];
    const bound = boundTo !== undefined && (await deps.org.isActive(boundTo)) ? boundTo : undefined;
    const resolution = resolveOwner(spec, { candidates, ...(bound === undefined ? {} : { bound }) });
    if (resolution.kind === 'vacant') throw new PositionVacantError(spec.position);

    await deps.events.publish({
      workflowId: record.id,
      kind: 'owner-resolved',
      detail: { nodeId: node.id, position: spec.position, staffId: resolution.staffId, by: resolution.by, rationale: resolution.rationale },
    });

    if (dispatch.kind === 'start-task') {
      const task = await deps.tasks.start({
        activityId: dispatch.node.activityId,
        ownerId: resolution.staffId as StaffId,
        title: dispatch.node.label,
        workflowId: record.id,
        nodeId: node.id,
        data: payload,
        ...(dispatch.node.message === undefined ? {} : { message: dispatch.node.message }),
      });
      await deps.store.insertInstance(record.id, {
        nodeId: node.id,
        instance,
        running: true,
        waiting: false,
        activated: true,
        taskId: task.id,
      });
      await deps.events.publish({ workflowId: record.id, kind: 'task-started', detail: { nodeId: node.id, taskId: task.id } });
      return;
    }

    const approval = await deps.approvals.raise({
      workflowId: record.id,
      nodeId: node.id,
      approverId: resolution.staffId as StaffId,
      subject: dispatch.node.label,
      basis: Object.entries(payload).map(([k, v]) => ({ k, v: String(v) })),
    });
    await deps.store.insertInstance(record.id, {
      nodeId: node.id,
      instance,
      running: false,
      waiting: true,
      activated: true,
      approvalId: approval.id,
    });
    await deps.events.publish({ workflowId: record.id, kind: 'approval-raised', detail: { nodeId: node.id, approvalId: approval.id } });
  }

  /**
   * 从某节点继续流转:选边 → 装配 → 进下一节点(AND 汇合要回齐)。
   *
   * 返回**有没有真的推出去**。审批点的驳回要用它:图上显式画了驳回出边就走边,
   * 没画的包才退回 `reject` 声明 —— 调用方据此判断,不必自己再选一遍边。
   *
   * `quiet` 为真时不发 `route-dead-end`:那种「先试边、不成再退回声明」的探路
   * 一条边都没中是**正常分支**,不是死路,报出来就是假警。
   */
  async function advanceFrom(
    record: WorkflowRecord,
    pkg: ProcessPackage,
    fromNodeId: string,
    payload: Record<string, unknown>,
    quiet = false,
  ): Promise<boolean> {
    const scope = scopeOf(record);
    const flat = { ...scope.initiation, ...scope.vars, ...payload };
    const edges = selectEdges(pkg, fromNodeId, flat, deps.sfeel.condition);

    /*
     * 一条边都没选中,而这个节点确实有出边 —— 流到此为止,再没有谁会来推它。
     * 这是**死路**,不是终点。异或网关最容易碰上:发布闸要求它的每条出边都带条件,
     * 所以没有无条件兜底边可用,判据一缺就全不成立(实测:A1 的出口少了两枚去向旗,
     * G1 一条分支都选不中,流无声停在 A1)。
     *
     * 状态聚合已经会把它报成 `abnormal`(没到终点就不算完成),但那只说「有问题」、
     * 不说「哪儿有问题」。这里留一条可诊断的痕:哪个节点、出边各自的条件、判据里到底有哪些键。
     * 只记键名不记值 —— 诊断够用,载荷不进事件明文。
     */
    if (edges.length === 0) {
      const outgoing = pkg.edges.filter((e) => e.from === fromNodeId);
      if (outgoing.length > 0 && !quiet) {
        await deps.events.publish({
          workflowId: record.id,
          kind: 'route-dead-end',
          detail: {
            nodeId: fromNodeId,
            edges: outgoing.map((e) => ({ id: e.id, to: e.to, condition: e.condition ?? '(无条件)' })),
            availableKeys: Object.keys(flat).sort(),
          },
        });
      }
    }

    for (const edge of edges) {
      const target = pkg.nodes.find((n) => n.id === edge.to);
      if (target === undefined) continue;

      if (target.kind === 'gateway' && target.mode === 'join') {
        const instances = await deps.store.listInstances(record.id);
        const converged = new Set(instances.filter((i) => i.terminal === 'released').map((i) => i.nodeId));
        const tolerated = new Set(instances.filter((i) => i.failureDigested === true).map((i) => i.nodeId));
        // 未回齐就等着 —— 别的分支到了会再来一次
        if (!joinReady(pkg, target.id, converged, tolerated)) continue;
      }

      const assembled = assemble(edge, { ...scope, ports: { ...scope.ports, [fromNodeId]: payload } }, deps.sfeel.expression);
      /*
       * 网关不是责任单元,**按到达的东西路由**:它没有入参契约要满足,
       * 上游的输出口载荷要原样穿过去,判据才读得到(如 G4 读 outcome)。
       * 若只给边映射装配出来的那份,无映射的边就把判据依据整个丢掉,
       * 异或网关会一条分支都选不中,整条流悄无声息地停住。
       * 活动与审批点仍只收**装配后的**载荷 —— 它们有契约,不该被上游多余字段污染。
       */
      const arriving = target.kind === 'gateway' ? { ...payload, ...assembled } : assembled;
      try {
        await enter(record, pkg, target, arriving);
      } catch (error) {
        /*
         * 一条分支起不来,不该把整条终态上报连坐掀翻 —— 那会让**上游任务的终态也报不上去**,
         * 流停在原地且一行日志都没有。这里把它记成工作流事件并落一个未激活的实例:
         * 其余分支照走,这一支在治理台「需处置」里现形,等人处置。
         */
        const detail = error instanceof Error ? error.message : String(error);
        await deps.events.publish({
          workflowId: record.id,
          kind: 'node-start-failed',
          detail: { nodeId: target.id, from: fromNodeId, error: detail },
        });
        const instances = await deps.store.listInstances(record.id);
        await deps.store.insertInstance(record.id, {
          nodeId: target.id,
          instance: nextInstanceNumber(instances, target.id),
          running: false,
          waiting: true,
          activated: true,
          terminal: 'blocked',
        });
      }
    }
    return edges.length > 0;
  }

  return {
    async savePackage(pkg) {
      await deps.store.savePackage(pkg);
      return pkg;
    },

    async listPackages() {
      return deps.store.listPackages();
    },

    async getPackage(id, version) {
      return deps.store.getPackage(id, version);
    },

    /** 画图即校验:发布前跑闭合校验 */
    async validate(pkg) {
      return checkClosure({
        pkg,
        contracts: await deps.activities.faces(),
        positions: await deps.org.positions(),
      });
    },

    /** 不闭合不得发布 */
    async publish(id, version) {
      const pkg = await requirePackage(id, version);
      const report = await this.validate(pkg);
      if (!report.publishable) throw new NotPublishableError(report);
      await deps.store.publishPackage(id, version);
      return report;
    },

    /** 起单:经调用接口调起流程包,生成一条工作流 */
    async start(input: StartWorkflowInput) {
      const pkg =
        input.packageVersion === undefined
          ? await latestPublished(input.packageId)
          : await requirePackage(input.packageId, input.packageVersion);
      if (pkg.state !== 'published') throw new Error(`流程包未发布,不得起单:${pkg.id}@${pkg.version}`);

      // 先校验后建单:不合调用接口契约的输入**一条工作流都不留**。
      // (校验若放在建单之后,失败就会留下一条永远推不动的在途单,污染在途口径。)
      const check = deps.schemas.validate(pkg.callInterface.input, input.data);
      if (!check.ok) throw new CallInterfaceViolationError(pkg.id, check.errors);

      // 工作组按**来源人**装载(ADR-011):无来源人、或此人没建组 → 不装载,全走四策略
      const binding =
        input.source.by === undefined ? undefined : await deps.org.workGroupBinding(input.source.by);

      const record: WorkflowRecord = {
        id: deps.ids.next(),
        packageId: pkg.id,
        packageVersion: pkg.version,
        subject: input.subject,
        state: 'running',
        source: input.source,
        initiation: input.data,
        ports: {},
        vars: {},
        startedAt: deps.clock.now().toISOString(),
        ...(binding === undefined ? {} : { workGroup: { ...binding } }),
      };
      await deps.store.insertWorkflow(record);
      await deps.events.publish({ workflowId: record.id, kind: 'workflow-started', detail: { source: input.source } });

      const start = startNodeOf(pkg);
      if (start !== undefined) await advanceFrom(record, pkg, start.id, input.data);
      return record;
    },

    async getWorkflow(id) {
      return requireWorkflow(id);
    },

    async listWorkflows(limit = 50) {
      return deps.store.listWorkflows(limit);
    },

    async instancesOf(workflowId) {
      return deps.store.listInstances(workflowId);
    },

    /** 任务终态到达:按终态决定继续流转还是走失败走向 */
    async onTaskTerminal(taskId, terminal: TerminalState, ports: Record<string, Record<string, unknown>>) {
      const located = await deps.store.findInstanceByTask(taskId);
      if (located === undefined) return undefined;

      const record = await requireWorkflow(located.workflowId);
      const pkg = await requirePackage(record.packageId, record.packageVersion);
      const node = pkg.nodes.find((n) => n.id === located.nodeId);

      await deps.store.updateInstance(record.id, located.nodeId, located.instance, {
        running: false,
        waiting: false,
        terminal,
      });
      await deps.events.publish({ workflowId: record.id, kind: 'task-terminal', detail: { taskId, terminal } });

      const merged = await deps.store.updateWorkflow(record.id, { ports: { ...record.ports, ...ports } });

      if (terminal === 'released') {
        // 落地收敛后才发射输出口 —— 下游只见落地完成的任务。
        // 各口载荷合并成一份路由作用域:只取第一口,活动没有必交付口时就什么都拿不到。
        const payload: Record<string, unknown> = Object.assign({}, ...Object.values(ports));
        await advanceFrom(merged, pkg, located.nodeId, payload);
      } else if (node?.kind === 'activity') {
        // 非完成终态触发**声明的失败走向**;流程层不自动重试
        await this.applyFailure(record.id, located.nodeId, located.instance, node, terminal);
      }

      return this.refreshState(record.id);
    },

    /** 失败走向四选一:终止流(默认)/ 改道 / 升审批 / 容忍 */
    async applyFailure(workflowId, nodeId, instance, node, terminal) {
      const record = await requireWorkflow(workflowId);
      const pkg = await requirePackage(record.packageId, record.packageVersion);
      const disposition = node.failure;
      const digested = isDigested(disposition.kind);

      await deps.store.updateInstance(workflowId, nodeId, instance, { failureDigested: digested });
      await deps.events.publish({
        workflowId,
        kind: 'failure-disposition',
        detail: { nodeId, terminal, disposition: disposition.kind },
      });

      if (disposition.kind === 'reroute') {
        const edge = pkg.edges.find((e) => e.id === disposition.edgeId);
        const target = edge === undefined ? undefined : pkg.nodes.find((n) => n.id === edge.to);
        if (edge !== undefined && target !== undefined) {
          // 失败载荷 = 终态、原因码与说明
          const payload = assemble(edge, scopeOf(record), deps.sfeel.expression);
          await enter(record, pkg, target, { ...payload, __terminal: terminal });
        }
      }

      if (disposition.kind === 'escalate') {
        const target = pkg.nodes.find((n) => n.id === disposition.approvalNodeId);
        if (target !== undefined) await enter(record, pkg, target, { __terminal: terminal, __from: nodeId });
      }

      if (disposition.kind === 'tolerate') {
        /*
         * 容忍 = 这条支线的失败被吸收,**流继续往下走**。
         * 只把实例标成「已消化」而不往下推,汇合就永远等不到它:
         * `joinReady` 认 tolerated,可没人再去问一次它够不够齐。
         * 载荷为空 —— 容忍分支本就没有产出,下游的必填映射不得依赖它(闭合校验已拦)。
         */
        await advanceFrom(record, pkg, nodeId, {});
      }

      return this.refreshState(workflowId);
    },

    /** 审批裁决:通过继续流转;驳回按**声明的走向**回指定节点重开新实例 */
    async onApprovalVerdict(workflowId, nodeId, verdict, payload) {
      const record = await requireWorkflow(workflowId);
      const pkg = await requirePackage(record.packageId, record.packageVersion);
      const node = pkg.nodes.find((n) => n.id === nodeId);
      if (node?.kind !== 'approval') throw new Error(`节点不是审批点:${nodeId}`);

      const instances = await deps.store.listInstances(workflowId);
      const current = instances.filter((i) => i.nodeId === nodeId).at(-1);
      if (current !== undefined) {
        await deps.store.updateInstance(workflowId, nodeId, current.instance, {
          waiting: false,
          terminal: 'released',
        });
      }
      await deps.events.publish({ workflowId, kind: 'approval-verdict', detail: { nodeId, verdict } });

      /*
       * **裁决进载荷**:审批点后面跟网关时,网关判的就是这个字段
       * (如 G3「审批结果」的 `verdict = "pass"` / `verdict = "reject"`)。
       *
       * 从前只有 pass 走 `advanceFrom`,而且 payload 里不含 `verdict` —— 于是审批点后面
       * 就算画了网关也判不了(异或分叉的每条出边都得带条件、没有无条件兜底边,
       * 判据一缺就全不成立,流无声停在网关上)。要让它判得了,只能在边上挂一枚
       * 常量 `verdict: "pass"` 骗过条件 —— 那网关就成了摆设:条件恒真、驳回那条边永不可达。
       * 图上画着两个去向、其中一个结构上走不到,是在说谎。
       */
      const carried = { ...payload, verdict };

      if (verdict === 'pass') {
        await advanceFrom(record, pkg, nodeId, carried);
      } else {
        /*
         * 驳回:**先看图**。图上显式画了驳回出边(审批点后面跟网关)就走边 —— 编排即定契约,
         * 图是真相。没画的包才退回 `reject` 声明:它是 schema 必填的通用机制,
         * 不是所有流程包都会把驳回画成边。
         *
         * `quiet` 传真:这一步「试边不成」是正常分支,不是死路,不该报 route-dead-end。
         */
        const routedByEdge = await advanceFrom(record, pkg, nodeId, carried, true);
        if (!routedByEdge) {
          const route = rejectRouteOf(node, scopeOf(record), deps.sfeel.expression);
          if (route.kind === 'terminate') {
            await deps.store.updateWorkflow(workflowId, { state: 'terminated', finishedAt: deps.clock.now().toISOString() });
          } else {
            // 回指定节点**重开新实例**(同节点、新任务 ID、留痕关联),不复活旧任务
            const target = pkg.nodes.find((n) => n.id === route.nodeId);
            if (target !== undefined) await enter(record, pkg, target, { ...route.payload, ...payload });
          }
        }
      }
      return this.refreshState(workflowId);
    },

    /** 工作流状态是聚合投影 —— 每次变动后重算 */
    async refreshState(workflowId): Promise<WorkflowState> {
      const record = await requireWorkflow(workflowId);
      if (record.state === 'terminated') return 'terminated';
      const instances = await deps.store.listInstances(workflowId);
      // 「完成」凭到过终点,不凭没有在跑的了 —— 所以要把包的终点节点交给聚合
      const pkg = await deps.store.getPackage(record.packageId, record.packageVersion);
      const endNodeIds = (pkg?.nodes ?? []).filter((n) => n.kind === 'end').map((n) => n.id);
      const state = aggregateWorkflowState(instances as NodeInstance[], false, endNodeIds);
      if (state !== record.state) {
        await deps.store.updateWorkflow(workflowId, {
          state,
          ...(state === 'completed' ? { finishedAt: deps.clock.now().toISOString() } : {}),
        });
      }
      return state;
    },

    async terminate(workflowId, reason) {
      await deps.store.updateWorkflow(workflowId, {
        state: 'terminated',
        finishedAt: deps.clock.now().toISOString(),
      });
      // 工作流终止级联:在途任务终止、待答卡与征答消息作废、审批点撤卡
      await deps.events.publish({ workflowId, kind: 'workflow-terminated', detail: { reason } });
      return 'terminated';
    },
  };
}