82b4b0c260
* fix(media): 创作结果恢复 - 结果定位符缺失时回退到任务结果 首页创作生成的图片/视频在本地素材索引尚未同步时（物化结果幂等命中、换标签页、 素材同步延迟）会被收敛成“任务已结束，但生成结果暂时无法读取”：结果定位符只查 本地素材 store，查不到就返回空，于是结果既不挂载也不显示，而服务端其实已经成功。 - generationTaskMaterializedStorageKeys 在本地查不到素材时回退到结果自带的 providerArtifactRef（resource:<id>），展示地址仍由渲染层按需续签； - 携带过期签名的历史地址不再直接交给 <img>/<video>，避免首帧 403 空白。 * fix(media): 前端等待上限对齐后端超时 - 长视频不再被前端误判失败 视频任务的前端等待上限写死 32 分钟，而后端视频超时默认 60 分钟（现网配到 120 分钟）、 上游 30 秒档常态 35-45 分钟。结果：任务在后端正常跑并最终成功，前端先报 “任务执行超时，请稍后重试”，消息被标成失败；而失败态消息不会被恢复流程接管 （isRecoverableCreationMessage 只认 pending），用户看到的就是“失败”且视频永远不显示。 - taskWaitTimeoutMs：视频 32 → 125 分钟、分镜 13 → 25 分钟，都不小于后端同类超时； - isRecoverableCreationMessage：媒体消息的失败态也纳入恢复，任务确实失败时收敛结果不变， 任务其实成功时把结果补回消息。 --------- Co-authored-by: xiaomuclaw <282836183+xiaomuclaw@users.noreply.github.com>
582 lines
26 KiB
TypeScript
582 lines
26 KiB
TypeScript
import { canvasNodeToAsset, declaredCanvasNodeAssetCategory, findCanvasNodeAsset, type CanvasAssetSource } from "@/lib/canvas/canvas-node-asset";
|
||
import { canvasVideoAssetPreviewUrl } from "@/lib/canvas/canvas-media-preview";
|
||
import { readImageMeta } from "@/lib/image-utils";
|
||
import { parseBackendGenerationResult, type BackendGenerationResult } from "@/services/api/generation-task";
|
||
import { ApiError } from "@/services/api/request";
|
||
import { linkProjectAsset, moveProjectAsset, updateProjectAssetCategory } from "@/services/api/projects";
|
||
import type { GenerationTask, GenerationTaskOutput } from "@/services/api/task-center";
|
||
import { getMediaBlob, resolveMediaUrl, setMediaBlob } from "@/services/file-storage";
|
||
import { createGenerationTaskMaterializer, createIdempotentMaterializeOutput, type MaterializeGenerationTaskOutput } from "@/services/generation-task-materializer";
|
||
import { withGenerationArtifactCommitLock } from "@/services/generation-asset-repository";
|
||
import { uploadGeneratedAssetToConfiguredSources } from "@/services/external-asset-sources";
|
||
import { getImageBlob, resolveImageUrl, setImageBlob } from "@/services/image-storage";
|
||
import { generationArtifactStorageKey, loadOrStoreGenerationArtifact } from "@/services/generation-artifact-sink";
|
||
import { createProviderNeutralGenerationTaskEffectStore } from "@/services/provider-neutral-generation-effects";
|
||
import { getCachedResourceBlob } from "@/services/resource-blob-cache";
|
||
import { saveRemoteUserDataNow } from "@/services/user-data-sync";
|
||
import { getActiveUserScope } from "@/lib/user-scope";
|
||
import { normalizeAssetCategory } from "@/lib/asset-category";
|
||
import { runGenerationConsumer } from "@/services/generation-consumer-lifecycle";
|
||
import { useAssetStore, type AssetCategory, type AssetStatus, type NewAsset } from "@/stores/use-asset-store";
|
||
import type { CanvasNodeData } from "@/types/canvas";
|
||
|
||
function throwIfAborted(signal?: AbortSignal) {
|
||
if (signal?.aborted) throw new DOMException("The operation was aborted", "AbortError");
|
||
}
|
||
|
||
type EnsureCanvasNodeAssetOptions = {
|
||
canvasId: string;
|
||
domainProjectId?: string;
|
||
node: CanvasNodeData;
|
||
source: CanvasAssetSource;
|
||
taskId?: string;
|
||
category?: AssetCategory;
|
||
folderId?: string;
|
||
signal?: AbortSignal;
|
||
};
|
||
|
||
export type CanvasNodeAssetResult = {
|
||
assetId: string;
|
||
created: boolean;
|
||
linkedToProject: boolean;
|
||
};
|
||
|
||
const pendingAssetSyncs = new Map<string, Promise<CanvasNodeAssetResult>>();
|
||
const DEFAULT_RATE_LIMIT_RETRY_MS = 60_000;
|
||
const MAX_RATE_LIMIT_RETRY_MS = 5 * 60_000;
|
||
|
||
type CanvasAssetSyncRetryOptions = {
|
||
signal?: AbortSignal;
|
||
maxRetries?: number;
|
||
wait?: (delayMs: number, signal?: AbortSignal) => Promise<void>;
|
||
};
|
||
|
||
function waitForCanvasAssetSyncRetry(delayMs: number, signal?: AbortSignal) {
|
||
return new Promise<void>((resolve, reject) => {
|
||
if (signal?.aborted) {
|
||
reject(new DOMException("The operation was aborted", "AbortError"));
|
||
return;
|
||
}
|
||
const timer = globalThis.setTimeout(() => {
|
||
signal?.removeEventListener("abort", onAbort);
|
||
resolve();
|
||
}, delayMs);
|
||
const onAbort = () => {
|
||
globalThis.clearTimeout(timer);
|
||
reject(new DOMException("The operation was aborted", "AbortError"));
|
||
};
|
||
signal?.addEventListener("abort", onAbort, { once: true });
|
||
});
|
||
}
|
||
|
||
export async function retryCanvasAssetSyncAfterRateLimit<T>(operation: () => Promise<T>, options: CanvasAssetSyncRetryOptions = {}): Promise<T> {
|
||
const maxRetries = Math.max(0, options.maxRetries ?? 2);
|
||
const wait = options.wait ?? waitForCanvasAssetSyncRetry;
|
||
for (let attempt = 0; ; attempt += 1) {
|
||
throwIfAborted(options.signal);
|
||
try {
|
||
return await operation();
|
||
} catch (error) {
|
||
if (!(error instanceof ApiError) || error.status !== 429 || attempt >= maxRetries) throw error;
|
||
const delayMs = Math.min(MAX_RATE_LIMIT_RETRY_MS, Math.max(0, error.retryAfterMs ?? DEFAULT_RATE_LIMIT_RETRY_MS));
|
||
await wait(delayMs, options.signal);
|
||
}
|
||
}
|
||
}
|
||
|
||
export function ensureCanvasNodeAsset(options: EnsureCanvasNodeAssetOptions) {
|
||
const scope = getActiveUserScope();
|
||
const identity = options.taskId || options.node.metadata?.taskId || options.node.metadata?.storageKey || options.node.id;
|
||
const key = [scope, options.domainProjectId || "personal", options.canvasId, options.node.id, identity].join(":");
|
||
const pending = pendingAssetSyncs.get(key);
|
||
if (pending) return pending;
|
||
const request = runGenerationConsumer(options.signal, (signal) => persistCanvasNodeAsset({ ...options, signal })).finally(() => pendingAssetSyncs.delete(key));
|
||
pendingAssetSyncs.set(key, request);
|
||
return request;
|
||
}
|
||
|
||
async function persistCanvasNodeAsset(options: EnsureCanvasNodeAssetOptions): Promise<CanvasNodeAssetResult> {
|
||
throwIfAborted(options.signal);
|
||
const store = useAssetStore.getState();
|
||
let asset = findCanvasNodeAsset(store.assets, options.node, options.canvasId, options.taskId);
|
||
const declaredCategory = options.category || declaredCanvasNodeAssetCategory(options.node);
|
||
let created = false;
|
||
if (!asset) {
|
||
const input = canvasNodeToAsset(options.node, { canvasId: options.canvasId, source: options.source, taskId: options.taskId });
|
||
if (!input) throw new Error("当前节点没有可保存的素材内容");
|
||
const assetId = store.addAsset(options.category ? { ...input, category: options.category } : input);
|
||
asset = useAssetStore.getState().assets.find((item) => item.id === assetId);
|
||
created = true;
|
||
}
|
||
if (!asset) throw new Error("素材写入本地失败");
|
||
if (declaredCategory && asset.category !== declaredCategory) {
|
||
store.updateAsset(asset.id, { category: declaredCategory });
|
||
asset = useAssetStore.getState().assets.find((item) => item.id === asset?.id) || asset;
|
||
}
|
||
if (!options.domainProjectId) {
|
||
// 个人画布也必须在返回成功前把素材提交到服务端，不能只依赖延迟自动同步。
|
||
await saveRemoteUserDataNow();
|
||
throwIfAborted(options.signal);
|
||
return { assetId: asset.id, created, linkedToProject: false };
|
||
}
|
||
await syncAssetToProject(asset.id, options.domainProjectId, declaredCategory, options.folderId, options.signal);
|
||
return { assetId: asset.id, created, linkedToProject: true };
|
||
}
|
||
|
||
async function syncAssetToProject(assetId: string, domainProjectId: string, category?: AssetCategory, folderId?: string, signal?: AbortSignal) {
|
||
throwIfAborted(signal);
|
||
const asset = useAssetStore.getState().assets.find((candidate) => candidate.id === assetId);
|
||
if (!asset) throw new Error("素材写入本地失败");
|
||
const linkedProjectIds = Array.isArray(asset.metadata?.projectIds) ? asset.metadata.projectIds.filter((id): id is string => typeof id === "string") : [];
|
||
if (linkedProjectIds.includes(domainProjectId)) {
|
||
if (folderId !== undefined) await moveProjectAsset(domainProjectId, asset.id, folderId, signal);
|
||
return;
|
||
}
|
||
|
||
// 项目关联依赖后端 assets 记录，先强制完成素材同步，不能依赖延迟自动同步的时序。
|
||
await saveRemoteUserDataNow();
|
||
throwIfAborted(signal);
|
||
const { asset: linkedAsset } = await linkProjectAsset(
|
||
domainProjectId,
|
||
{
|
||
assetId: asset.id,
|
||
category: normalizeAssetCategory(category || asset.category),
|
||
folderId,
|
||
},
|
||
signal,
|
||
);
|
||
throwIfAborted(signal);
|
||
let linked = category && linkedAsset.category !== category ? (await updateProjectAssetCategory(domainProjectId, asset.id, category, signal)).asset : linkedAsset;
|
||
if (folderId !== undefined && (linked.folderId || "") !== folderId) linked = (await moveProjectAsset(domainProjectId, asset.id, folderId, signal)).asset;
|
||
throwIfAborted(signal);
|
||
useAssetStore.getState().updateAsset(asset.id, {
|
||
category: normalizeAssetCategory(linked.category),
|
||
status: linked.status as AssetStatus,
|
||
primaryVersionId: linked.primaryVersionId,
|
||
metadata: { ...asset.metadata, projectIds: [...new Set([...linkedProjectIds, domainProjectId])] },
|
||
});
|
||
await saveRemoteUserDataNow();
|
||
throwIfAborted(signal);
|
||
}
|
||
|
||
function generationTaskResult(task: GenerationTask): BackendGenerationResult {
|
||
if (task.resultJson) return parseBackendGenerationResult(task);
|
||
if (!task.previewUrl) return {};
|
||
return task.previewKind === "video" ? { mode: "video", video: { dataUrl: task.previewUrl } } : { mode: "image", images: [{ dataUrl: task.previewUrl }] };
|
||
}
|
||
|
||
export function projectGenerationTaskResult(task: GenerationTask, result?: BackendGenerationResult): GenerationTask {
|
||
const projectedResult = result ?? generationTaskResult(task);
|
||
const outputs: GenerationTaskOutput[] = projectedResult.images?.length
|
||
? projectedResult.images.map((image, outputIndex) => ({
|
||
outputIndex,
|
||
mediaType: "image" as const,
|
||
...(image.storageKey ? { providerArtifactRef: image.storageKey } : {}),
|
||
}))
|
||
: projectedResult.video
|
||
? [
|
||
{
|
||
outputIndex: 0,
|
||
mediaType: "video" as const,
|
||
...(projectedResult.video.storageKey ? { providerArtifactRef: projectedResult.video.storageKey } : {}),
|
||
},
|
||
]
|
||
: projectedResult.audio
|
||
? [
|
||
{
|
||
outputIndex: 0,
|
||
mediaType: "audio" as const,
|
||
...(projectedResult.audio.storageKey ? { providerArtifactRef: projectedResult.audio.storageKey } : {}),
|
||
},
|
||
]
|
||
: (task.outputs?.map((output) => ({ ...output })) ?? []);
|
||
|
||
return {
|
||
...task,
|
||
...(result ? { resultJson: JSON.stringify(result) } : {}),
|
||
outputs,
|
||
...(task.status === "succeeded" && outputs.length && !outputs.every((output) => output.materializedAssetId) ? { resultState: "PENDING_MATERIALIZATION" as const } : {}),
|
||
};
|
||
}
|
||
|
||
async function storedGenerationImage(result: NonNullable<BackendGenerationResult["images"]>[number], effectKey: string, scope: string, signal?: AbortSignal) {
|
||
throwIfAborted(signal);
|
||
if (result.storageKey) {
|
||
const url = await resolveImageUrl(result.storageKey, result.dataUrl);
|
||
if (!url) throw new Error("图片结果资源不可用");
|
||
const meta = result.width && result.height ? undefined : await readImageMeta(url, signal);
|
||
throwIfAborted(signal);
|
||
return {
|
||
url,
|
||
storageKey: result.storageKey,
|
||
width: result.width || meta?.width || 1024,
|
||
height: result.height || meta?.height || 1024,
|
||
bytes: result.bytes || 0,
|
||
mimeType: result.mimeType || "image/png",
|
||
};
|
||
}
|
||
const storageKey = generationArtifactStorageKey(effectKey, "image", scope);
|
||
const blob = await loadOrStoreGenerationArtifact({
|
||
effectKey: storageKey,
|
||
read: (key) => getImageBlob(key),
|
||
materialize: async () => (await fetch(result.dataUrl, { signal })).blob(),
|
||
write: async (key, artifact) => {
|
||
await setImageBlob(key, artifact);
|
||
},
|
||
});
|
||
throwIfAborted(signal);
|
||
const url = await resolveImageUrl(storageKey);
|
||
throwIfAborted(signal);
|
||
if (!url) throw new Error("图片结果资源不可用");
|
||
const meta = result.width && result.height ? undefined : await readImageMeta(url, signal);
|
||
throwIfAborted(signal);
|
||
return {
|
||
url,
|
||
storageKey,
|
||
width: result.width || meta?.width || 1024,
|
||
height: result.height || meta?.height || 1024,
|
||
bytes: result.bytes || blob.size,
|
||
mimeType: result.mimeType || blob.type || "image/png",
|
||
};
|
||
}
|
||
|
||
async function storedGenerationMedia(dataUrl: string, effectKey: string, mediaType: "video" | "audio", metadata: { width?: number; height?: number; durationMs?: number; bytes?: number; mimeType: string }, scope: string, signal?: AbortSignal) {
|
||
throwIfAborted(signal);
|
||
const storageKey = generationArtifactStorageKey(effectKey, mediaType, scope);
|
||
const blob = await loadOrStoreGenerationArtifact({
|
||
effectKey: storageKey,
|
||
read: (key) => getMediaBlob(key),
|
||
materialize: async () => (await fetch(dataUrl, { signal })).blob(),
|
||
write: async (key, artifact) => {
|
||
await setMediaBlob(key, artifact);
|
||
},
|
||
});
|
||
throwIfAborted(signal);
|
||
const url = await resolveMediaUrl(storageKey);
|
||
throwIfAborted(signal);
|
||
if (!url) throw new Error(`${mediaType === "video" ? "视频" : "音频"}结果资源不可用`);
|
||
return {
|
||
url,
|
||
storageKey,
|
||
width: metadata.width,
|
||
height: metadata.height,
|
||
durationMs: metadata.durationMs,
|
||
bytes: metadata.bytes || blob.size,
|
||
mimeType: metadata.mimeType || blob.type,
|
||
};
|
||
}
|
||
|
||
async function cachedRemoteGenerationVideo(video: NonNullable<BackendGenerationResult["video"]>, signal?: AbortSignal) {
|
||
if (!video.storageKey) throw new Error("生成视频缺少云端资源标识");
|
||
throwIfAborted(signal);
|
||
const blob = await getCachedResourceBlob(video.storageKey);
|
||
throwIfAborted(signal);
|
||
if (!blob) throw new Error("生成视频资源缓存失败，未标记为成功");
|
||
const url = await resolveMediaUrl(video.storageKey, video.dataUrl);
|
||
throwIfAborted(signal);
|
||
if (!url) throw new Error("生成视频资源地址为空，未标记为成功");
|
||
return {
|
||
url,
|
||
storageKey: video.storageKey,
|
||
width: video.width || 0,
|
||
height: video.height || 0,
|
||
durationMs: video.durationMs,
|
||
bytes: video.bytes || blob.size,
|
||
mimeType: video.mimeType || blob.type || "video/mp4",
|
||
};
|
||
}
|
||
|
||
async function generationOutputAsset(input: Parameters<MaterializeGenerationTaskOutput>[0], scope: string): Promise<NewAsset> {
|
||
throwIfAborted(input.signal);
|
||
const result = generationTaskResult(input.task);
|
||
const metadata = {
|
||
source: "generation-task",
|
||
generationEffectKey: input.effectKey,
|
||
taskId: input.task.id,
|
||
outputIndex: input.output.outputIndex,
|
||
conversationId: input.task.clientContext?.conversationId,
|
||
messageId: input.task.clientContext?.messageId,
|
||
batchIndex: input.task.clientContext?.batchIndex,
|
||
};
|
||
|
||
if (input.output.mediaType === "image") {
|
||
const image = result.images?.[input.output.outputIndex];
|
||
if (!image) throw new Error("生成任务缺少图片输出");
|
||
const stored = await storedGenerationImage(image, input.effectKey, scope, input.signal);
|
||
return {
|
||
kind: "image",
|
||
title: "生成图片",
|
||
coverUrl: stored.url,
|
||
tags: ["生成"],
|
||
status: "confirmed",
|
||
source: "生成任务",
|
||
metadata,
|
||
data: {
|
||
dataUrl: stored.url,
|
||
storageKey: stored.storageKey,
|
||
width: stored.width,
|
||
height: stored.height,
|
||
bytes: stored.bytes,
|
||
mimeType: stored.mimeType,
|
||
},
|
||
};
|
||
}
|
||
|
||
if (input.output.mediaType === "video") {
|
||
const video = result.video;
|
||
if (!video) throw new Error("生成任务缺少视频输出");
|
||
const stored = video.storageKey
|
||
? await cachedRemoteGenerationVideo(video, input.signal)
|
||
: await storedGenerationMedia(
|
||
video.dataUrl,
|
||
input.effectKey,
|
||
"video",
|
||
{
|
||
width: video.width,
|
||
height: video.height,
|
||
durationMs: video.durationMs,
|
||
bytes: video.bytes,
|
||
mimeType: video.mimeType || "video/mp4",
|
||
},
|
||
scope,
|
||
input.signal,
|
||
);
|
||
if (!stored.url) throw new Error("视频结果资源不可用");
|
||
return {
|
||
kind: "video",
|
||
title: "生成视频",
|
||
coverUrl: canvasVideoAssetPreviewUrl(stored.url),
|
||
tags: ["生成"],
|
||
status: "confirmed",
|
||
source: "生成任务",
|
||
metadata,
|
||
data: {
|
||
url: stored.url,
|
||
storageKey: stored.storageKey,
|
||
width: stored.width || 0,
|
||
height: stored.height || 0,
|
||
durationMs: stored.durationMs,
|
||
bytes: stored.bytes,
|
||
mimeType: stored.mimeType || "video/mp4",
|
||
},
|
||
};
|
||
}
|
||
|
||
const audio = result.audio;
|
||
if (!audio) throw new Error("生成任务缺少音频输出");
|
||
const stored = audio.storageKey
|
||
? {
|
||
url: await resolveMediaUrl(audio.storageKey, audio.dataUrl),
|
||
storageKey: audio.storageKey,
|
||
durationMs: audio.durationMs,
|
||
bytes: audio.bytes || 0,
|
||
mimeType: audio.mimeType || "audio/mpeg",
|
||
}
|
||
: await storedGenerationMedia(
|
||
audio.dataUrl,
|
||
input.effectKey,
|
||
"audio",
|
||
{
|
||
durationMs: audio.durationMs,
|
||
bytes: audio.bytes,
|
||
mimeType: audio.mimeType || "audio/mpeg",
|
||
},
|
||
scope,
|
||
input.signal,
|
||
);
|
||
if (!stored.url) throw new Error("音频结果资源不可用");
|
||
return {
|
||
kind: "audio",
|
||
title: "生成音频",
|
||
coverUrl: "",
|
||
tags: ["生成"],
|
||
status: "confirmed",
|
||
source: "生成任务",
|
||
metadata,
|
||
data: {
|
||
url: stored.url,
|
||
storageKey: stored.storageKey,
|
||
durationMs: stored.durationMs,
|
||
bytes: stored.bytes,
|
||
mimeType: stored.mimeType || "audio/mpeg",
|
||
},
|
||
};
|
||
}
|
||
|
||
const materializeGenerationOutput: MaterializeGenerationTaskOutput = createIdempotentMaterializeOutput({
|
||
async insertOrReturnAsset(input) {
|
||
const scope = getActiveUserScope();
|
||
const assetId = await withGenerationArtifactCommitLock(
|
||
scope,
|
||
async () => {
|
||
throwIfAborted(input.signal);
|
||
const asset = await generationOutputAsset(input, scope);
|
||
throwIfAborted(input.signal);
|
||
const createdAssetId = await useAssetStore.getState().addGenerationAsset(input.effectKey, asset, input.signal);
|
||
throwIfAborted(input.signal);
|
||
return createdAssetId;
|
||
},
|
||
{ requireCrossRealmLock: true },
|
||
);
|
||
const storedAsset = useAssetStore.getState().assets.find((candidate) => candidate.id === assetId);
|
||
if (storedAsset) {
|
||
const syncResult = await uploadGeneratedAssetToConfiguredSources(storedAsset, input.signal);
|
||
throwIfAborted(input.signal);
|
||
if (syncResult.records.length) {
|
||
const currentAsset = useAssetStore.getState().assets.find((candidate) => candidate.id === assetId) || storedAsset;
|
||
useAssetStore.getState().updateAsset(assetId, {
|
||
metadata: { ...currentAsset.metadata, externalSync: syncResult.records },
|
||
});
|
||
}
|
||
}
|
||
return assetId;
|
||
},
|
||
});
|
||
|
||
function remoteGenerationTaskMaterializer() {
|
||
return createGenerationTaskMaterializer({
|
||
// 每个页面实例持有自己的租约；共享浏览器 storage + Web Lock 提供跨标签原子权威。
|
||
effects: createProviderNeutralGenerationTaskEffectStore(),
|
||
materializeOutput: materializeGenerationOutput,
|
||
});
|
||
}
|
||
|
||
function generationTaskMaterializer(task: GenerationTask) {
|
||
return remoteGenerationTaskMaterializer();
|
||
}
|
||
|
||
export async function materializeGenerationTaskAssets(task: GenerationTask, signal?: AbortSignal): Promise<GenerationTask> {
|
||
return generationTaskMaterializer(task).materialize(projectGenerationTaskResult(task), signal);
|
||
}
|
||
|
||
export function attachGenerationTaskNode(task: GenerationTask, nodeId: string, outputIndex: number, consumer: Parameters<ReturnType<typeof generationTaskMaterializer>["attachNode"]>[3], signal?: AbortSignal) {
|
||
return generationTaskMaterializer(task).attachNode(task, nodeId, outputIndex, consumer, signal);
|
||
}
|
||
|
||
export function attachGenerationTaskMessage(task: GenerationTask, messageId: string, outputIndex: number, consumer: Parameters<ReturnType<typeof generationTaskMaterializer>["attachMessage"]>[3], signal?: AbortSignal) {
|
||
return generationTaskMaterializer(task).attachMessage(task, messageId, outputIndex, consumer, signal);
|
||
}
|
||
|
||
export function resumeGenerationTaskAgent(task: GenerationTask, continuationId: string, consumer: Parameters<ReturnType<typeof generationTaskMaterializer>["resumeAgent"]>[2], signal?: AbortSignal) {
|
||
return generationTaskMaterializer(task).resumeAgent(task, continuationId, consumer, signal);
|
||
}
|
||
|
||
export async function consumeGenerationTaskNode(
|
||
task: GenerationTask,
|
||
nodeId: string,
|
||
outputIndex: number,
|
||
consumer: (input: { task: GenerationTask; output: GenerationTaskOutput; effectKey: string; signal?: AbortSignal }) => Promise<void> | void,
|
||
dependencies: {
|
||
signal?: AbortSignal;
|
||
managed?: true;
|
||
materialize?: typeof materializeGenerationTaskAssets;
|
||
attachNode?: typeof attachGenerationTaskNode;
|
||
} = {},
|
||
): Promise<GenerationTask> {
|
||
if (!dependencies.managed) {
|
||
return runGenerationConsumer(dependencies.signal, (signal) => consumeGenerationTaskNode(task, nodeId, outputIndex, consumer, { ...dependencies, signal, managed: true }));
|
||
}
|
||
const materialized = await (dependencies.materialize ?? materializeGenerationTaskAssets)(task, dependencies.signal);
|
||
const output = materialized.outputs?.find((candidate) => candidate.outputIndex === outputIndex);
|
||
if (!output?.materializedAssetId) throw new Error("生成任务输出尚未物化");
|
||
await (dependencies.attachNode ?? attachGenerationTaskNode)(
|
||
materialized,
|
||
nodeId,
|
||
outputIndex,
|
||
async ({ effectKey, signal }) => {
|
||
await consumer({ task: materialized, output, effectKey, signal });
|
||
},
|
||
dependencies.signal,
|
||
);
|
||
return materialized;
|
||
}
|
||
|
||
export async function consumeGenerationTaskAgent(
|
||
task: GenerationTask,
|
||
continuationId: string,
|
||
consumer: (input: { task: GenerationTask; effectKey: string; signal?: AbortSignal }) => Promise<void> | void,
|
||
dependencies: {
|
||
signal?: AbortSignal;
|
||
managed?: true;
|
||
materialize?: typeof materializeGenerationTaskAssets;
|
||
resumeAgent?: typeof resumeGenerationTaskAgent;
|
||
} = {},
|
||
): Promise<GenerationTask> {
|
||
if (!dependencies.managed) {
|
||
return runGenerationConsumer(dependencies.signal, (signal) => consumeGenerationTaskAgent(task, continuationId, consumer, { ...dependencies, signal, managed: true }));
|
||
}
|
||
const materialized = task.outputs?.length ? await (dependencies.materialize ?? materializeGenerationTaskAssets)(task, dependencies.signal) : task;
|
||
await (dependencies.resumeAgent ?? resumeGenerationTaskAgent)(
|
||
materialized,
|
||
continuationId,
|
||
async ({ effectKey, signal }) => {
|
||
await consumer({ task: materialized, effectKey, signal });
|
||
},
|
||
dependencies.signal,
|
||
);
|
||
return materialized;
|
||
}
|
||
|
||
export async function consumeGenerationTaskMessage(
|
||
task: GenerationTask,
|
||
messageId: string,
|
||
consumer: (input: { task: GenerationTask; resultUrls: string[]; resultStorageKeys: string[]; effectKey: string; signal?: AbortSignal }) => Promise<void> | void,
|
||
dependencies: {
|
||
signal?: AbortSignal;
|
||
managed?: true;
|
||
materialize?: typeof materializeGenerationTaskAssets;
|
||
materializedUrls?: typeof generationTaskMaterializedUrls;
|
||
materializedStorageKeys?: typeof generationTaskMaterializedStorageKeys;
|
||
attachMessage?: typeof attachGenerationTaskMessage;
|
||
} = {},
|
||
): Promise<GenerationTask> {
|
||
if (!dependencies.managed) {
|
||
return runGenerationConsumer(dependencies.signal, (signal) => consumeGenerationTaskMessage(task, messageId, consumer, { ...dependencies, signal, managed: true }));
|
||
}
|
||
const materialized = await (dependencies.materialize ?? materializeGenerationTaskAssets)(task, dependencies.signal);
|
||
const resultUrls = (dependencies.materializedUrls ?? generationTaskMaterializedUrls)(materialized);
|
||
const resultStorageKeys = (dependencies.materializedStorageKeys ?? generationTaskMaterializedStorageKeys)(materialized);
|
||
const attach = dependencies.attachMessage ?? attachGenerationTaskMessage;
|
||
const outputs = materialized.outputs?.filter((output) => output.materializedAssetId) ?? [];
|
||
for (const output of outputs) {
|
||
await attach(
|
||
materialized,
|
||
messageId,
|
||
output.outputIndex,
|
||
async ({ effectKey, signal }) => {
|
||
await consumer({ task: materialized, resultUrls, resultStorageKeys, effectKey, signal });
|
||
},
|
||
dependencies.signal,
|
||
);
|
||
}
|
||
return materialized;
|
||
}
|
||
|
||
export function generationTaskMaterializedUrls(task: GenerationTask): string[] {
|
||
const assets = useAssetStore.getState().assets;
|
||
return (task.outputs || []).flatMap((output) => {
|
||
const asset = output.materializedAssetId ? assets.find((candidate) => candidate.id === output.materializedAssetId) : undefined;
|
||
if (!asset) return [];
|
||
if (asset.kind === "image") return [asset.data.dataUrl || asset.coverUrl].filter(Boolean);
|
||
if (asset.kind === "video" || asset.kind === "audio") return [asset.data.url].filter(Boolean);
|
||
return [];
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 返回生成结果的稳定素材定位符。对于远程资源是 resource:<id>，对于本地降级素材是
|
||
* 本地 IndexedDB storageKey。它可以持久化到历史记录，不能被短时效签名 URL 替代。
|
||
*/
|
||
export function generationTaskMaterializedStorageKeys(task: GenerationTask): string[] {
|
||
const assets = useAssetStore.getState().assets;
|
||
return (task.outputs || []).flatMap((output) => {
|
||
const asset = output.materializedAssetId ? assets.find((candidate) => candidate.id === output.materializedAssetId) : undefined;
|
||
const storageKey = asset && (asset.kind === "image" || asset.kind === "video" || asset.kind === "audio") ? asset.data.storageKey : "";
|
||
if (storageKey) return [storageKey];
|
||
// 物化结果复用时（幂等命中、换标签页、素材尚未同步）本地索引里可能没有这条素材，
|
||
// 结果自己带着稳定的 resource:<id>；丢掉它会把已经生成成功的结果判成无法读取，
|
||
// 展示地址随后仍由渲染层按需续签。
|
||
return output.providerArtifactRef ? [output.providerArtifactRef] : [];
|
||
});
|
||
}
|