UHD_Media_Tool/apps/server/test/acquisition-organize/enqueue-recovery.test.ts

import type { TaskRunView } from "@umt/tasks";
import { createTempTree, type TempTree } from "@umt/testkit";
import { sql } from "kysely";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
import { enqueueSweep, enqueueTracked } from "../../src/modules/acquisition-organize/services/common.ts";
import {
  type QueueScrapeInput,
  queueScrape,
  runScrapeRecord,
} from "../../src/modules/acquisition-organize/services/scrape.ts";
import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts";

const hasPg = await pgAvailable();
const runView = (runId: string, input: unknown, status: TaskRunView["status"] = "pending"): TaskRunView => ({
  runId,
  taskName: "scrapeMedia",
  version: 2,
  status,
  attempts: 0,
  idempotencyKey: null,
  traceId: null,
  priority: 0,
  input,
  output: null,
  error: null,
  cancelRequestedAt: null,
  cancelReason: null,
  availableAt: null,
  createdAt: new Date(),
  startedAt: null,
  finishedAt: null,
});

describe.skipIf(!hasPg)("入队关联与巡检恢复", () => {
  let h: Harness;
  let lib: TempTree;
  let rootId: string;
  let entityId: string;

  beforeAll(async () => {
    lib = await createTempTree({}, { sentinel: true });
    h = await createHarness("enqueue_recovery");
    ({ rootId } = await seedBase(h.db, { libraryPath: lib.root, cachePath: lib.root, qbUrl: "http://127.0.0.1:9" }));
    entityId = await seedEntity(h.db, { kind: "movie", title: "恢复测试", year: 2020 });
    await sql`create function reject_association() returns trigger language plpgsql as $$
      begin
        if new.engine_run_id is not null and new.engine_run_id is distinct from old.engine_run_id
          then raise exception 'injected association failure'; end if;
        return new;
      end $$`.execute(h.db);
  });

  async function rejectAssociation(table: "task" | "scrape_record") {
    await sql
      .raw(`create trigger reject_association before update on ${table}
      for each row execute function reject_association()`)
      .execute(h.db);
  }

  async function allowAssociation() {
    for (const table of ["task", "scrape_record"])
      await sql.raw(`drop trigger if exists reject_association on ${table}`).execute(h.db);
  }

  afterEach(async () => {
    vi.restoreAllMocks();
    await allowAssociation();
  });
  afterAll(async () => {
    await h?.close();
    await lib?.cleanup();
  });

  const task = (key: string) => ({ kind: "reconcileLibrary" as const, idempotencyKey: key, input: {} });
  const scrape = (path: string, idempotencyKey?: string): QueueScrapeInput => ({
    rootId: `lib:${rootId}`,
    rootPath: lib.root,
    libraryRootId: rootId,
    path,
    match: {
      mediaEntityId: entityId,
      provider: null,
      externalId: null,
      kind: "movie",
      title: "恢复测试",
      year: 2020,
      source: "chosen",
    },
    trigger: "manual",
    mode: "missing",
    ...(idempotencyKey ? { idempotencyKey } : {}),
  });

  it("引擎查询异常跳过本轮,下一轮仍查原任务", async () => {
    const t = task("query_error");
    const q = await enqueueTracked(h.d, t);
    const enqueue = vi.spyOn(h.engine, "enqueue");
    const get = vi
      .spyOn(h.engine, "get")
      .mockRejectedValueOnce(new Error("temporary engine outage"))
      .mockResolvedValue(runView(q.runId, {}, "running"));
    expect(await enqueueSweep(h.d, t, "next")).toBeNull();
    expect(await enqueueSweep(h.d, t, "later")).toBeNull();
    expect(enqueue).not.toHaveBeenCalled();
    expect(get).toHaveBeenNthCalledWith(2, q.runId);
    expect(
      (await h.db.selectFrom("task").select("engine_run_id").where("id", "=", q.taskId).executeTakeFirstOrThrow())
        .engine_run_id,
    ).toBe(q.runId);
  });

  it.each(["missing", "completed"])("明确 %s 时允许补投", async (state) => {
    const t = task(`retry_${state}`);
    const q = await enqueueTracked(h.d, t);
    vi.spyOn(h.engine, "get").mockResolvedValue(state === "missing" ? null : runView(q.runId, {}, "completed"));
    const enqueue = vi.spyOn(h.engine, "enqueue");
    expect(await enqueueSweep(h.d, t, "next")).not.toBeNull();
    expect(enqueue).toHaveBeenCalledOnce();
    expect(enqueue.mock.calls[0]?.[2].idempotencyKey).toBe(`${t.kind}:${t.idempotencyKey}:next`);
  });

  it.each([false, true])("关联写回失败后巡检按原 taskId 找回已接受的 run (旧关联=%s)", async (oldAssociation) => {
    const t = task(`association_${oldAssociation}`);
    if (oldAssociation) await enqueueTracked(h.d, t);
    await rejectAssociation("task");
    const q = await enqueueTracked(h.d, t, { engineKey: "new-slot" });
    await h.db
      .insertInto("tasks_run_meta")
      .values({ run_id: q.runId, task_name: t.kind, task_version: 1, idempotency_key: "new-slot", trace_id: q.taskId })
      .execute();
    await allowAssociation();
    vi.spyOn(h.engine, "get").mockResolvedValue(runView(q.runId, {}, "running"));
    const enqueue = vi.spyOn(h.engine, "enqueue");
    expect(await enqueueSweep(h.d, t, "later")).toBeNull();
    expect(enqueue).not.toHaveBeenCalled();
    expect(
      (await h.db.selectFrom("task").select("engine_run_id").where("id", "=", q.taskId).executeTakeFirstOrThrow())
        .engine_run_id,
    ).toBe(q.runId);
  });

  it("刮削已接受而关联失败时保持 queued,重试复用记录与原幂等键", async () => {
    const input = scrape("association-retry");
    await rejectAssociation("scrape_record");
    const q = await queueScrape(h.d, input);
    expect(
      await h.db
        .selectFrom("scrape_record")
        .select(["status", "reason", "engine_run_id"])
        .where("id", "=", q.recordId)
        .executeTakeFirstOrThrow(),
    ).toMatchObject({ status: "queued", reason: null, engine_run_id: null });
    const key = h.engine.enqueued.at(-1)?.key;
    await allowAssociation();
    const enqueue = vi.spyOn(h.engine, "enqueue").mockResolvedValue({ runId: q.taskId!, deduplicated: true });
    vi.spyOn(h.engine, "get").mockResolvedValue(runView(q.taskId!, { recordId: q.recordId }));
    expect(await queueScrape(h.d, input)).toEqual({ recordId: q.recordId, taskId: q.taskId, deduplicated: true });
    expect(enqueue.mock.calls[0]?.[2].idempotencyKey).toBe(key);
    expect((await h.db.selectFrom("scrape_record").select("id").where("path", "=", input.path).execute()).length).toBe(
      1,
    );
  });

  it("刮削处理器开始时补关联,不会把已接受的工作误标为入队失败", async () => {
    await rejectAssociation("scrape_record");
    const q = await queueScrape(h.d, scrape("worker-backfill"));
    await allowAssociation();
    await h.db.updateTable("library_root").set({ management_mode: "register_only" }).where("id", "=", rootId).execute();
    const r = await runScrapeRecord(h.d, q.recordId, undefined, q.taskId!);
    expect(r.reason).toBe("scrape_root_read_only");
    expect(
      (
        await h.db
          .selectFrom("scrape_record")
          .select("engine_run_id")
          .where("id", "=", q.recordId)
          .executeTakeFirstOrThrow()
      ).engine_run_id,
    ).toBe(q.taskId);
  });

  it("确实入队失败的新刮削记录标失败", async () => {
    vi.spyOn(h.engine, "enqueue").mockRejectedValue(new Error("enqueue rejected"));
    const input = scrape("enqueue-failed");
    await expect(queueScrape(h.d, input)).rejects.toThrow("enqueue rejected");
    expect(
      await h.db
        .selectFrom("scrape_record")
        .select(["status", "reason"])
        .where("path", "=", input.path)
        .executeTakeFirstOrThrow(),
    ).toMatchObject({ status: "failed", reason: "task_enqueue_failed" });
  });

  it("自动重放从原 run 输入找回未关联记录,只删除本次多建的记录", async () => {
    const input = scrape("auto-replay", "auto-stable-key");
    await rejectAssociation("scrape_record");
    const q = await queueScrape(h.d, input);
    await allowAssociation();
    await h.db.updateTable("scrape_record").set({ status: "succeeded" }).where("id", "=", q.recordId).execute();
    vi.spyOn(h.engine, "enqueue").mockResolvedValue({ runId: q.taskId!, deduplicated: true });
    vi.spyOn(h.engine, "get").mockResolvedValue(runView(q.taskId!, { recordId: q.recordId }, "completed"));
    expect(await queueScrape(h.d, input)).toEqual({ recordId: q.recordId, taskId: q.taskId, deduplicated: true });
    const recs = await h.db
      .selectFrom("scrape_record")
      .select(["id", "status", "engine_run_id"])
      .where("path", "=", input.path)
      .execute();
    expect(recs).toEqual([{ id: q.recordId, status: "succeeded", engine_run_id: q.taskId }]);
  });
});