ai-chatbot/lib/video-canvas/task-worker.ts

79 lines
2.9 KiB
TypeScript

import "server-only";
import { executeDirectVideoGeneration } from "@/lib/ai/tools/generate-video";
import { executeCanvasImageResponse } from "@/lib/canvas/generate-image-response";
import { executeWorkflowAnalysisResponse } from "./workflow-analysis-response";
import { requireAnalysisResult, type CanvasTaskRequest } from "./task-contract";
import { validateVideoReferences } from "./reference-input";
import { probeRemoteUrl } from "@/lib/net/url-access";
import {
claimCanvasGenerationTask,
finishCanvasGenerationTask,
heartbeatCanvasGenerationTask,
} from "./task-store";
async function execute(request: CanvasTaskRequest, userId: string) {
const params = request.params;
if (request.kind === "video") {
const checked = await validateVideoReferences(params.images, (url) =>
probeRemoteUrl(url, { timeoutMs: 8000 }),
);
if (checked.error) throw Error(checked.error);
const result = await executeDirectVideoGeneration({
...params,
prompt: String(params.prompt || ""),
mode: params.mode as any,
images: checked.images,
});
if (!result.success || !result.videoUrl) throw Error("视频生成未返回结果");
return {
url: result.videoUrl,
metadata: {
...result.metadata,
temporaryVideoUrl: result.videoUrl,
storageStatus:
result.metadata?.permanentVideoUrl === result.videoUrl ? "archived" : "temporary",
},
modelId: result.modelId,
modelRef: result.modelRef,
};
}
const response =
request.kind === "image"
? await executeCanvasImageResponse(
{ ...params, prompt: String(params.prompt || ""), async: false },
userId,
)
: await executeWorkflowAnalysisResponse(params);
const body = await response.json();
if (!response.ok || !body.success) throw Error(body.error || "任务执行失败");
if (
request.kind === "image" &&
!body.data?.images?.some((image: any) => image.originalUrl || image.thumbnailUrl)
)
throw Error("图片生成没有返回结果");
if (request.kind === "analysis") requireAnalysisResult(body.data, String(params.mode));
return body.data;
}
export async function runCanvasGenerationQueue(userId: string, projectId: string) {
// Work is independent of the browser request. Atomic DB claiming also covers multiple processes.
for (let i = 0; i < 40; i++) {
const task = await claimCanvasGenerationTask(userId, projectId);
if (!task) return;
const timer = setInterval(
() => void heartbeatCanvasGenerationTask(task.id).catch(() => {}),
20000,
);
try {
await finishCanvasGenerationTask(task.id, await execute(task.request, userId));
} catch (error) {
await finishCanvasGenerationTask(
task.id,
undefined,
error instanceof Error ? error.message : "任务执行失败",
);
} finally {
clearInterval(timer);
}
}
}