// Local disposable PostgreSQL only. Install embedded-postgres into .qa/video-task-db first.
const fs = require("fs"),
  path = require("path"),
  assert = require("node:assert/strict");
(async () => {
  const { default: EmbeddedPostgres } = await import(
    path.resolve(".qa/video-task-db/node_modules/embedded-postgres/dist/index.js")
  );
  const pg = new EmbeddedPostgres({
    databaseDir: path.resolve(".qa/video-task-db/data"),
    port: 55439,
    user: "canvas_qa",
    password: "local-disposable-only",
    persistent: false,
    onLog: (message) => console.log(message),
    onError: (message) => console.error(message),
  });
  let client;
  try {
    await pg.initialise();
    await pg.start();
    client = pg.getPgClient();
    await client.connect();
    process.env.POSTGRES_URL =
      "postgres://canvas_qa:local-disposable-only@127.0.0.1:55439/postgres";
    await client.query(
      'CREATE TABLE "User" (id uuid PRIMARY KEY); CREATE TABLE "CanvasProject" (id uuid PRIMARY KEY, "userId" uuid);',
    );
    await client.query(
      fs.readFileSync("lib/db/migrations/0091_add_canvas_capability_tasks.sql", "utf8"),
    );
    const esbuild = require(
      path.resolve(
        "node_modules/.pnpm",
        fs.readdirSync("node_modules/.pnpm").find((name) => name.startsWith("esbuild@0.25.")),
        "node_modules/esbuild",
      ),
    );
    await esbuild.build({
      entryPoints: ["lib/video-canvas/task-store.ts"],
      outfile: ".qa/video-task-db/store.cjs",
      bundle: true,
      platform: "node",
      format: "cjs",
      packages: "external",
      alias: { "@": process.cwd() },
      plugins: [
        {
          name: "server-only",
          setup(build) {
            build.onResolve({ filter: /^server-only$/ }, () => ({
              path: "empty",
              namespace: "qa",
            }));
            build.onLoad({ filter: /.*/, namespace: "qa" }, () => ({ contents: "" }));
          },
        },
      ],
    });
    const store = require(path.resolve(".qa/video-task-db/store.cjs"));
    const user = crypto.randomUUID(),
      other = crypto.randomUUID(),
      project = crypto.randomUUID();
    await client.query('INSERT INTO "User" VALUES ($1),($2)', [user, other]);
    await client.query('INSERT INTO "CanvasProject" VALUES ($1,$2)', [project, user]);
    assert.equal(await store.ownsTaskProject(user, project), true);
    assert.equal(await store.ownsTaskProject(other, project), false);
    const request = {
      requestId: crypto.randomUUID(),
      projectId: project,
      kind: "video",
      params: { mode: "text-to-video", prompt: "QA" },
      node: {
        id: "node-a",
        title: "QA",
        kind: "media",
        inputType: "video",
        x: 0,
        y: 0,
        width: 300,
        height: 200,
      },
      edges: [],
    };
    const copies = await Promise.all(
      Array.from({ length: 10 }, () => store.createCanvasGenerationTask(user, request)),
    );
    assert.equal(new Set(copies.map((task) => task.id)).size, 1);
    const task = copies[0];
    await assert.rejects(
      store.createCanvasGenerationTask(user, {
        ...request,
        params: { ...request.params, prompt: "different" },
      }),
    );
    const claims = await Promise.all(
      Array.from({ length: 10 }, () => store.claimCanvasGenerationTask(user, project)),
    );
    assert.equal(claims.filter(Boolean).length, 1);
    assert.equal((await store.cancelCanvasGenerationTask(user, project, task.id)).ok, false);
    assert.equal((await store.cancelCanvasGenerationTask(other, project, task.id)).task, null);
    const pending = await store.createCanvasGenerationTask(user, {
      ...request,
      requestId: crypto.randomUUID(),
      node: { ...request.node, id: "node-b" },
    });
    assert.equal(await store.claimCanvasGenerationTask(user, project), null);
    assert.equal((await store.cancelCanvasGenerationTask(user, project, pending.id)).ok, true);
    await store.finishCanvasGenerationTask(task.id, { url: "https://example.com/qa.mp4" });
    assert.equal(
      (await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === task.id).result
        .url,
      "https://example.com/qa.mp4",
    );
    assert.equal((await store.listCanvasGenerationTasks(other, project)).length, 0);
    assert.equal(await store.hideCanvasGenerationNode(user, project, task.id, "node-a"), true);
    assert.deepEqual(
      (await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === task.id).request
        .hiddenNodeIds,
      ["node-a"],
    );
    const stale = await store.createCanvasGenerationTask(user, {
      ...request,
      requestId: crypto.randomUUID(),
      node: { ...request.node, id: "node-c" },
    });
    await store.claimCanvasGenerationTask(user, project);
    await client.query(
      "UPDATE \"CanvasCapabilityTask\" SET \"updatedAt\"=(now() at time zone 'UTC')-interval '3 minutes' WHERE id=$1",
      [stale.id],
    );
    assert.equal(
      (await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === stale.id).status,
      "interrupted",
    );
    assert.equal(await store.claimCanvasGenerationTask(user, project), null);
    await store.checkpointCanvasGenerationTask(stale.id, {
      upstreamTask: { taskId: "ignored-after-interruption" },
    });
    assert.equal(
      (await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === stale.id).result,
      null,
    );
    await store.finishCanvasGenerationTask(stale.id, { url: "https://example.com/late.mp4" });
    assert.equal(
      (await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === stale.id).status,
      "succeeded",
    );
    const upstream = await store.createCanvasGenerationTask(user, {
      ...request,
      requestId: crypto.randomUUID(),
      node: { ...request.node, id: "node-upstream" },
    });
    await store.claimCanvasGenerationTask(user, project);
    await store.checkpointCanvasGenerationTask(upstream.id, {
      upstreamTask: { operation: "enhance", taskId: "provider-123" },
    });
    await store.interruptCanvasGenerationTask(upstream.id, "查询连接中断");
    const recovered = (await store.listCanvasGenerationTasks(user, project)).find(
      (t) => t.id === upstream.id,
    );
    assert.equal(recovered.status, "interrupted");
    assert.equal(recovered.result.upstreamTask.taskId, "provider-123");
    assert.equal(await store.claimCanvasGenerationTask(user, project), null);
    // Reclaim one checkpoint under concurrency and reject all stale worker writes.
    const enhanceRequest = {
      ...request,
      requestId: crypto.randomUUID(),
      kind: "image-operation",
      node: { ...request.node, id: "node-recoverable" },
      params: { operation: "enhance", imageUrl: "https://example.com/source.png" },
    };
    const resumable = await store.createCanvasGenerationTask(user, enhanceRequest);
    const firstLease = await store.claimCanvasGenerationTask(user, project);
    await store.checkpointCanvasGenerationTask(
      resumable.id,
      {
        upstreamTask: {
          operation: "enhance",
          taskId: "provider-resume",
          sourceImageUrl: enhanceRequest.params.imageUrl,
        },
      },
      firstLease.attempts,
    );
    await store.interruptCanvasGenerationTask(resumable.id, "query lost", firstLease.attempts);
    assert.equal(
      (
        await store.createCanvasGenerationTask(user, {
          ...enhanceRequest,
          requestId: crypto.randomUUID(),
        })
      ).id,
      resumable.id,
    );
    assert.equal(await store.claimCanvasGenerationTask(user, project), null);
    await client.query(
      "UPDATE \"CanvasCapabilityTask\" SET \"updatedAt\"=(now() at time zone 'UTC')-interval '6 minutes' WHERE id=$1",
      [resumable.id],
    );
    const resumed = (
      await Promise.all(
        Array.from({ length: 10 }, () => store.claimCanvasGenerationTask(user, project)),
      )
    ).filter(Boolean);
    assert.equal(resumed.length, 1);
    assert.equal(resumed[0].attempts, 2);
    assert.equal(await store.ownsCanvasTaskLease(resumable.id, 1), false);
    assert.equal(await store.ownsCanvasTaskLease(resumable.id, 2), true);
    await store.finishCanvasGenerationTask(
      resumable.id,
      { images: [{ originalUrl: "stale" }] },
      undefined,
      1,
    );
    await store.interruptCanvasGenerationTask(resumable.id, "stale failure", 1);
    await store.checkpointCanvasGenerationTask(
      resumable.id,
      { upstreamTask: { taskId: "wrong" } },
      1,
    );
    const fenced = (await store.listCanvasGenerationTasks(user, project)).find(
      (t) => t.id === resumable.id,
    );
    assert.equal(fenced.status, "running");
    assert.equal(fenced.result.upstreamTask.taskId, "provider-resume");
    await store.finishCanvasGenerationTask(
      resumable.id,
      { images: [{ originalUrl: "https://example.com/recovered.png" }] },
      undefined,
      2,
    );

    // Video checkpoints use the same durable lease and original provider identity.
    const videoRequest = {
      ...request,
      requestId: crypto.randomUUID(),
      kind: "video",
      node: { ...request.node, id: "video-recoverable" },
      params: { prompt: "QA only", preferredModelRef: "video-model" },
    };
    const videoTask = await store.createCanvasGenerationTask(user, videoRequest);
    const videoLease = await store.claimCanvasGenerationTask(user, project);
    const videoCheckpoint = {
      operation: "video",
      provider: "aliyun",
      taskId: "original-video",
      modelRef: "video-model",
      modelId: "wan-qa",
      providerId: "provider-qa",
      baseUrl: "https://provider.invalid",
    };
    await store.checkpointCanvasGenerationTask(
      videoTask.id,
      { upstreamTask: videoCheckpoint },
      videoLease.attempts,
    );
    await store.interruptCanvasGenerationTask(
      videoTask.id,
      "query disconnected",
      videoLease.attempts,
    );
    assert.equal(
      (
        await store.createCanvasGenerationTask(user, {
          ...videoRequest,
          requestId: crypto.randomUUID(),
        })
      ).id,
      videoTask.id,
    );
    await client.query(
      `UPDATE "CanvasCapabilityTask" SET "updatedAt"=(now() at time zone 'UTC')-interval '6 minutes' WHERE id=$1`,
      [videoTask.id],
    );
    const videoReclaim = (
      await Promise.all(
        Array.from({ length: 10 }, () => store.claimCanvasGenerationTask(user, project)),
      )
    ).filter(Boolean);
    assert.equal(videoReclaim.length, 1);
    assert.deepEqual(videoReclaim[0].result.upstreamTask, videoCheckpoint);
    await store.finishCanvasGenerationTask(
      videoTask.id,
      { videoUrl: "https://output.invalid/video.mp4" },
      undefined,
      videoReclaim[0].attempts,
    );
    console.log(
      "PASS: video checkpoint recovery, ten concurrent reclaimers, original upstream identity and duplicate prevention",
    );

    // A separate worker process exits after its first GET. Restart must read the persisted ID.
    const callsFile = path.resolve(".qa/video-task-db/provider-calls.jsonl");
    fs.writeFileSync(callsFile, "");
    const unexpected = `throw Error("unexpected provider path")`;
    const stubs = {
      "./smart-edit/execute": `export const executeSmartEdit=async()=>{${unexpected}};`,
      "./audio-generation": `export const executeCanvasAudio=async()=>{${unexpected}};`,
      "@/lib/ai/tools/generate-video": `export const executeDirectVideoGeneration=async()=>{${unexpected}};`,
      "@/lib/canvas/generate-image-response": `export const executeCanvasImageResponse=async()=>{${unexpected}};`,
      "./workflow-analysis-response": `export const executeWorkflowAnalysisResponse=async()=>{${unexpected}};`,
      "@/lib/canvas/operations/extend": `export const executeExtendPOST=async()=>{${unexpected}};`,
      "@/lib/canvas/operations/rmbg": `export const executeRemoveBackgroundPOST=async()=>{${unexpected}};`,
      "@/lib/canvas/operations/angle-edit": `export const executeAngleEditPOST=async()=>{${unexpected}};`,
      "@/lib/canvas/operations/enhance": `
        import fs from 'node:fs';
        const log=(method,id)=>fs.appendFileSync(${JSON.stringify(callsFile)},JSON.stringify({method,id})+'\\n');
        export const executeEnhancePOST=async(req,user,onCreated)=>{log('POST','durable-upstream');await onCreated?.('durable-upstream');return Response.json({processing:true,taskId:'durable-upstream'},{status:202});};
        export const executeEnhanceGET=async req=>{const id=new URL(req.url).searchParams.get('taskId');log('GET',id);if(process.env.QA_CRASH==='1')process.exit(17);return Response.json({success:true,original_url:'https://example.com/process-recovered.png'});};`,
      "server-only": "",
    };
    await esbuild.build({
      entryPoints: ["lib/video-canvas/task-worker.ts"],
      outfile: ".qa/video-task-db/worker.cjs",
      bundle: true,
      platform: "node",
      format: "cjs",
      packages: "external",
      alias: { "@": process.cwd() },
      plugins: [
        {
          name: "mock-provider-only",
          setup(build) {
            build.onResolve({ filter: /.*/ }, (args) =>
              stubs[args.path] !== undefined ? { path: args.path, namespace: "qa" } : undefined,
            );
            build.onLoad({ filter: /.*/, namespace: "qa" }, (args) => ({
              contents: stubs[args.path],
              loader: "js",
            }));
          },
        },
      ],
    });
    const restartTask = await store.createCanvasGenerationTask(user, {
      ...enhanceRequest,
      requestId: crypto.randomUUID(),
      node: { ...enhanceRequest.node, id: "node-process-restart" },
    });
    const childCode = `require(${JSON.stringify(path.resolve(".qa/video-task-db/worker.cjs"))}).runCanvasGenerationQueue(${JSON.stringify(user)},${JSON.stringify(project)}).then(()=>process.exit(0),e=>{console.error(e);process.exit(1)});`;
    const runChild = (crash) =>
      new Promise((resolve, reject) => {
        require("node:child_process").execFile(
          process.execPath,
          ["-e", childCode],
          { env: { ...process.env, QA_CRASH: crash ? "1" : "0" }, timeout: 30000 },
          (error, stdout, stderr) => {
            if (error && !(crash && error.code === 17)) reject(Error(stderr || String(error)));
            else resolve();
          },
        );
      });
    await runChild(true);
    const crashed = (await store.listCanvasGenerationTasks(user, project)).find(
      (t) => t.id === restartTask.id,
    );
    assert.equal(crashed.status, "running");
    assert.equal(crashed.result.upstreamTask.taskId, "durable-upstream");
    await client.query(
      "UPDATE \"CanvasCapabilityTask\" SET \"updatedAt\"=(now() at time zone 'UTC')-interval '6 minutes' WHERE id=$1",
      [restartTask.id],
    );
    assert.equal(
      (await store.listCanvasGenerationTasks(user, project)).find((t) => t.id === restartTask.id)
        .status,
      "interrupted",
    );
    await client.query(
      "UPDATE \"CanvasCapabilityTask\" SET \"updatedAt\"=(now() at time zone 'UTC')-interval '6 minutes' WHERE id=$1",
      [restartTask.id],
    );
    await runChild(false);
    const final = (await store.listCanvasGenerationTasks(user, project)).find(
      (t) => t.id === restartTask.id,
    );
    assert.equal(final.status, "succeeded");
    assert.equal(final.attempts, 2);
    assert.equal(final.result.images[0].originalUrl, "https://example.com/process-recovered.png");
    const providerCalls = fs
      .readFileSync(callsFile, "utf8")
      .trim()
      .split("\n")
      .map((line) => JSON.parse(line));
    assert.deepEqual(
      providerCalls.map((call) => call.method),
      ["POST", "GET", "GET"],
    );
    assert(providerCalls.every((call) => call.id === "durable-upstream"));
    console.log(
      "PASS: real PostgreSQL checkpoint recovery, ten concurrent reclaimers, retry cooldown, stale-worker fencing, duplicate prevention, abrupt worker process restart with POST/GET/GET only.",
    );
    console.log(
      "PASS: real PostgreSQL ownership, idempotency (10 submits), exclusive claiming (10 workers), serial queue, cancellation CAS, durable results, hidden nodes, interrupted task without resubmission, late-result recovery.",
    );
  } finally {
    if (client) await client.end();
    await pg.stop();
  }
})().then(
  () => process.exit(0),
  (error) => {
    console.error(error);
    process.exit(1);
  },
);