UHD_Media_Tool/apps/server/test/auth-settings/engine-retry.test.ts

import { sql } from "@umt/db";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
import { DEV_KEY, json } from "../helpers.ts";
import { bearer, createHarness, type Harness, pgAvailable } from "./harness.ts";

// 引擎面板 retry(验收 S8):仅 admin;按原任务名与信封里的原始输入重新入队,幂等键 retry:<id>:<序号>;
// 运行中/待执行的 run 不能重跑(409);未知 run 404;写审计。

const hasPg = await pgAvailable();

describe.skipIf(!hasPg)("POST /engine/runs/{id}/retry", () => {
  let h: Harness;
  let admin: ReturnType<typeof bearer>;
  let n = 0;
  const enqueue = vi.fn(async (_task: string, _input: unknown, _opts: { idempotencyKey: string }) => {
    n++;
    await sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key)
      values (${`new-${n}`}, ${_task}, 1, ${_opts.idempotencyKey})`.execute(h.db);
    return { runId: `new-${n}`, deduplicated: false };
  });

  beforeAll(async () => {
    h = await createHarness({ engine: true });
    admin = bearer(h.app, DEV_KEY);
    (h.ctx.engine as unknown as { enqueue: typeof enqueue }).enqueue = enqueue;
    const t = sql.raw(`"${h.engineSchema}".workflow_runs`);
    const run = (id: string, status: string, input: unknown) =>
      sql`insert into ${t} (namespace_id, id, workflow_name, version, status, config, context, input, attempts,
          created_at, updated_at)
        values ('default', ${id}, 'verifyAndImport', 'v1', ${status}, '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 3,
          now(), now())`.execute(h.db);
    // 信封里的原始输入含敏感字段:重跑用原值(详情展示才脱敏)
    await run("f1", "failed", { input: { downloadTaskId: "dt-1", password: "p@ss" }, traceId: "trace-1", priority: 5 });
    await run("p1", "pending", { input: { downloadTaskId: "dt-2" }, traceId: "trace-2", priority: 0 });
  }, 60_000);
  afterAll(async () => {
    await h?.close();
  });

  it("工作流多值筛选保留排队核验与独立刮削，单值筛选兼容", async () => {
    const table = sql.raw(`"${h.engineSchema}".workflow_runs`);
    await sql`insert into ${table} (namespace_id, id, workflow_name, version, status, config, context, input, attempts, created_at, updated_at)
      values ('default', 'scrape-queue', 'scrapeMedia', 'v2', 'pending', '{}'::jsonb, null, '{"input":{}}'::jsonb, 0, now(), now())`.execute(
      h.db,
    );
    try {
      const both = await json(
        await admin("GET", "/engine/runs?workflowName=verifyAndImport,scrapeMedia&status=pending"),
      );
      expect(both.data.items.map((run: { id: string }) => run.id).sort()).toEqual(["p1", "scrape-queue"]);
      const single = await json(await admin("GET", "/engine/runs?workflowName=verifyAndImport&status=pending"));
      expect(single.data.items.map((run: { id: string }) => run.id)).toEqual(["p1"]);
    } finally {
      await sql`delete from ${table} where id = 'scrape-queue'`.execute(h.db);
    }
  });

  it("失败的 run:按原任务名与原始输入重新入队,序号递增,写审计", async () => {
    const r1 = await json(await admin("POST", "/engine/runs/f1/retry"));
    expect(r1.data).toEqual({ runId: "new-1", retryOf: "f1", attempt: 1 });
    expect(enqueue).toHaveBeenLastCalledWith(
      "verifyAndImport",
      { downloadTaskId: "dt-1", password: "p@ss" },
      { idempotencyKey: "retry:f1:1", traceId: "trace-1", priority: 5 },
    );
    const r2 = await json(await admin("POST", "/engine/runs/f1/retry"));
    expect(r2.data).toMatchObject({ runId: "new-2", attempt: 2 });
    expect(enqueue.mock.calls.at(-1)![2].idempotencyKey).toBe("retry:f1:2");
    const a = await sql<{ n: number }>`
      select count(*)::int as n from audit_event where action = 'engine.run_retry'`.execute(h.db);
    expect(a.rows[0]!.n).toBe(2);
  });

  it("未结束的 run 409,未知 run 404,非 admin 403", async () => {
    expect((await admin("POST", "/engine/runs/p1/retry")).status).toBe(409);
    expect((await admin("POST", "/engine/runs/nope/retry")).status).toBe(404);
    const created = await json(await admin("POST", "/api-keys", { name: "exec", scopes: ["organize.execute"] }));
    const other = bearer(h.app, created.data.plaintextKey);
    expect((await other("POST", "/engine/runs/f1/retry")).status).toBe(403);
    expect(n).toBe(2);
  });
});