UHD_Media_Tool/packages/tasks/test/engine.test.ts

import { sql } from "kysely";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import type { OpenWorkflowEngine } from "../src";
import { createTestRegistry } from "./fixtures/tasks";
import {
  connectDb,
  createSideEffectTable,
  dropSchemas,
  pgAvailable,
  resetSchemas,
  setupEngine,
  sideEffects,
  type TestDb,
  testSchemas,
  waitFor,
} from "./helpers";

const schemas = testSchemas("engine");
const hasPg = await pgAvailable();

describe.skipIf(!hasPg)("OpenWorkflowEngine(进程内 worker)", () => {
  let db: TestDb;
  let engine: OpenWorkflowEngine;

  beforeAll(async () => {
    await resetSchemas(schemas);
    db = connectDb(schemas.biz);
    await createSideEffectTable(db);
    engine = await setupEngine(db, schemas, createTestRegistry(db));
    await engine.startWorker({ concurrency: 4 });
  }, 30_000);

  afterAll(async () => {
    await engine?.stop();
    await db?.destroy();
    await dropSchemas(schemas);
  }, 60_000);

  it("enqueue 幂等:同 key 返回同一 run,不同 key 新建;traceId 进入 handler", async () => {
    const a = await engine.enqueue("echo", { n: 1 }, { idempotencyKey: "k-1", traceId: "t-1", priority: 5 });
    const b = await engine.enqueue("echo", { n: 2 }, { idempotencyKey: "k-1", traceId: "t-2" });
    const c = await engine.enqueue("echo", { n: 3 }, { idempotencyKey: "k-2", traceId: "t-3" });
    expect(a.deduplicated).toBe(false);
    expect(b).toEqual({ runId: a.runId, deduplicated: true });
    expect(c.runId).not.toBe(a.runId);

    const done = await waitFor(async () => {
      const v = await engine.get(a.runId);
      return v?.status === "completed" ? v : null;
    }, 15_000);
    expect(done.output).toEqual({ n: 1, trace: "t-1" });
    expect(done).toMatchObject({ taskName: "echo", version: 1, traceId: "t-1", priority: 5, idempotencyKey: "k-1" });

    const listed = await engine.list({ taskName: "echo" });
    expect(listed.map((r) => r.runId).sort()).toEqual([a.runId, c.runId].sort());
  }, 30_000);

  it("enqueue 拒绝不合 schema 的输入与未注册任务", async () => {
    await expect(engine.enqueue("echo", { n: "x" }, { idempotencyKey: "bad", traceId: "t" })).rejects.toThrow();
    await expect(engine.enqueue("nope", {}, { idempotencyKey: "bad2", traceId: "t" })).rejects.toThrow(
      /not registered/,
    );
  });

  it("取消:进行中的长步骤经 abortSignal 中止,后续步骤不执行", async () => {
    const { runId } = await engine.enqueue("cancellable", {}, { idempotencyKey: "c-1", traceId: "t" });
    await waitFor(async () => (await sideEffects(db, runId)).some((s) => s.step === "long"), 15_000, "long start");
    expect(await engine.cancel(runId, "user requested")).toBe(true);

    await waitFor(
      async () => (await sideEffects(db, runId)).some((s) => s.note === "aborted"),
      5_000,
      "abort propagated",
    );
    const v = await engine.get(runId);
    expect(v?.status).toBe("canceled");
    expect(v?.cancelReason).toBe("user requested");
    await new Promise((r) => setTimeout(r, 1500));
    expect((await sideEffects(db, runId)).map((s) => s.step)).not.toContain("after");
    // 已取消再取消:幂等
    expect(await engine.cancel(runId)).toBe(true);
  }, 30_000);

  it("取消已完成的 run 返回 false", async () => {
    const { runId } = await engine.enqueue("echo", { n: 9 }, { idempotencyKey: "c-done", traceId: "t" });
    await waitFor(async () => (await engine.get(runId))?.status === "completed", 15_000);
    expect(await engine.cancel(runId)).toBe(false);
  }, 30_000);

  it("waitForFact:事实先于等待到达不丢失;等待中到达的事实经信号唤醒", async () => {
    const { runId } = await engine.enqueue("waitFacts", {}, { idempotencyKey: "f-1", traceId: "t" });
    // 此刻 handler 还在 before 步骤(1.5s),没有等待者,引擎信号会丢,但事实已落库
    const early = await engine.signal(runId, "a", { approved: true });
    expect(early.delivered).toBe(0);

    await waitFor(async () => (await sideEffects(db, runId)).some((s) => s.step === "got-a"), 15_000, "got-a");
    await waitFor(async () => (await engine.get(runId))?.status === "waiting", 10_000, "waiting on b");
    const late = await engine.signal(runId, "b", 42);
    expect(late.delivered).toBe(1);

    const v = await waitFor(async () => {
      const r = await engine.get(runId);
      return r?.status === "completed" ? r : null;
    }, 15_000);
    expect(v.output).toEqual({ a: { approved: true }, b: 42 });
    const facts = await sql<{ n: string }>`SELECT count(*) n FROM tasks_fact WHERE run_id = ${runId}`.execute(db);
    expect(Number(facts.rows[0]!.n)).toBe(2);
  }, 45_000);
});