BPOS-MVP/apps/api/src/modules/process/internal/adapter/process-store.ts

/**
 * process 存储 adapter —— Drizzle 只准出现在 adapter(编码规范 §5.3)。
 */
import { and, asc, desc, eq } from 'drizzle-orm';
import type { ProcessPackage, TerminalState, WorkflowState } from '@contracts';
import { processPackageSchema } from '@contracts';
import type { Db } from '@shared/db.ts';
import type { ProcessStore, WorkflowRecord } from '../../port.ts';
import type { NodeInstance } from '../domain/owner-resolution.ts';
import { nodeInstanceTable, processPackageTable, workflowTable } from './process-schema.ts';

type WorkflowRow = typeof workflowTable.$inferSelect;

function toWorkflow(row: WorkflowRow): WorkflowRecord {
  return {
    id: row.id,
    packageId: row.packageId,
    packageVersion: row.packageVersion,
    subject: row.subject,
    state: row.state as WorkflowState,
    source: row.source as WorkflowRecord['source'],
    initiation: row.initiation as Record<string, unknown>,
    ...(row.workGroup === null ? {} : { workGroup: row.workGroup as Record<string, string> }),
    ports: row.ports as Record<string, Record<string, unknown>>,
    vars: row.vars as Record<string, unknown>,
    startedAt: row.startedAt.toISOString(),
    ...(row.finishedAt === null ? {} : { finishedAt: row.finishedAt.toISOString() }),
  };
}

export function createProcessStore(db: Db): ProcessStore {
  return {
    /*
     * **只插不改**:版本一旦落库就是不可变的历史。
     * 版本号由用例层算(恒新增),所以这里不该撞键;真撞了就让唯一约束报出来 ——
     * 静默覆盖会改掉在途任务正在读的定义,那是最不该无声发生的事。
     */
    async savePackage(pkg: ProcessPackage) {
      await db
        .insert(processPackageTable)
        .values({ id: pkg.id, version: pkg.version, name: pkg.name, state: pkg.state, definition: pkg });
    },

    async setPackageState(id, version, state) {
      const rows = await db
        .select()
        .from(processPackageTable)
        .where(and(eq(processPackageTable.id, id), eq(processPackageTable.version, version)))
        .limit(1);
      const row = rows[0];
      if (row === undefined) return;
      // state 在列上与 definition(jsonb)里各有一份 —— 读包走 definition,列供查询;两处必须一起改
      const definition = { ...(row.definition as Record<string, unknown>), state };
      await db
        .update(processPackageTable)
        .set({ state, definition })
        .where(and(eq(processPackageTable.id, id), eq(processPackageTable.version, version)));
    },

    async getPackage(id, version) {
      const conditions = [eq(processPackageTable.id, id)];
      if (version !== undefined) conditions.push(eq(processPackageTable.version, version));
      const rows = await db
        .select()
        .from(processPackageTable)
        .where(and(...conditions))
        .orderBy(desc(processPackageTable.version))
        .limit(1);
      return rows[0] === undefined ? undefined : processPackageSchema.parse(rows[0].definition);
    },

    async listPackages() {
      const rows = await db.select().from(processPackageTable).orderBy(asc(processPackageTable.name));
      return rows.map((r) => processPackageSchema.parse(r.definition));
    },

    async insertWorkflow(record) {
      await db.insert(workflowTable).values({
        id: record.id,
        packageId: record.packageId,
        packageVersion: record.packageVersion,
        subject: record.subject,
        state: record.state,
        source: record.source,
        initiation: record.initiation,
        workGroup: record.workGroup ?? null,
        ports: record.ports,
        vars: record.vars,
        startedAt: new Date(record.startedAt),
      });
    },

    async getWorkflow(id) {
      const rows = await db.select().from(workflowTable).where(eq(workflowTable.id, id)).limit(1);
      return rows[0] === undefined ? undefined : toWorkflow(rows[0]);
    },

    async listWorkflows(limit) {
      const rows = await db.select().from(workflowTable).orderBy(desc(workflowTable.startedAt)).limit(limit);
      return rows.map(toWorkflow);
    },

    async updateWorkflow(id, patch) {
      const values: Record<string, unknown> = {};
      if (patch.state !== undefined) values['state'] = patch.state;
      if (patch.ports !== undefined) values['ports'] = patch.ports;
      if (patch.vars !== undefined) values['vars'] = patch.vars;
      if (patch.finishedAt !== undefined) values['finishedAt'] = new Date(patch.finishedAt);
      const [row] = await db.update(workflowTable).set(values).where(eq(workflowTable.id, id)).returning();
      if (row === undefined) throw new Error(`工作流不存在:${id}`);
      return toWorkflow(row);
    },

    async listInstances(workflowId) {
      const rows = await db
        .select()
        .from(nodeInstanceTable)
        .where(eq(nodeInstanceTable.workflowId, workflowId))
        .orderBy(asc(nodeInstanceTable.createdAt));
      return rows.map(
        (r): NodeInstance => ({
          nodeId: r.nodeId,
          instance: r.instance,
          running: r.running,
          waiting: r.waiting,
          activated: r.activated,
          ...(r.terminal === null ? {} : { terminal: r.terminal as TerminalState }),
          ...(r.failureDigested === null ? {} : { failureDigested: r.failureDigested }),
          ...(r.taskId === null ? {} : { taskId: r.taskId }),
          ...(r.approvalId === null ? {} : { approvalId: r.approvalId }),
        }),
      );
    },

    async insertInstance(workflowId, instance) {
      await db.insert(nodeInstanceTable).values({
        workflowId,
        nodeId: instance.nodeId,
        instance: instance.instance,
        running: instance.running,
        waiting: instance.waiting,
        activated: instance.activated,
        terminal: instance.terminal ?? null,
        failureDigested: instance.failureDigested ?? null,
        taskId: instance.taskId ?? null,
        approvalId: instance.approvalId ?? null,
      });
    },

    async updateInstance(workflowId, nodeId, instance, patch) {
      const values: Record<string, unknown> = {};
      if (patch.running !== undefined) values['running'] = patch.running;
      if (patch.waiting !== undefined) values['waiting'] = patch.waiting;
      if (patch.terminal !== undefined) values['terminal'] = patch.terminal;
      if (patch.failureDigested !== undefined) values['failureDigested'] = patch.failureDigested;
      if (patch.taskId !== undefined) values['taskId'] = patch.taskId;
      if (Object.keys(values).length === 0) return;
      await db
        .update(nodeInstanceTable)
        .set(values)
        .where(
          and(
            eq(nodeInstanceTable.workflowId, workflowId),
            eq(nodeInstanceTable.nodeId, nodeId),
            eq(nodeInstanceTable.instance, instance),
          ),
        );
    },

    async findInstanceByTask(taskId) {
      const rows = await db
        .select()
        .from(nodeInstanceTable)
        .where(eq(nodeInstanceTable.taskId, taskId))
        .limit(1);
      const row = rows[0];
      return row === undefined ? undefined : { workflowId: row.workflowId, nodeId: row.nodeId, instance: row.instance };
    },
  };
}