// 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); 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); }, );