feat(canvas): deliver image previews before background archiving and remove generation caps
This commit is contained in:
134
parent
c3f5903a86
commit
50569ca88a
@ -153,8 +153,9 @@ PRODUCT_PLANNING_DB_SSL=false
|
||||
# Home Performance / Queues
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
CANVAS_IMAGE_TASK_CONCURRENCY=3
|
||||
CANVAS_IMAGE_TASK_BATCH_DISPATCH_INTERVAL_MS=1000
|
||||
# Legacy image queue fields: 0 = unrestricted; model/user admission caps are disabled.
|
||||
CANVAS_IMAGE_TASK_CONCURRENCY=0
|
||||
CANVAS_IMAGE_TASK_BATCH_DISPATCH_INTERVAL_MS=0
|
||||
EXTERNAL_IMAGE_API_MAX_CONCURRENT=3
|
||||
EXTERNAL_IMAGE_API_MAX_QUEUE=20
|
||||
EXTERNAL_IMAGE_API_QUEUE_TIMEOUT_MS=120000
|
||||
|
||||
@ -1078,139 +1078,15 @@ export function LocalAiRuntimeSections({
|
||||
</Button>
|
||||
</LocalAiSectionShell>
|
||||
|
||||
{/* 全站生图单模型并发 */}
|
||||
<LocalAiSectionShell
|
||||
description={
|
||||
<>
|
||||
控制全站生图链路中，同一个模型允许同时运行的任务数，以及排队后下一批任务的发起间隔。配置保存到数据库，异步队列和直调生成都会自动读取，无需重启服务。
|
||||
</>
|
||||
}
|
||||
description="生图任务直接发起，图片先显示，后台转存到存储桶。"
|
||||
expanded={isLocalAiSectionExpanded("canvas-image-task-queue")}
|
||||
id="local-ai-section-canvas-image-task-queue"
|
||||
onToggle={() => toggleLocalAiSection("canvas-image-task-queue")}
|
||||
summary={
|
||||
<LocalAiSectionSummary
|
||||
items={[
|
||||
{
|
||||
label: `${canvasImageTaskQueueConfig.concurrency} 并发/模型`,
|
||||
tone: "neutral",
|
||||
},
|
||||
{
|
||||
label: `${(
|
||||
canvasImageTaskQueueConfig.batchDispatchIntervalMs / 1000
|
||||
).toFixed(1)} 秒补位`,
|
||||
tone: "neutral",
|
||||
},
|
||||
{ label: "全站生图生效", tone: "enabled" },
|
||||
]}
|
||||
/>
|
||||
}
|
||||
title={
|
||||
<span className="flex items-center gap-2">
|
||||
<ImageIcon className="h-5 w-5" />
|
||||
全站生图单模型并发与批次间隔
|
||||
</span>
|
||||
}
|
||||
summary={<LocalAiSectionSummary items={[{ label: "并发不限", tone: "enabled" }, { label: "即时发起", tone: "enabled" }]} />}
|
||||
title={<span className="flex items-center gap-2"><ImageIcon className="h-5 w-5" />生图任务执行</span>}
|
||||
>
|
||||
<div className="grid gap-6 lg:grid-cols-[minmax(0,1fr)_minmax(220px,320px)]">
|
||||
<div className="rounded-lg border border-amber-200 bg-amber-50 p-4 text-amber-900 text-sm">
|
||||
<div className="font-medium">生效范围</div>
|
||||
<div className="mt-2 space-y-1 text-amber-800 text-xs leading-5">
|
||||
<p>
|
||||
•
|
||||
影响画布异步生图、智能体直调生图、视频分镜高清预览等全站生图入口。
|
||||
</p>
|
||||
<p>
|
||||
•
|
||||
多用户共享同一套并发限制，但会按模型分别计数，不同模型之间不再互相阻塞。
|
||||
</p>
|
||||
<p>
|
||||
• 当前配置的是单模型并发，不是全站总并发，也不是单用户并发。
|
||||
</p>
|
||||
<p>
|
||||
•
|
||||
批次间隔只控制排队补位的“下一批”发起节奏，不影响首批任务即时启动。
|
||||
</p>
|
||||
<p>
|
||||
• 数值过高可能触发上游模型 429、超时或 504，建议配合模型 Key
|
||||
池继续细分。
|
||||
</p>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="space-y-4">
|
||||
<div className="space-y-2">
|
||||
<Label htmlFor="canvasImageTaskQueueConcurrency">
|
||||
单模型最大同时运行任务数
|
||||
</Label>
|
||||
<Input
|
||||
id="canvasImageTaskQueueConcurrency"
|
||||
max={MAX_CANVAS_IMAGE_TASK_CONCURRENCY}
|
||||
min={MIN_CANVAS_IMAGE_TASK_CONCURRENCY}
|
||||
onChange={(e) =>
|
||||
setCanvasImageTaskQueueConfig(
|
||||
normalizeCanvasImageTaskQueueConfig({
|
||||
...canvasImageTaskQueueConfig,
|
||||
concurrency: e.target.value,
|
||||
})
|
||||
)
|
||||
}
|
||||
type="number"
|
||||
value={canvasImageTaskQueueConfig.concurrency}
|
||||
/>
|
||||
<div className="text-muted-foreground text-xs">
|
||||
可配置范围 {MIN_CANVAS_IMAGE_TASK_CONCURRENCY}-
|
||||
{MAX_CANVAS_IMAGE_TASK_CONCURRENCY}，当前为{" "}
|
||||
{canvasImageTaskQueueConfig.concurrency} 并发，默认值为{" "}
|
||||
{DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY}。
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="space-y-2">
|
||||
<Label htmlFor="canvasImageTaskBatchDispatchIntervalSeconds">
|
||||
下一批排队任务发起间隔（秒）
|
||||
</Label>
|
||||
<Input
|
||||
id="canvasImageTaskBatchDispatchIntervalSeconds"
|
||||
max={MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000}
|
||||
min={MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000}
|
||||
onChange={(e) =>
|
||||
setCanvasImageTaskQueueConfig(
|
||||
normalizeCanvasImageTaskQueueConfig({
|
||||
...canvasImageTaskQueueConfig,
|
||||
batchDispatchIntervalMs: Math.round(
|
||||
Number.parseFloat(e.target.value || "0") * 1000
|
||||
),
|
||||
})
|
||||
)
|
||||
}
|
||||
step="0.5"
|
||||
type="number"
|
||||
value={String(
|
||||
canvasImageTaskQueueConfig.batchDispatchIntervalMs / 1000
|
||||
)}
|
||||
/>
|
||||
<div className="text-muted-foreground text-xs">
|
||||
可配置范围 {MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000}-
|
||||
{MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000} 秒，当前为{" "}
|
||||
{(
|
||||
canvasImageTaskQueueConfig.batchDispatchIntervalMs / 1000
|
||||
).toFixed(1)}{" "}
|
||||
秒，默认值为{" "}
|
||||
{(
|
||||
DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000
|
||||
).toFixed(1)}{" "}
|
||||
秒。
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="flex justify-end">
|
||||
<Button onClick={handleSaveCanvasImageTaskQueueConfig}>
|
||||
<Save className="mr-2 h-4 w-4" /> 保存并发与间隔
|
||||
</Button>
|
||||
</div>
|
||||
<p className="text-muted-foreground text-sm">已取消模型及单用户并发限制，任务无需等待其他图片生成完成。任务状态、超时处理和失败重试继续保留。</p>
|
||||
</LocalAiSectionShell>
|
||||
|
||||
{/* 批量处理配置 */}
|
||||
|
||||
@ -8906,9 +8906,7 @@ export function AdminSettingsClient() {
|
||||
{
|
||||
key: "realtime-canvas",
|
||||
title: "实时与队列",
|
||||
description: `单模型并发 ${canvasImageTaskQueueConfig.concurrency}，批次间隔 ${(
|
||||
canvasImageTaskQueueConfig.batchDispatchIntervalMs / 1000
|
||||
).toFixed(1)} 秒`,
|
||||
description: "生图并发不限，任务即时发起",
|
||||
status: realtimeCanvasConfig.enabled ? "实时已启用" : "实时已关闭",
|
||||
tone: realtimeCanvasConfig.enabled ? "enabled" : "neutral",
|
||||
icon: SlidersHorizontal,
|
||||
|
||||
65
app/api/canvas/[id]/image-archives/route.ts
Normal file
65
app/api/canvas/[id]/image-archives/route.ts
Normal file
@ -0,0 +1,65 @@
|
||||
import { NextRequest, NextResponse } from "next/server";
|
||||
import { and, eq, inArray, sql } from "drizzle-orm";
|
||||
import { auth } from "@/app/(auth)/auth";
|
||||
import { db } from "@/lib/db";
|
||||
import { canvasProject, imageArchiveTask, generatedImage, canvasImageTask } from "@/lib/db/schema";
|
||||
import { imageArchiveSourceKey } from "@/lib/storage/image-archive-db";
|
||||
import { isGeneratedImageSource } from "@/lib/storage/image-archive-urls";
|
||||
import { scheduleImageArchiveWorker } from "@/lib/storage/image-archive-worker";
|
||||
|
||||
export async function POST(request: NextRequest, { params }: { params: Promise<{ id: string }> }) {
|
||||
const session = await auth();
|
||||
if (!session?.user?.id) return NextResponse.json({ error: "未登录" }, { status: 401 });
|
||||
const { id } = await params;
|
||||
const [project] = await db
|
||||
.select({ id: canvasProject.id })
|
||||
.from(canvasProject)
|
||||
.where(and(eq(canvasProject.id, id), eq(canvasProject.userId, session.user.id)))
|
||||
.limit(1);
|
||||
if (!project) return NextResponse.json({ error: "画布不存在" }, { status: 404 });
|
||||
const body = await request.json();
|
||||
if (
|
||||
!Array.isArray(body.urls) ||
|
||||
body.urls.length > 100 ||
|
||||
body.urls.some((url: unknown) => typeof url !== "string" || !isGeneratedImageSource(url))
|
||||
) {
|
||||
return NextResponse.json({ error: "图片地址格式错误" }, { status: 400 });
|
||||
}
|
||||
const urls: string[] = [...new Set<string>(body.urls)];
|
||||
if (!urls.length) return NextResponse.json({ replacements: {}, pending: [] });
|
||||
const source = imageArchiveTask.sourceUrl;
|
||||
const rows = await db
|
||||
.select({
|
||||
sourceUrl: imageArchiveTask.sourceUrl,
|
||||
targetUrl: imageArchiveTask.targetUrl,
|
||||
status: imageArchiveTask.status,
|
||||
})
|
||||
.from(imageArchiveTask)
|
||||
.where(
|
||||
and(
|
||||
inArray(imageArchiveTask.sourceUrl, urls.map(imageArchiveSourceKey)),
|
||||
sql`(
|
||||
exists (select 1 from ${generatedImage} where ${generatedImage.userId} = ${session.user.id}
|
||||
and (${generatedImage.originalUrl} = ${source} or ${generatedImage.thumbnailUrl} = ${source}
|
||||
or ${generatedImage.originalUrl} = ${imageArchiveTask.targetUrl} or ${generatedImage.thumbnailUrl} = ${imageArchiveTask.targetUrl}))
|
||||
or exists (select 1 from ${canvasImageTask} where ${canvasImageTask.userId} = ${session.user.id}
|
||||
and (position(${source} in ${canvasImageTask.result}::text) > 0 or position(${imageArchiveTask.targetUrl} in ${canvasImageTask.result}::text) > 0))
|
||||
or exists (select 1 from ${canvasProject} where ${canvasProject.id} = ${id} and ${canvasProject.userId} = ${session.user.id}
|
||||
and (position(${source} in ${canvasProject.canvasData}::text) > 0 or position(${imageArchiveTask.targetUrl} in ${canvasProject.canvasData}::text) > 0))
|
||||
)`,
|
||||
),
|
||||
);
|
||||
const byKey = new Map(rows.map((row) => [row.sourceUrl, row]));
|
||||
const replacements: Record<string, string> = {};
|
||||
const pending: string[] = [];
|
||||
for (const url of urls) {
|
||||
const row = byKey.get(imageArchiveSourceKey(url));
|
||||
if (row?.targetUrl) replacements[url] = row.targetUrl;
|
||||
else if (row && row.status !== "abandoned") pending.push(url);
|
||||
}
|
||||
if (rows.some((row) => row.status === "pending")) scheduleImageArchiveWorker(true);
|
||||
return NextResponse.json(
|
||||
{ replacements, pending },
|
||||
{ headers: { "Cache-Control": "private, no-store" } },
|
||||
);
|
||||
}
|
||||
@ -3,6 +3,7 @@ import { auth } from '@/app/(auth)/auth';
|
||||
import { db } from '@/lib/db';
|
||||
import { canvasProject } from '@/lib/db/schema';
|
||||
import { syncCanvasNodesForProject } from '@/lib/db/queries';
|
||||
import { resolveArchivedImageUrls } from '@/lib/storage/image-archive-db';
|
||||
import { eq, and } from 'drizzle-orm';
|
||||
import {
|
||||
buildStaleCanvasSaveConflictPayload,
|
||||
@ -122,7 +123,8 @@ export async function PUT(
|
||||
);
|
||||
}
|
||||
|
||||
const { title, description, thumbnail, canvasData } = body;
|
||||
const { title, description, canvasData } = body;
|
||||
const thumbnail = await resolveArchivedImageUrls(body.thumbnail);
|
||||
const baseCanvasRevision =
|
||||
body?.baseCanvasUpdatedAt || body?.baseCanvasRevision;
|
||||
const workspaceType =
|
||||
|
||||
34
app/api/generated-image-preview/[key]/route.ts
Normal file
34
app/api/generated-image-preview/[key]/route.ts
Normal file
@ -0,0 +1,34 @@
|
||||
import { NextRequest, NextResponse } from "next/server";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { db } from "@/lib/db";
|
||||
import { imageArchiveTask } from "@/lib/db/schema";
|
||||
|
||||
// The SHA-256 URL is a capability, like the provider's signed image URL. It must be
|
||||
// readable by image nodes and reference-image providers without a session cookie.
|
||||
export async function GET(request: NextRequest, { params }: { params: Promise<{ key: string }> }) {
|
||||
const { key } = await params;
|
||||
if (!/^[a-f0-9]{64}$/.test(key)) return new NextResponse(null, { status: 404 });
|
||||
const configured =
|
||||
process.env.NEXT_PUBLIC_APP_URL || process.env.AUTH_URL || process.env.NEXTAUTH_URL;
|
||||
const origin = configured
|
||||
? new URL(configured).origin
|
||||
: `http://localhost:${process.env.PORT || "3000"}`;
|
||||
const sourceUrl = `${origin}/api/generated-image-preview/${key}`;
|
||||
const [task] = await db
|
||||
.select({ targetUrl: imageArchiveTask.targetUrl, metadata: imageArchiveTask.metadata })
|
||||
.from(imageArchiveTask)
|
||||
.where(eq(imageArchiveTask.sourceUrl, sourceUrl))
|
||||
.limit(1);
|
||||
if (task?.targetUrl) return NextResponse.redirect(task.targetUrl, 307);
|
||||
const inline = (task?.metadata as { inlineSource?: string } | null)?.inlineSource;
|
||||
const match = inline?.match(/^data:(image\/[a-z0-9.+-]+);base64,([a-z0-9+/=]+)$/i);
|
||||
if (!match) return new NextResponse(null, { status: 404 });
|
||||
return new NextResponse(Buffer.from(match[2], "base64"), {
|
||||
headers: {
|
||||
"Content-Type": match[1],
|
||||
"Cache-Control": "private, no-store",
|
||||
"X-Content-Type-Options": "nosniff",
|
||||
"Content-Security-Policy": "default-src 'none'; sandbox",
|
||||
},
|
||||
});
|
||||
}
|
||||
@ -74,6 +74,7 @@ import {
|
||||
useState,
|
||||
} from "react";
|
||||
import { toast } from "sonner";
|
||||
import { useCanvasImageArchives } from "@/hooks/use-canvas-image-archives";
|
||||
import useSWR from "swr";
|
||||
// 深色主题组件 V2 (新版布局)
|
||||
import type { AgentGenerationWorkspaceDarkV2Props } from "@/components/agents/AgentGenerationWorkspaceDarkV2";
|
||||
@ -1906,6 +1907,8 @@ export default function CanvasEditorLegacyPage({
|
||||
[isVideoMode]
|
||||
);
|
||||
|
||||
useCanvasImageArchives(id);
|
||||
|
||||
// Store state
|
||||
const projectTitle = useCanvasStore((state) => state.projectTitle);
|
||||
const viewport = useCanvasStore(canvasSelectors.viewport);
|
||||
|
||||
105
hooks/use-canvas-image-archives.ts
Normal file
105
hooks/use-canvas-image-archives.ts
Normal file
@ -0,0 +1,105 @@
|
||||
"use client";
|
||||
import { useEffect } from "react";
|
||||
import { useCanvasStore } from "@/store/canvas";
|
||||
import {
|
||||
collectImageArchiveSources,
|
||||
type ArchivedImageUrlMap,
|
||||
} from "@/lib/storage/image-archive-urls";
|
||||
|
||||
function preload(url: string, signal: AbortSignal): Promise<boolean> {
|
||||
return new Promise((resolve) => {
|
||||
const image = new Image();
|
||||
let done = false;
|
||||
const finish = (loaded: boolean) => {
|
||||
if (done) return;
|
||||
done = true;
|
||||
clearTimeout(timer);
|
||||
signal.removeEventListener("abort", cancel);
|
||||
image.onload = image.onerror = null;
|
||||
resolve(loaded);
|
||||
};
|
||||
const cancel = () => finish(false);
|
||||
const timer = setTimeout(cancel, 15_000);
|
||||
signal.addEventListener("abort", cancel, { once: true });
|
||||
image.onload = () => finish(true);
|
||||
image.onerror = cancel;
|
||||
if (signal.aborted) cancel();
|
||||
else image.src = url;
|
||||
});
|
||||
}
|
||||
|
||||
export function useCanvasImageArchives(projectId: string) {
|
||||
useEffect(() => {
|
||||
const controller = new AbortController();
|
||||
const checked = new Set<string>();
|
||||
const unknownChecks = new Map<string, number>();
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
let polling = false;
|
||||
const poll = async () => {
|
||||
if (polling || controller.signal.aborted || document.visibilityState === "hidden") return;
|
||||
const state = useCanvasStore.getState();
|
||||
if (state.projectId !== projectId) return;
|
||||
const urls = collectImageArchiveSources(state.nodes)
|
||||
.filter((url) => !checked.has(url))
|
||||
.slice(0, 100);
|
||||
if (!urls.length) return;
|
||||
polling = true;
|
||||
try {
|
||||
const response = await fetch(
|
||||
`/api/canvas/${encodeURIComponent(projectId)}/image-archives`,
|
||||
{
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ urls }),
|
||||
signal: controller.signal,
|
||||
},
|
||||
);
|
||||
if (!response.ok) return;
|
||||
const result = (await response.json()) as {
|
||||
replacements: ArchivedImageUrlMap;
|
||||
pending: string[];
|
||||
};
|
||||
const pending = new Set(result.pending);
|
||||
const ready: ArchivedImageUrlMap = {};
|
||||
await Promise.all(
|
||||
urls.map(async (url) => {
|
||||
const target = result.replacements[url];
|
||||
if (target) {
|
||||
if (await preload(target, controller.signal)) {
|
||||
ready[url] = target;
|
||||
checked.add(url);
|
||||
checked.add(target);
|
||||
}
|
||||
} else if (!pending.has(url)) {
|
||||
// A direct generation can reach the browser before its first autosave.
|
||||
// Allow ownership records to catch up before classifying a URL as untracked.
|
||||
const checks = (unknownChecks.get(url) || 0) + 1;
|
||||
unknownChecks.set(url, checks);
|
||||
if (checks >= 10) checked.add(url);
|
||||
}
|
||||
}),
|
||||
);
|
||||
if (
|
||||
!controller.signal.aborted &&
|
||||
useCanvasStore.getState().projectId === projectId &&
|
||||
Object.keys(ready).length
|
||||
) {
|
||||
useCanvasStore.getState().replaceArchivedImageUrls(ready);
|
||||
}
|
||||
} catch {
|
||||
// Temporary network errors keep provider previews and are retried on the next poll.
|
||||
} finally {
|
||||
polling = false;
|
||||
}
|
||||
};
|
||||
const tick = () => {
|
||||
void poll();
|
||||
timer = setTimeout(tick, 3_000);
|
||||
};
|
||||
timer = setTimeout(tick, 1_000);
|
||||
return () => {
|
||||
controller.abort();
|
||||
if (timer) clearTimeout(timer);
|
||||
};
|
||||
}, [projectId]);
|
||||
}
|
||||
@ -357,14 +357,12 @@ async function uploadImageBuffer(buffer: Buffer, mimeType: string) {
|
||||
}
|
||||
|
||||
async function archiveOutputImage(imageUrl: string): Promise<UploadedImage> {
|
||||
const immediatePreview = await createDeferredUploadedImage(imageUrl, { providerType: "atlascloud" });
|
||||
if (immediatePreview) return immediatePreview;
|
||||
if (imageUrl.startsWith("data:")) {
|
||||
const parsed = parseDataUrl(imageUrl);
|
||||
return uploadImageBuffer(parsed.buffer, parsed.mimeType);
|
||||
}
|
||||
const deferred = await createDeferredUploadedImage(imageUrl, {
|
||||
providerType: "atlascloud",
|
||||
});
|
||||
if (deferred) return deferred;
|
||||
const response = await fetch(imageUrl, {
|
||||
cache: "no-store",
|
||||
redirect: "follow",
|
||||
|
||||
@ -329,15 +329,13 @@ async function uploadImageBuffer(
|
||||
}
|
||||
|
||||
async function archiveOutputImage(imageUrl: string): Promise<UploadedImage> {
|
||||
const immediatePreview = await createDeferredUploadedImage(imageUrl, { providerType: "bananaimage" });
|
||||
if (immediatePreview) return immediatePreview;
|
||||
if (imageUrl.startsWith("data:")) {
|
||||
const parsed = parseDataUrl(imageUrl);
|
||||
return uploadImageBuffer(parsed.buffer, parsed.mimeType);
|
||||
}
|
||||
|
||||
const deferred = await createDeferredUploadedImage(imageUrl, {
|
||||
providerType: "bananaimage",
|
||||
});
|
||||
if (deferred) return deferred;
|
||||
|
||||
const response = await fetch(imageUrl, {
|
||||
cache: "no-store",
|
||||
|
||||
@ -119,6 +119,9 @@ async function inferImageAspectRatioFromUrl(url: string): Promise<string | null>
|
||||
* 将 base64 图片上传到 S3，返回缩略图和原图的 URL
|
||||
*/
|
||||
async function uploadBase64ToS3(base64Data: string, mimeType: string = "image/jpeg"): Promise<{ thumbnail: string; original: string }> {
|
||||
const deferred = await createDeferredUploadedImage(`data:${mimeType || "image/png"};base64,${base64Data}`, { providerType: "gemini-image" });
|
||||
if (deferred) return deferred;
|
||||
|
||||
try {
|
||||
// 解析 base64 数据
|
||||
const buffer = Buffer.from(base64Data, "base64");
|
||||
|
||||
@ -62,6 +62,9 @@ async function createThumbnail(buffer: Buffer, mimeType: string): Promise<Buffer
|
||||
* 将 base64 图片上传到 S3，返回缩略图和原图的 URL
|
||||
*/
|
||||
async function uploadBase64ToS3(base64Data: string, mimeType: string = "image/jpeg"): Promise<{ thumbnail: string; original: string }> {
|
||||
const deferred = await createDeferredUploadedImage(`data:${mimeType || "image/png"};base64,${base64Data}`, { providerType: "huanke-gemini" });
|
||||
if (deferred) return deferred;
|
||||
|
||||
try {
|
||||
// 解析 base64 数据
|
||||
const buffer = Buffer.from(base64Data, "base64");
|
||||
|
||||
@ -430,6 +430,9 @@ async function downloadAndUploadImage(imageUrl: string): Promise<UploadedImage>
|
||||
}
|
||||
|
||||
async function uploadBase64ToS3(base64Data: string, mimeType: string) {
|
||||
const deferred = await createDeferredUploadedImage(`data:${mimeType || "image/png"};base64,${base64Data}`, { providerType: "openai-image" });
|
||||
if (deferred) return deferred;
|
||||
|
||||
const buffer = Buffer.from(base64Data, "base64");
|
||||
const extension = detectImageExtension("", mimeType);
|
||||
return uploadGeneratedBuffer(
|
||||
|
||||
@ -665,6 +665,9 @@ function generateFilename(prefix: string, extension: string) {
|
||||
}
|
||||
|
||||
async function uploadImageBuffer(buffer: Buffer, extension: string): Promise<UploadedImage> {
|
||||
const mime = extension === "jpg" || extension === "jpeg" ? "image/jpeg" : `image/${extension}`;
|
||||
const deferred = await createDeferredUploadedImage(`data:${mime};base64,${buffer.toString("base64")}`, { providerType: "snapgen-image" });
|
||||
if (deferred) return deferred;
|
||||
const thumbnailBuffer = await createThumbnail(buffer);
|
||||
|
||||
const originalFilename = generateFilename("snapgen-image-orig", extension);
|
||||
|
||||
@ -177,6 +177,9 @@ async function downloadAndUploadImage(imageUrl: string): Promise<UploadedImage>
|
||||
}
|
||||
|
||||
async function uploadBase64ToS3(base64Data: string, mimeType: string) {
|
||||
const deferred = await createDeferredUploadedImage(`data:${mimeType || "image/png"};base64,${base64Data}`, { providerType: "stepfun-image" });
|
||||
if (deferred) return deferred;
|
||||
|
||||
const buffer = Buffer.from(base64Data, "base64");
|
||||
const extension = detectImageExtension("", mimeType);
|
||||
return uploadGeneratedBuffer(
|
||||
|
||||
@ -511,14 +511,12 @@ async function uploadBuffer(buffer: Buffer, mimeType: string) {
|
||||
}
|
||||
|
||||
async function archiveOutputImage(imageUrl: string): Promise<UploadedImage> {
|
||||
const immediatePreview = await createDeferredUploadedImage(imageUrl, { providerType: "vip98k" });
|
||||
if (immediatePreview) return immediatePreview;
|
||||
if (imageUrl.startsWith("data:")) {
|
||||
const parsed = parseDataUrl(imageUrl);
|
||||
return uploadBuffer(parsed.buffer, parsed.mimeType);
|
||||
}
|
||||
const deferred = await createDeferredUploadedImage(imageUrl, {
|
||||
providerType: "vip98k",
|
||||
});
|
||||
if (deferred) return deferred;
|
||||
const response = await fetch(imageUrl, {
|
||||
cache: "no-store",
|
||||
redirect: "follow",
|
||||
|
||||
@ -269,15 +269,13 @@ async function uploadImageBuffer(
|
||||
}
|
||||
|
||||
async function archiveOutputImage(imageUrl: string): Promise<UploadedImage> {
|
||||
const immediatePreview = await createDeferredUploadedImage(imageUrl, { providerType: "visionary" });
|
||||
if (immediatePreview) return immediatePreview;
|
||||
if (imageUrl.startsWith("data:")) {
|
||||
const parsed = parseDataUrl(imageUrl);
|
||||
return uploadImageBuffer(parsed.buffer, parsed.mimeType);
|
||||
}
|
||||
|
||||
const deferred = await createDeferredUploadedImage(imageUrl, {
|
||||
providerType: "visionary",
|
||||
});
|
||||
if (deferred) return deferred;
|
||||
|
||||
const response = await fetch(imageUrl, {
|
||||
cache: "no-store",
|
||||
|
||||
@ -428,9 +428,9 @@ async function createThumbnail(buffer: Buffer, mimeType: string): Promise<Buffer
|
||||
/**
|
||||
* 将图片 URL 或 base64 上传到 S3，返回缩略图和原图的 URL
|
||||
*/
|
||||
async function uploadImageToS3(imageData: string): Promise<{ thumbnail: string; original: string }> {
|
||||
async function uploadImageToS3(imageData: string, deferArchive = true): Promise<{ thumbnail: string; original: string }> {
|
||||
try {
|
||||
if (/^https?:\/\//i.test(imageData)) {
|
||||
if (deferArchive && /^(?:https?:\/\/|data:image\/)/i.test(imageData)) {
|
||||
const deferred = await createDeferredUploadedImage(imageData, {
|
||||
providerType: "volcano-vision",
|
||||
});
|
||||
@ -653,7 +653,7 @@ async function convertMessagesToVolcanoFormat(messages: any[]): Promise<{ prompt
|
||||
if (imageUrl.startsWith('data:image')) {
|
||||
// Base64格式：需要先上传到S3获取URL
|
||||
logger.debug("[Volcano Vision] Found Base64 image, uploading to S3...");
|
||||
const uploaded = await uploadImageToS3(imageUrl);
|
||||
const uploaded = await uploadImageToS3(imageUrl, false);
|
||||
const rewritten = await rewriteUploadPublicUrlForVolcano(uploaded.original);
|
||||
processedImages.push(rewritten);
|
||||
} else {
|
||||
|
||||
@ -1,3 +1,4 @@
|
||||
import { resolveArchivedImageUrls } from "@/lib/storage/image-archive-db";
|
||||
import { db } from "@/lib/db";
|
||||
import { generatedImage } from "@/lib/db/schema";
|
||||
import {
|
||||
@ -122,6 +123,7 @@ export async function saveGeneratedImages(
|
||||
savedOriginalUrls: string[];
|
||||
}> {
|
||||
const imageRegex = /!\[([^\]]*)\]\(([^)]+)\)/g;
|
||||
imageMarkdown = await resolveArchivedImageUrls(imageMarkdown);
|
||||
const matches = [...imageMarkdown.matchAll(imageRegex)];
|
||||
|
||||
let savedImageCount = 0;
|
||||
|
||||
@ -1,192 +1,40 @@
|
||||
import "server-only";
|
||||
|
||||
import { getSystemSetting } from "@/lib/db/queries";
|
||||
import {
|
||||
CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY,
|
||||
DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS,
|
||||
DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY,
|
||||
type CanvasImageTaskQueueConfig,
|
||||
normalizeCanvasImageTaskQueueConfig,
|
||||
} from "./image-task-config";
|
||||
import { normalizeCanvasImageTaskQueueConfig } from "./image-task-config";
|
||||
|
||||
const CONCURRENCY_CONFIG_CACHE_MS = 5 * 1000;
|
||||
const RECENT_EXECUTION_SAMPLE_LIMIT = 24;
|
||||
const DEFAULT_ESTIMATED_WAIT_MS = 45 * 1000;
|
||||
const concurrencyStateKey = "__canvas_image_model_concurrency__";
|
||||
|
||||
interface ModelWaiter {
|
||||
ticketId: string;
|
||||
ownerKey: string;
|
||||
resolve: () => void;
|
||||
}
|
||||
|
||||
interface ModelConcurrencyState {
|
||||
running: Map<string, string>;
|
||||
waiters: ModelWaiter[];
|
||||
dispatchTimer: ReturnType<typeof setTimeout> | null;
|
||||
delayedUntil: number | null;
|
||||
recentExecutionDurationsMs: number[];
|
||||
}
|
||||
|
||||
interface CanvasImageConcurrencyState {
|
||||
models: Map<string, ModelConcurrencyState>;
|
||||
}
|
||||
|
||||
let cachedConfig:
|
||||
| {
|
||||
expiresAt: number;
|
||||
value: CanvasImageTaskQueueConfig;
|
||||
}
|
||||
| null = null;
|
||||
|
||||
function getGlobalConcurrencyState(): CanvasImageConcurrencyState {
|
||||
const stateKey = "__unrestricted_canvas_image_models__";
|
||||
type ModelState = { running: Set<string>; durations: number[] };
|
||||
function getModels() {
|
||||
const globalState = globalThis as typeof globalThis & {
|
||||
[concurrencyStateKey]?: CanvasImageConcurrencyState;
|
||||
[stateKey]?: Map<string, ModelState>;
|
||||
};
|
||||
|
||||
if (!globalState[concurrencyStateKey]) {
|
||||
globalState[concurrencyStateKey] = {
|
||||
models: new Map<string, ModelConcurrencyState>(),
|
||||
};
|
||||
}
|
||||
|
||||
return globalState[concurrencyStateKey] as CanvasImageConcurrencyState;
|
||||
return (globalState[stateKey] ??= new Map());
|
||||
}
|
||||
|
||||
function getOrCreateModelState(modelKey: string) {
|
||||
const state = getGlobalConcurrencyState();
|
||||
const existing = state.models.get(modelKey);
|
||||
if (existing) {
|
||||
return existing;
|
||||
function getModel(key: string) {
|
||||
const models = getModels();
|
||||
const normalized = normalizeCanvasImageModelKey(key);
|
||||
let state = models.get(normalized);
|
||||
if (!state) {
|
||||
state = { running: new Set(), durations: [] };
|
||||
models.set(normalized, state);
|
||||
}
|
||||
|
||||
const created: ModelConcurrencyState = {
|
||||
running: new Map<string, string>(),
|
||||
waiters: [],
|
||||
dispatchTimer: null,
|
||||
delayedUntil: null,
|
||||
recentExecutionDurationsMs: [],
|
||||
};
|
||||
state.models.set(modelKey, created);
|
||||
return created;
|
||||
return state;
|
||||
}
|
||||
|
||||
function clearModelDispatchTimer(modelState: ModelConcurrencyState) {
|
||||
if (modelState.dispatchTimer) {
|
||||
clearTimeout(modelState.dispatchTimer);
|
||||
modelState.dispatchTimer = null;
|
||||
}
|
||||
|
||||
modelState.delayedUntil = null;
|
||||
}
|
||||
|
||||
function cleanupModelState(modelKey: string) {
|
||||
const state = getGlobalConcurrencyState();
|
||||
const modelState = state.models.get(modelKey);
|
||||
if (!modelState) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (
|
||||
modelState.running.size === 0 &&
|
||||
modelState.waiters.length === 0 &&
|
||||
modelState.recentExecutionDurationsMs.length === 0
|
||||
) {
|
||||
clearModelDispatchTimer(modelState);
|
||||
state.models.delete(modelKey);
|
||||
}
|
||||
}
|
||||
|
||||
export function normalizeCanvasImageModelKey(value: unknown) {
|
||||
if (typeof value !== "string") {
|
||||
return "__default__";
|
||||
}
|
||||
|
||||
const normalized = value.trim();
|
||||
return normalized || "__default__";
|
||||
return typeof value === "string" && value.trim() ? value.trim() : "__default__";
|
||||
}
|
||||
|
||||
function normalizeCanvasImageOwnerKey(value: unknown) {
|
||||
if (typeof value !== "string") {
|
||||
return "__anonymous__";
|
||||
}
|
||||
|
||||
const normalized = value.trim();
|
||||
return normalized || "__anonymous__";
|
||||
export async function getCanvasImageTaskQueueConfig() {
|
||||
return normalizeCanvasImageTaskQueueConfig(undefined);
|
||||
}
|
||||
|
||||
function getMaxConcurrentPerOwner(maxConcurrent: number) {
|
||||
return Math.max(1, Math.ceil(Math.max(1, maxConcurrent) / 2));
|
||||
export async function getCanvasImageSingleModelConcurrency() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
function getRunningCountForOwner(
|
||||
modelState: ModelConcurrencyState,
|
||||
ownerKey: string
|
||||
) {
|
||||
let count = 0;
|
||||
for (const runningOwnerKey of modelState.running.values()) {
|
||||
if (runningOwnerKey === ownerKey) {
|
||||
count += 1;
|
||||
}
|
||||
}
|
||||
return count;
|
||||
export async function getCanvasImageBatchDispatchIntervalMs() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
export async function getCanvasImageTaskQueueConfig(): Promise<CanvasImageTaskQueueConfig> {
|
||||
const now = Date.now();
|
||||
if (cachedConfig && cachedConfig.expiresAt > now) {
|
||||
return cachedConfig.value;
|
||||
}
|
||||
|
||||
const envConcurrency = process.env.CANVAS_IMAGE_TASK_CONCURRENCY;
|
||||
const envBatchDispatchIntervalMs = process.env.CANVAS_IMAGE_TASK_BATCH_DISPATCH_INTERVAL_MS;
|
||||
let config = normalizeCanvasImageTaskQueueConfig(undefined, {
|
||||
envConcurrency,
|
||||
envBatchDispatchIntervalMs,
|
||||
});
|
||||
|
||||
try {
|
||||
const setting = await getSystemSetting({
|
||||
key: CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY,
|
||||
});
|
||||
if (setting?.value !== undefined && setting.value !== null) {
|
||||
config = normalizeCanvasImageTaskQueueConfig(setting.value, {
|
||||
envConcurrency,
|
||||
envBatchDispatchIntervalMs,
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn("[Canvas Image Concurrency] Failed to load DB config, using fallback", {
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
}
|
||||
|
||||
cachedConfig = {
|
||||
expiresAt: now + CONCURRENCY_CONFIG_CACHE_MS,
|
||||
value: {
|
||||
concurrency: config.concurrency || DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY,
|
||||
batchDispatchIntervalMs: Number.isFinite(config.batchDispatchIntervalMs)
|
||||
? config.batchDispatchIntervalMs
|
||||
: DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS,
|
||||
},
|
||||
};
|
||||
|
||||
return cachedConfig.value;
|
||||
}
|
||||
|
||||
export async function getCanvasImageSingleModelConcurrency(): Promise<number> {
|
||||
return (await getCanvasImageTaskQueueConfig()).concurrency;
|
||||
}
|
||||
|
||||
export async function getCanvasImageBatchDispatchIntervalMs(): Promise<number> {
|
||||
return (await getCanvasImageTaskQueueConfig()).batchDispatchIntervalMs;
|
||||
}
|
||||
|
||||
export function getCanvasImageRunningCountForModel(modelKey: string) {
|
||||
const normalizedModelKey = normalizeCanvasImageModelKey(modelKey);
|
||||
return getGlobalConcurrencyState().models.get(normalizedModelKey)?.running.size || 0;
|
||||
return getModels().get(normalizeCanvasImageModelKey(modelKey))?.running.size || 0;
|
||||
}
|
||||
|
||||
export type CanvasImageModelQueueStatus = {
|
||||
status: "pending" | "running" | "succeeded" | "failed";
|
||||
queuePosition: number;
|
||||
@ -197,143 +45,27 @@ export type CanvasImageModelQueueStatus = {
|
||||
source: "model_queue";
|
||||
};
|
||||
|
||||
function getRecentAverageExecutionMs(modelState: ModelConcurrencyState) {
|
||||
const samples = modelState.recentExecutionDurationsMs.filter(
|
||||
(value) => Number.isFinite(value) && value > 0
|
||||
);
|
||||
if (samples.length === 0) {
|
||||
return DEFAULT_ESTIMATED_WAIT_MS;
|
||||
}
|
||||
const total = samples.reduce((sum, value) => sum + value, 0);
|
||||
return Math.max(1000, Math.round(total / samples.length));
|
||||
}
|
||||
|
||||
function buildModelQueueSnapshot(params: {
|
||||
modelState: ModelConcurrencyState;
|
||||
queuePosition: number;
|
||||
sameModelPendingCount: number;
|
||||
sameModelRunningCount: number;
|
||||
concurrencyLimit: number;
|
||||
}): CanvasImageModelQueueStatus {
|
||||
const averageExecutionMs = getRecentAverageExecutionMs(params.modelState);
|
||||
const estimatedWaitMs =
|
||||
params.sameModelPendingCount <= 0
|
||||
? 0
|
||||
: Math.max(
|
||||
averageExecutionMs,
|
||||
Math.ceil((params.queuePosition + 1) / Math.max(1, params.concurrencyLimit)) *
|
||||
averageExecutionMs
|
||||
);
|
||||
|
||||
return {
|
||||
status: "pending",
|
||||
queuePosition: Math.max(0, params.queuePosition),
|
||||
sameModelPendingCount: Math.max(0, params.sameModelPendingCount),
|
||||
sameModelRunningCount: Math.max(0, params.sameModelRunningCount),
|
||||
concurrencyLimit: Math.max(1, params.concurrencyLimit),
|
||||
estimatedWaitMs,
|
||||
source: "model_queue",
|
||||
};
|
||||
}
|
||||
|
||||
export async function getCanvasImageModelQueueStatus(params: {
|
||||
modelKey: string;
|
||||
}): Promise<CanvasImageModelQueueStatus> {
|
||||
const normalizedModelKey = normalizeCanvasImageModelKey(params.modelKey);
|
||||
const { concurrency: maxConcurrent } = await getCanvasImageTaskQueueConfig();
|
||||
const modelState =
|
||||
getGlobalConcurrencyState().models.get(normalizedModelKey) || {
|
||||
running: new Map<string, string>(),
|
||||
waiters: [],
|
||||
dispatchTimer: null,
|
||||
delayedUntil: null,
|
||||
recentExecutionDurationsMs: [],
|
||||
};
|
||||
return buildModelQueueSnapshot({
|
||||
modelState,
|
||||
queuePosition: modelState.waiters.length > 0 ? modelState.waiters.length - 1 : 0,
|
||||
sameModelPendingCount: modelState.waiters.length,
|
||||
sameModelRunningCount: modelState.running.size,
|
||||
concurrencyLimit: maxConcurrent,
|
||||
});
|
||||
}
|
||||
|
||||
export function recordCanvasImageModelExecutionDuration(
|
||||
modelKey: string,
|
||||
durationMs: number
|
||||
) {
|
||||
if (!Number.isFinite(durationMs) || durationMs <= 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const normalizedModelKey = normalizeCanvasImageModelKey(modelKey);
|
||||
const modelState = getOrCreateModelState(normalizedModelKey);
|
||||
modelState.recentExecutionDurationsMs.push(durationMs);
|
||||
if (modelState.recentExecutionDurationsMs.length > RECENT_EXECUTION_SAMPLE_LIMIT) {
|
||||
modelState.recentExecutionDurationsMs.splice(
|
||||
0,
|
||||
modelState.recentExecutionDurationsMs.length - RECENT_EXECUTION_SAMPLE_LIMIT
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async function drainModelWaiters(
|
||||
modelKey: string,
|
||||
trigger: "arrival" | "slot-release" | "timer"
|
||||
) {
|
||||
const normalizedModelKey = normalizeCanvasImageModelKey(modelKey);
|
||||
const modelState = getGlobalConcurrencyState().models.get(normalizedModelKey);
|
||||
if (!modelState || modelState.waiters.length === 0) {
|
||||
cleanupModelState(normalizedModelKey);
|
||||
return;
|
||||
}
|
||||
|
||||
const { concurrency: maxConcurrent, batchDispatchIntervalMs } =
|
||||
await getCanvasImageTaskQueueConfig();
|
||||
if (modelState.running.size >= maxConcurrent) {
|
||||
return;
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
if (modelState.delayedUntil && modelState.delayedUntil > now) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (trigger === "slot-release" && batchDispatchIntervalMs > 0) {
|
||||
if (!modelState.dispatchTimer) {
|
||||
modelState.delayedUntil = now + batchDispatchIntervalMs;
|
||||
modelState.dispatchTimer = setTimeout(() => {
|
||||
const state = getGlobalConcurrencyState().models.get(normalizedModelKey);
|
||||
if (!state) return;
|
||||
clearModelDispatchTimer(state);
|
||||
void drainModelWaiters(normalizedModelKey, "timer");
|
||||
}, batchDispatchIntervalMs);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
clearModelDispatchTimer(modelState);
|
||||
while (modelState.running.size < maxConcurrent && modelState.waiters.length > 0) {
|
||||
const maxConcurrentPerOwner = getMaxConcurrentPerOwner(maxConcurrent);
|
||||
const waiterIndex = modelState.waiters.findIndex(
|
||||
(item) =>
|
||||
getRunningCountForOwner(modelState, item.ownerKey) < maxConcurrentPerOwner
|
||||
);
|
||||
if (waiterIndex < 0) {
|
||||
break;
|
||||
}
|
||||
|
||||
const [waiter] = modelState.waiters.splice(waiterIndex, 1);
|
||||
if (!waiter) {
|
||||
break;
|
||||
}
|
||||
modelState.running.set(waiter.ticketId, waiter.ownerKey);
|
||||
waiter.resolve();
|
||||
}
|
||||
|
||||
cleanupModelState(normalizedModelKey);
|
||||
return {
|
||||
status: "running",
|
||||
queuePosition: 0,
|
||||
sameModelPendingCount: 0,
|
||||
sameModelRunningCount: getCanvasImageRunningCountForModel(params.modelKey),
|
||||
concurrencyLimit: 0,
|
||||
estimatedWaitMs: 0,
|
||||
source: "model_queue",
|
||||
};
|
||||
}
|
||||
export function recordCanvasImageModelExecutionDuration(modelKey: string, durationMs: number) {
|
||||
if (!Number.isFinite(durationMs) || durationMs <= 0) return;
|
||||
const state = getModel(modelKey);
|
||||
state.durations.push(durationMs);
|
||||
if (state.durations.length > 24) state.durations.shift();
|
||||
}
|
||||
|
||||
// Keep execution tracking/release semantics, without model or owner admission limits.
|
||||
export async function acquireCanvasImageModelSlot(params: {
|
||||
modelKey: string;
|
||||
ownerKey?: string;
|
||||
@ -341,86 +73,15 @@ export async function acquireCanvasImageModelSlot(params: {
|
||||
debugMeta?: Record<string, unknown>;
|
||||
onQueued?: (status: CanvasImageModelQueueStatus) => void | Promise<void>;
|
||||
}) {
|
||||
const normalizedModelKey = normalizeCanvasImageModelKey(params.modelKey);
|
||||
const normalizedOwnerKey = normalizeCanvasImageOwnerKey(params.ownerKey);
|
||||
const {
|
||||
concurrency: maxConcurrent,
|
||||
batchDispatchIntervalMs,
|
||||
} = await getCanvasImageTaskQueueConfig();
|
||||
const modelState = getOrCreateModelState(normalizedModelKey);
|
||||
const ticketId = `${normalizedModelKey}:${Date.now()}:${Math.random().toString(36).slice(2, 10)}`;
|
||||
const waitingSince = Date.now();
|
||||
const hasActiveDelay =
|
||||
modelState.delayedUntil !== null && modelState.delayedUntil > waitingSince;
|
||||
const maxConcurrentPerOwner = getMaxConcurrentPerOwner(maxConcurrent);
|
||||
|
||||
if (
|
||||
modelState.waiters.length === 0 &&
|
||||
!hasActiveDelay &&
|
||||
modelState.running.size < maxConcurrent &&
|
||||
getRunningCountForOwner(modelState, normalizedOwnerKey) < maxConcurrentPerOwner
|
||||
) {
|
||||
modelState.running.set(ticketId, normalizedOwnerKey);
|
||||
return {
|
||||
release: async () => {
|
||||
const state = getGlobalConcurrencyState().models.get(normalizedModelKey);
|
||||
if (!state) return;
|
||||
state.running.delete(ticketId);
|
||||
await drainModelWaiters(normalizedModelKey, "slot-release");
|
||||
},
|
||||
waitMs: 0,
|
||||
maxConcurrent,
|
||||
batchDispatchIntervalMs,
|
||||
};
|
||||
}
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
const queuePosition = modelState.waiters.length;
|
||||
modelState.waiters.push({
|
||||
ticketId,
|
||||
ownerKey: normalizedOwnerKey,
|
||||
resolve,
|
||||
});
|
||||
if (typeof params.onQueued === "function") {
|
||||
const queueStatus = buildModelQueueSnapshot({
|
||||
modelState,
|
||||
queuePosition,
|
||||
sameModelPendingCount: modelState.waiters.length,
|
||||
sameModelRunningCount: modelState.running.size,
|
||||
concurrencyLimit: maxConcurrent,
|
||||
});
|
||||
Promise.resolve(params.onQueued(queueStatus)).catch((error) => {
|
||||
console.warn("[Canvas Image Concurrency] onQueued callback failed:", {
|
||||
debugLabel: params.debugLabel || "canvas-image",
|
||||
modelKey: normalizedModelKey,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
...params.debugMeta,
|
||||
});
|
||||
});
|
||||
}
|
||||
void drainModelWaiters(normalizedModelKey, "arrival");
|
||||
});
|
||||
|
||||
const waitMs = Date.now() - waitingSince;
|
||||
if (waitMs > 250) {
|
||||
console.info("[Canvas Image Concurrency] Waited for model slot:", {
|
||||
debugLabel: params.debugLabel || "canvas-image",
|
||||
modelKey: normalizedModelKey,
|
||||
waitMs,
|
||||
batchDispatchIntervalMs,
|
||||
...params.debugMeta,
|
||||
});
|
||||
}
|
||||
|
||||
const state = getModel(params.modelKey);
|
||||
const ticket = crypto.randomUUID();
|
||||
state.running.add(ticket);
|
||||
return {
|
||||
release: async () => {
|
||||
const state = getGlobalConcurrencyState().models.get(normalizedModelKey);
|
||||
if (!state) return;
|
||||
state.running.delete(ticketId);
|
||||
await drainModelWaiters(normalizedModelKey, "slot-release");
|
||||
state.running.delete(ticket);
|
||||
},
|
||||
waitMs,
|
||||
maxConcurrent,
|
||||
batchDispatchIntervalMs,
|
||||
waitMs: 0,
|
||||
maxConcurrent: 0,
|
||||
batchDispatchIntervalMs: 0,
|
||||
};
|
||||
}
|
||||
|
||||
@ -1,130 +1,21 @@
|
||||
export const CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY = "canvas_image_task_queue_config";
|
||||
|
||||
export const MIN_CANVAS_IMAGE_TASK_CONCURRENCY = 1;
|
||||
export const DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY = 3;
|
||||
export const MAX_CANVAS_IMAGE_TASK_CONCURRENCY = 100;
|
||||
// Zero means unrestricted. Legacy DB/environment caps must not re-enable queueing.
|
||||
export const MIN_CANVAS_IMAGE_TASK_CONCURRENCY = 0;
|
||||
export const DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY = 0;
|
||||
export const MAX_CANVAS_IMAGE_TASK_CONCURRENCY = 0;
|
||||
export const MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS = 0;
|
||||
export const DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS = 0;
|
||||
export const MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS = 60 * 1000;
|
||||
export const MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS = 0;
|
||||
|
||||
export interface CanvasImageTaskQueueConfig {
|
||||
concurrency: number;
|
||||
batchDispatchIntervalMs: number;
|
||||
}
|
||||
|
||||
function clampConcurrency(value: number) {
|
||||
return Math.min(
|
||||
MAX_CANVAS_IMAGE_TASK_CONCURRENCY,
|
||||
Math.max(MIN_CANVAS_IMAGE_TASK_CONCURRENCY, Math.floor(value))
|
||||
);
|
||||
}
|
||||
|
||||
function clampBatchDispatchIntervalMs(value: number) {
|
||||
return Math.min(
|
||||
MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS,
|
||||
Math.max(MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS, Math.floor(value))
|
||||
);
|
||||
}
|
||||
|
||||
function parseConcurrencyCandidate(value: unknown): number | null {
|
||||
if (typeof value === "number" && Number.isFinite(value)) {
|
||||
return clampConcurrency(value);
|
||||
}
|
||||
|
||||
if (typeof value === "string" && value.trim()) {
|
||||
const parsed = Number.parseInt(value.trim(), 10);
|
||||
if (Number.isFinite(parsed)) {
|
||||
return clampConcurrency(parsed);
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
function parseBatchDispatchIntervalCandidate(
|
||||
value: unknown,
|
||||
options?: { treatAsSeconds?: boolean }
|
||||
): number | null {
|
||||
const normalize = (candidate: number) =>
|
||||
clampBatchDispatchIntervalMs(
|
||||
options?.treatAsSeconds ? candidate * 1000 : candidate
|
||||
);
|
||||
|
||||
if (typeof value === "number" && Number.isFinite(value)) {
|
||||
return normalize(value);
|
||||
}
|
||||
|
||||
if (typeof value === "string" && value.trim()) {
|
||||
const parsed = Number.parseFloat(value.trim());
|
||||
if (Number.isFinite(parsed)) {
|
||||
return normalize(parsed);
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
export function normalizeCanvasImageTaskQueueConfig(
|
||||
value: unknown,
|
||||
options?: { envConcurrency?: unknown; envBatchDispatchIntervalMs?: unknown }
|
||||
_value: unknown,
|
||||
_options?: { envConcurrency?: unknown; envBatchDispatchIntervalMs?: unknown },
|
||||
): CanvasImageTaskQueueConfig {
|
||||
const raw = value && typeof value === "object" && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
: {};
|
||||
|
||||
const concurrencyCandidates = [
|
||||
raw.concurrency,
|
||||
raw.maxConcurrency,
|
||||
raw.maxConcurrent,
|
||||
typeof value === "number" || typeof value === "string" ? value : undefined,
|
||||
options?.envConcurrency,
|
||||
];
|
||||
|
||||
let concurrency = DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY;
|
||||
for (const candidate of concurrencyCandidates) {
|
||||
const parsed = parseConcurrencyCandidate(candidate);
|
||||
if (parsed !== null) {
|
||||
concurrency = parsed;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
const intervalCandidates = [
|
||||
raw.batchDispatchIntervalMs,
|
||||
raw.dispatchIntervalMs,
|
||||
raw.batchDispatchDelayMs,
|
||||
raw.batchDispatchInterval,
|
||||
raw.dispatchInterval,
|
||||
options?.envBatchDispatchIntervalMs,
|
||||
];
|
||||
const intervalSecondsCandidates = [
|
||||
raw.batchDispatchIntervalSeconds,
|
||||
raw.dispatchIntervalSeconds,
|
||||
];
|
||||
|
||||
let batchDispatchIntervalMs = DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS;
|
||||
for (const candidate of intervalCandidates) {
|
||||
const parsed = parseBatchDispatchIntervalCandidate(candidate);
|
||||
if (parsed !== null) {
|
||||
batchDispatchIntervalMs = parsed;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (batchDispatchIntervalMs === DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS) {
|
||||
for (const candidate of intervalSecondsCandidates) {
|
||||
const parsed = parseBatchDispatchIntervalCandidate(candidate, {
|
||||
treatAsSeconds: true,
|
||||
});
|
||||
if (parsed !== null) {
|
||||
batchDispatchIntervalMs = parsed;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
concurrency,
|
||||
batchDispatchIntervalMs,
|
||||
};
|
||||
return { concurrency: 0, batchDispatchIntervalMs: 0 };
|
||||
}
|
||||
|
||||
@ -1,10 +1,10 @@
|
||||
import "server-only";
|
||||
|
||||
import { and, asc, desc, eq, lt, notInArray, or, sql } from "drizzle-orm";
|
||||
import { resolveArchivedImageUrls } from "@/lib/storage/image-archive-db";
|
||||
import { and, asc, eq, lt, notInArray, or, sql } from "drizzle-orm";
|
||||
import { db } from "@/lib/db";
|
||||
import { canvasImageTask } from "@/lib/db/schema";
|
||||
import {
|
||||
getCanvasImageBatchDispatchIntervalMs,
|
||||
getCanvasImageRunningCountForModel,
|
||||
getCanvasImageSingleModelConcurrency,
|
||||
normalizeCanvasImageModelKey,
|
||||
@ -78,10 +78,6 @@ interface QueueState {
|
||||
pendingSet: Set<string>;
|
||||
running: Map<string, RunningTaskEntry>;
|
||||
runningByModel: Map<string, Set<string>>;
|
||||
runningByModelUser: Map<string, Set<string>>;
|
||||
nextBatchEligibleAt: Map<string, number>;
|
||||
nextBatchTimers: Map<string, ReturnType<typeof setTimeout>>;
|
||||
awaitingNextBatch: Set<string>;
|
||||
lastStaleRunningCleanupAt: number;
|
||||
draining: boolean;
|
||||
}
|
||||
@ -91,9 +87,6 @@ const SCALE_ADJUST_RUNNING_TIMEOUT_MS = 12 * 60 * 1000;
|
||||
const LOCAL_ACTIVE_RUNNING_HARD_TIMEOUT_MS = 30 * 60 * 1000;
|
||||
const STALE_RUNNING_CLEANUP_INTERVAL_MS = 60 * 1000;
|
||||
const PENDING_LOOKAHEAD_LIMIT = 100;
|
||||
const QUEUE_STATUS_PENDING_LOOKAHEAD_LIMIT = 500;
|
||||
const QUEUE_STATUS_RECENT_SAMPLE_LIMIT = 120;
|
||||
const DEFAULT_ESTIMATED_WAIT_MS = 45 * 1000;
|
||||
const queueStateKey = "__canvas_image_task_queue_state__";
|
||||
|
||||
function getQueueState(): QueueState {
|
||||
@ -106,10 +99,6 @@ function getQueueState(): QueueState {
|
||||
pendingSet: new Set<string>(),
|
||||
running: new Map<string, RunningTaskEntry>(),
|
||||
runningByModel: new Map<string, Set<string>>(),
|
||||
runningByModelUser: new Map<string, Set<string>>(),
|
||||
nextBatchEligibleAt: new Map<string, number>(),
|
||||
nextBatchTimers: new Map<string, ReturnType<typeof setTimeout>>(),
|
||||
awaitingNextBatch: new Set<string>(),
|
||||
lastStaleRunningCleanupAt: 0,
|
||||
draining: false,
|
||||
};
|
||||
@ -193,14 +182,6 @@ function isStyleProTaskRequest(request: unknown): request is StyleProTaskRequest
|
||||
);
|
||||
}
|
||||
|
||||
function getModelUserKey(modelKey: string, userId: string) {
|
||||
return `${modelKey}::${userId || "__anonymous__"}`;
|
||||
}
|
||||
|
||||
function getMaxConcurrentPerUser(maxConcurrentPerModel: number) {
|
||||
return Math.max(1, Math.ceil(Math.max(1, maxConcurrentPerModel) / 2));
|
||||
}
|
||||
|
||||
async function getTaskQueueMeta(taskId: string) {
|
||||
const rows = await db
|
||||
.select({
|
||||
@ -218,27 +199,12 @@ async function getTaskQueueMeta(taskId: string) {
|
||||
};
|
||||
}
|
||||
|
||||
function getRunningCountForModel(modelKey: string) {
|
||||
const state = getQueueState();
|
||||
return state.runningByModel.get(modelKey)?.size || 0;
|
||||
}
|
||||
|
||||
function getRunningCountForModelUser(modelKey: string, userId: string) {
|
||||
const state = getQueueState();
|
||||
return state.runningByModelUser.get(getModelUserKey(modelKey, userId))?.size || 0;
|
||||
}
|
||||
|
||||
function markRunningTask(taskId: string, modelKey: string, userId: string) {
|
||||
const state = getQueueState();
|
||||
state.running.set(taskId, { modelKey, userId });
|
||||
const tasks = state.runningByModel.get(modelKey) || new Set<string>();
|
||||
tasks.add(taskId);
|
||||
state.runningByModel.set(modelKey, tasks);
|
||||
|
||||
const modelUserKey = getModelUserKey(modelKey, userId);
|
||||
const userTasks = state.runningByModelUser.get(modelUserKey) || new Set<string>();
|
||||
userTasks.add(taskId);
|
||||
state.runningByModelUser.set(modelUserKey, userTasks);
|
||||
}
|
||||
|
||||
function releaseRunningTask(taskId: string) {
|
||||
@ -249,7 +215,7 @@ function releaseRunningTask(taskId: string) {
|
||||
}
|
||||
|
||||
state.running.delete(taskId);
|
||||
const { modelKey, userId } = entry;
|
||||
const { modelKey } = entry;
|
||||
const tasks = state.runningByModel.get(modelKey);
|
||||
if (tasks) {
|
||||
tasks.delete(taskId);
|
||||
@ -258,65 +224,9 @@ function releaseRunningTask(taskId: string) {
|
||||
}
|
||||
}
|
||||
|
||||
const modelUserKey = getModelUserKey(modelKey, userId);
|
||||
const userTasks = state.runningByModelUser.get(modelUserKey);
|
||||
if (userTasks) {
|
||||
userTasks.delete(taskId);
|
||||
if (userTasks.size === 0) {
|
||||
state.runningByModelUser.delete(modelUserKey);
|
||||
}
|
||||
}
|
||||
|
||||
return modelKey;
|
||||
}
|
||||
|
||||
function isModelBatchDispatchDelayed(modelKey: string) {
|
||||
const state = getQueueState();
|
||||
const delayedUntil = state.nextBatchEligibleAt.get(modelKey);
|
||||
if (!delayedUntil) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (delayedUntil <= Date.now()) {
|
||||
state.nextBatchEligibleAt.delete(modelKey);
|
||||
const timer = state.nextBatchTimers.get(modelKey);
|
||||
if (timer) {
|
||||
clearTimeout(timer);
|
||||
state.nextBatchTimers.delete(modelKey);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
function markModelAwaitingNextBatch(modelKey: string) {
|
||||
getQueueState().awaitingNextBatch.add(modelKey);
|
||||
}
|
||||
|
||||
function scheduleModelNextBatch(modelKey: string, delayMs: number) {
|
||||
const state = getQueueState();
|
||||
if (delayMs <= 0) {
|
||||
state.awaitingNextBatch.delete(modelKey);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (state.nextBatchTimers.has(modelKey)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
state.awaitingNextBatch.delete(modelKey);
|
||||
state.nextBatchEligibleAt.set(modelKey, Date.now() + delayMs);
|
||||
const timer = setTimeout(() => {
|
||||
const latestState = getQueueState();
|
||||
latestState.nextBatchTimers.delete(modelKey);
|
||||
latestState.nextBatchEligibleAt.delete(modelKey);
|
||||
void drainCanvasImageTaskQueue();
|
||||
}, delayMs);
|
||||
state.nextBatchTimers.set(modelKey, timer);
|
||||
return true;
|
||||
}
|
||||
|
||||
async function enqueueLocalTask(
|
||||
taskId: string,
|
||||
meta?: { modelKey?: string; userId?: string } | string
|
||||
@ -343,42 +253,14 @@ async function enqueueLocalTask(
|
||||
});
|
||||
state.pendingSet.add(taskId);
|
||||
|
||||
const maxConcurrentPerModel = await getCanvasImageSingleModelConcurrency();
|
||||
if (
|
||||
getRunningCountForModel(resolvedModelKey) >= maxConcurrentPerModel ||
|
||||
getRunningCountForModelUser(resolvedModelKey, resolvedUserId) >=
|
||||
getMaxConcurrentPerUser(maxConcurrentPerModel)
|
||||
) {
|
||||
markModelAwaitingNextBatch(resolvedModelKey);
|
||||
}
|
||||
}
|
||||
|
||||
function hasModelCapacity(
|
||||
modelKey: string,
|
||||
userId: string,
|
||||
maxConcurrentPerModel: number
|
||||
) {
|
||||
return (
|
||||
getRunningCountForModel(modelKey) < maxConcurrentPerModel &&
|
||||
getRunningCountForModelUser(modelKey, userId) <
|
||||
getMaxConcurrentPerUser(maxConcurrentPerModel)
|
||||
);
|
||||
}
|
||||
|
||||
async function claimNextPendingTask(
|
||||
maxConcurrentPerModel: number
|
||||
): Promise<PendingTaskEntry | null> {
|
||||
async function claimNextPendingTask(): Promise<PendingTaskEntry | null> {
|
||||
const state = getQueueState();
|
||||
for (let index = 0; index < state.pending.length; index += 1) {
|
||||
const entry = state.pending[index];
|
||||
if (!entry) continue;
|
||||
if (state.running.has(entry.taskId)) continue;
|
||||
if (isModelBatchDispatchDelayed(entry.modelKey)) continue;
|
||||
if (!hasModelCapacity(entry.modelKey, entry.userId, maxConcurrentPerModel)) {
|
||||
markModelAwaitingNextBatch(entry.modelKey);
|
||||
continue;
|
||||
}
|
||||
|
||||
state.pending.splice(index, 1);
|
||||
state.pendingSet.delete(entry.taskId);
|
||||
return entry;
|
||||
@ -405,15 +287,7 @@ async function claimNextPendingTask(
|
||||
if (state.pendingSet.has(row.id) || state.running.has(row.id)) continue;
|
||||
|
||||
const modelKey = normalizeTaskModelKey(row.request);
|
||||
if (isModelBatchDispatchDelayed(modelKey)) {
|
||||
continue;
|
||||
}
|
||||
const userId = String(row.userId || "__anonymous__");
|
||||
if (!hasModelCapacity(modelKey, userId, maxConcurrentPerModel)) {
|
||||
markModelAwaitingNextBatch(modelKey);
|
||||
continue;
|
||||
}
|
||||
|
||||
return {
|
||||
taskId: row.id,
|
||||
modelKey,
|
||||
@ -585,7 +459,7 @@ async function runSingleTask(entry: PendingTaskEntry) {
|
||||
.update(canvasImageTask)
|
||||
.set({
|
||||
status: "succeeded",
|
||||
result: result as CanvasImageTaskResult,
|
||||
result: await resolveArchivedImageUrls(result) as CanvasImageTaskResult,
|
||||
error: null,
|
||||
completedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
@ -611,7 +485,7 @@ async function runSingleTask(entry: PendingTaskEntry) {
|
||||
and(eq(canvasImageTask.id, entry.taskId), eq(canvasImageTask.status, "running"))
|
||||
);
|
||||
} finally {
|
||||
const modelKey = releaseRunningTask(entry.taskId);
|
||||
releaseRunningTask(entry.taskId);
|
||||
let nextBatchTaskId: string | null = null;
|
||||
if (completedBatchRequest?.batchId) {
|
||||
try {
|
||||
@ -628,12 +502,6 @@ async function runSingleTask(entry: PendingTaskEntry) {
|
||||
});
|
||||
}
|
||||
}
|
||||
if (modelKey) {
|
||||
const batchDispatchIntervalMs = await getCanvasImageBatchDispatchIntervalMs();
|
||||
if (getQueueState().awaitingNextBatch.has(modelKey)) {
|
||||
scheduleModelNextBatch(modelKey, batchDispatchIntervalMs);
|
||||
}
|
||||
}
|
||||
if (nextBatchTaskId) {
|
||||
await enqueueLocalTask(nextBatchTaskId);
|
||||
}
|
||||
@ -648,9 +516,8 @@ export async function drainCanvasImageTaskQueue() {
|
||||
state.draining = true;
|
||||
try {
|
||||
await markStaleDatabaseRunningTasksFailed();
|
||||
const maxConcurrentPerModel = await getCanvasImageSingleModelConcurrency();
|
||||
while (true) {
|
||||
const nextTask = await claimNextPendingTask(maxConcurrentPerModel);
|
||||
const nextTask = await claimNextPendingTask();
|
||||
if (!nextTask) break;
|
||||
if (state.running.has(nextTask.taskId)) continue;
|
||||
markRunningTask(nextTask.taskId, nextTask.modelKey, nextTask.userId);
|
||||
@ -658,6 +525,11 @@ export async function drainCanvasImageTaskQueue() {
|
||||
}
|
||||
} finally {
|
||||
state.draining = false;
|
||||
// An arrival can occur while the final database scan is in flight. Dispatch it
|
||||
// immediately rather than relying on a later status poll to wake the queue.
|
||||
if (state.pending.some((entry) => !state.running.has(entry.taskId))) {
|
||||
void drainCanvasImageTaskQueue();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -811,58 +683,6 @@ export async function createCanvasImageTask(params: {
|
||||
return existing;
|
||||
}
|
||||
|
||||
async function estimateModelQueueWaitMs(params: {
|
||||
modelKey: string;
|
||||
sameModelRunningCount: number;
|
||||
queuePosition: number;
|
||||
concurrencyLimit: number;
|
||||
}) {
|
||||
const rows = await db
|
||||
.select({
|
||||
request: canvasImageTask.request,
|
||||
startedAt: canvasImageTask.startedAt,
|
||||
completedAt: canvasImageTask.completedAt,
|
||||
})
|
||||
.from(canvasImageTask)
|
||||
.where(eq(canvasImageTask.status, "succeeded"))
|
||||
.orderBy(desc(canvasImageTask.completedAt))
|
||||
.limit(QUEUE_STATUS_RECENT_SAMPLE_LIMIT);
|
||||
|
||||
const durations = rows
|
||||
.filter((row) => normalizeTaskModelKey(row.request) === params.modelKey)
|
||||
.map((row) => {
|
||||
if (!row.startedAt || !row.completedAt) return 0;
|
||||
return row.completedAt.getTime() - row.startedAt.getTime();
|
||||
})
|
||||
.filter((value) => Number.isFinite(value) && value > 1000)
|
||||
.slice(0, 12);
|
||||
|
||||
const averageDurationMs =
|
||||
durations.length > 0
|
||||
? Math.round(durations.reduce((sum, value) => sum + value, 0) / durations.length)
|
||||
: DEFAULT_ESTIMATED_WAIT_MS;
|
||||
const immediatelyStartable = Math.max(
|
||||
params.concurrencyLimit - params.sameModelRunningCount,
|
||||
0
|
||||
);
|
||||
|
||||
if (params.queuePosition < immediatelyStartable) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const tasksAheadAfterImmediateSlots = Math.max(
|
||||
0,
|
||||
params.queuePosition - immediatelyStartable
|
||||
);
|
||||
const waveCount =
|
||||
Math.floor(
|
||||
tasksAheadAfterImmediateSlots / Math.max(1, params.concurrencyLimit)
|
||||
) +
|
||||
(params.sameModelRunningCount > 0 || immediatelyStartable === 0 ? 1 : 0);
|
||||
|
||||
return Math.max(0, waveCount * averageDurationMs);
|
||||
}
|
||||
|
||||
async function buildCanvasImageTaskQueueStatus(task: {
|
||||
id: string;
|
||||
status: CanvasImageTaskStatus;
|
||||
@ -882,51 +702,13 @@ async function buildCanvasImageTaskQueueStatus(task: {
|
||||
getCanvasImageRunningCountForModel(modelKey)
|
||||
);
|
||||
|
||||
if (task.status !== "pending") {
|
||||
return {
|
||||
status: task.status,
|
||||
queuePosition: 0,
|
||||
sameModelPendingCount: 0,
|
||||
sameModelRunningCount,
|
||||
concurrencyLimit,
|
||||
estimatedWaitMs: 0,
|
||||
source: "model_queue" as const,
|
||||
};
|
||||
}
|
||||
|
||||
const pendingRows = await db
|
||||
.select({
|
||||
id: canvasImageTask.id,
|
||||
request: canvasImageTask.request,
|
||||
createdAt: canvasImageTask.createdAt,
|
||||
})
|
||||
.from(canvasImageTask)
|
||||
.where(eq(canvasImageTask.status, "pending"))
|
||||
.orderBy(asc(canvasImageTask.createdAt))
|
||||
.limit(QUEUE_STATUS_PENDING_LOOKAHEAD_LIMIT);
|
||||
|
||||
const sameModelPendingRows = pendingRows.filter(
|
||||
(row) => normalizeTaskModelKey(row.request) === modelKey
|
||||
);
|
||||
const queuePosition = Math.max(
|
||||
sameModelPendingRows.findIndex((row) => row.id === task.id),
|
||||
0
|
||||
);
|
||||
const sameModelPendingCount = sameModelPendingRows.length;
|
||||
const estimatedWaitMs = await estimateModelQueueWaitMs({
|
||||
modelKey,
|
||||
sameModelRunningCount,
|
||||
queuePosition,
|
||||
concurrencyLimit,
|
||||
});
|
||||
|
||||
return {
|
||||
status: task.status,
|
||||
queuePosition,
|
||||
sameModelPendingCount,
|
||||
queuePosition: 0,
|
||||
sameModelPendingCount: 0,
|
||||
sameModelRunningCount,
|
||||
concurrencyLimit,
|
||||
estimatedWaitMs,
|
||||
estimatedWaitMs: 0,
|
||||
source: "model_queue" as const,
|
||||
};
|
||||
}
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
import "server-only";
|
||||
import { resolveArchivedImageUrls } from "@/lib/storage/image-archive-db";
|
||||
|
||||
import {
|
||||
and,
|
||||
@ -1590,6 +1591,7 @@ export async function syncCanvasNodesForProject(params: {
|
||||
viewport?: { x: number; y: number; zoom: number };
|
||||
} | null | undefined;
|
||||
}, executor: Pick<typeof db, "select" | "insert" | "delete"> = db) {
|
||||
params = { ...params, canvasData: await resolveArchivedImageUrls(params.canvasData) };
|
||||
const nodes = Array.isArray(params.canvasData?.nodes) ? params.canvasData?.nodes : [];
|
||||
const assetUrls = Array.from(
|
||||
new Set(
|
||||
|
||||
@ -3,6 +3,7 @@ import "server-only";
|
||||
import { archiveTemporaryImage } from "@/lib/storage/image-archive";
|
||||
import {
|
||||
getImageArchiveTaskStats,
|
||||
recoverStaleImageArchiveTasks,
|
||||
getImageArchiveTasksToReplace,
|
||||
getImageArchiveTasksToStore,
|
||||
markImageArchiveTaskArchiving,
|
||||
@ -26,14 +27,16 @@ export async function runImageArchiveJob() {
|
||||
const batchSize = getBatchSize();
|
||||
console.log(`[ImageArchiveJob] Start image archive job (batchSize=${batchSize})`);
|
||||
|
||||
await recoverStaleImageArchiveTasks();
|
||||
const storeTasks = await getImageArchiveTasksToStore({ limit: batchSize });
|
||||
let storeSucceeded = 0;
|
||||
let storeFailed = 0;
|
||||
|
||||
for (const task of storeTasks) {
|
||||
const storeOne = async (task: (typeof storeTasks)[number]) => {
|
||||
try {
|
||||
await markImageArchiveTaskArchiving(task.id);
|
||||
const archived = await archiveTemporaryImage(task.sourceUrl);
|
||||
if (!(await markImageArchiveTaskArchiving(task.id))) return;
|
||||
const metadata = task.metadata as { inlineSource?: string } | null;
|
||||
const archived = await archiveTemporaryImage(metadata?.inlineSource || task.sourceUrl);
|
||||
await markImageArchiveTaskStored({
|
||||
id: task.id,
|
||||
targetUrl: archived.url,
|
||||
@ -54,12 +57,13 @@ export async function runImageArchiveJob() {
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const remainingSlots = Math.max(1, batchSize - storeTasks.length);
|
||||
const replaceTasks = await getImageArchiveTasksToReplace({
|
||||
limit: remainingSlots,
|
||||
});
|
||||
// Bucket work has a small worker pool independent of unrestricted model generation.
|
||||
for (let index = 0; index < storeTasks.length; index += 4) {
|
||||
await Promise.all(storeTasks.slice(index, index + 4).map(storeOne));
|
||||
}
|
||||
const replaceTasks = await getImageArchiveTasksToReplace({ limit: 50 });
|
||||
let replaceSucceeded = 0;
|
||||
let replaceFailed = 0;
|
||||
|
||||
@ -74,7 +78,7 @@ export async function runImageArchiveJob() {
|
||||
}
|
||||
|
||||
try {
|
||||
await markImageArchiveTaskReplacing(task.id);
|
||||
if (!(await markImageArchiveTaskReplacing(task.id))) continue;
|
||||
const result = await replaceImageUrlInDatabase(task.sourceUrl, task.targetUrl);
|
||||
await markImageArchiveTaskCompleted({
|
||||
id: task.id,
|
||||
|
||||
@ -1,30 +1,34 @@
|
||||
import "server-only";
|
||||
|
||||
import { and, asc, eq, inArray, lt, lte, sql } from "drizzle-orm";
|
||||
import { createHash } from "node:crypto";
|
||||
import {
|
||||
isGeneratedImageSource,
|
||||
collectImageArchiveSources,
|
||||
replaceArchivedImageUrls,
|
||||
} from "./image-archive-urls";
|
||||
import { and, asc, eq, inArray, lt, lte, or, sql } from "drizzle-orm";
|
||||
import { db } from "@/lib/db";
|
||||
import { imageArchiveTask, type ImageArchiveTask } from "@/lib/db/schema";
|
||||
|
||||
export type ImageArchiveTaskRecord = ImageArchiveTask;
|
||||
|
||||
const DEFAULT_DISPLAY_HOLD_HOURS = 1;
|
||||
const MAX_IMAGE_ARCHIVE_ATTEMPTS = 12;
|
||||
const ARCHIVE_LEASE_MS = 30 * 60 * 1000;
|
||||
|
||||
function getDisplayHoldHours() {
|
||||
const raw = Number(
|
||||
process.env.IMAGE_ARCHIVE_DISPLAY_HOLD_HOURS || DEFAULT_DISPLAY_HOLD_HOURS
|
||||
);
|
||||
if (!Number.isFinite(raw)) return DEFAULT_DISPLAY_HOLD_HOURS;
|
||||
return Math.max(0, Math.min(24 * 30, raw));
|
||||
}
|
||||
|
||||
// Replace as soon as the durable object exists, regardless of legacy hold settings.
|
||||
export function getImageArchiveDisplayHoldMs() {
|
||||
return getDisplayHoldHours() * 60 * 60 * 1000;
|
||||
return 0;
|
||||
}
|
||||
|
||||
function isArchivableHttpUrl(value: string) {
|
||||
if (!/^https?:\/\//i.test(value)) return false;
|
||||
if (/^(data|blob):/i.test(value)) return false;
|
||||
return true;
|
||||
export function imageArchiveSourceKey(source: string) {
|
||||
if (!source.startsWith("data:")) return source;
|
||||
const configured =
|
||||
process.env.NEXT_PUBLIC_APP_URL || process.env.AUTH_URL || process.env.NEXTAUTH_URL;
|
||||
const origin = configured
|
||||
? new URL(configured).origin
|
||||
: `http://localhost:${process.env.PORT || "3000"}`;
|
||||
const key = createHash("sha256").update(source).digest("hex");
|
||||
return `${origin}/api/generated-image-preview/${key}`;
|
||||
}
|
||||
|
||||
export async function enqueueImageArchiveTask(params: {
|
||||
@ -34,30 +38,35 @@ export async function enqueueImageArchiveTask(params: {
|
||||
metadata?: Record<string, unknown> | null;
|
||||
}) {
|
||||
const sourceUrl = String(params.sourceUrl || "").trim();
|
||||
if (!isArchivableHttpUrl(sourceUrl)) return null;
|
||||
if (!isGeneratedImageSource(sourceUrl)) return null;
|
||||
|
||||
const now = new Date();
|
||||
const archiveAfterAt = params.archiveAfterAt || now;
|
||||
const replaceAfterAt =
|
||||
params.replaceAfterAt ||
|
||||
new Date(now.getTime() + getImageArchiveDisplayHoldMs());
|
||||
params.replaceAfterAt || new Date(now.getTime() + getImageArchiveDisplayHoldMs());
|
||||
|
||||
const [task] = await db
|
||||
.insert(imageArchiveTask)
|
||||
.values({
|
||||
sourceUrl,
|
||||
sourceUrl: imageArchiveSourceKey(sourceUrl),
|
||||
status: "pending",
|
||||
archiveAfterAt,
|
||||
replaceAfterAt,
|
||||
metadata: params.metadata || null,
|
||||
metadata: sourceUrl.startsWith("data:")
|
||||
? { ...params.metadata, inlineSource: sourceUrl }
|
||||
: params.metadata || null,
|
||||
updatedAt: now,
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: imageArchiveTask.sourceUrl,
|
||||
set: {
|
||||
replaceAfterAt,
|
||||
metadata: params.metadata || undefined,
|
||||
updatedAt: now,
|
||||
metadata: sql`case when ${imageArchiveTask.targetUrl} is not null then ${imageArchiveTask.metadata}
|
||||
else coalesce(${imageArchiveTask.metadata}, '{}'::jsonb) || ${JSON.stringify(
|
||||
sourceUrl.startsWith("data:")
|
||||
? { ...params.metadata, inlineSource: sourceUrl }
|
||||
: params.metadata || {},
|
||||
)}::jsonb end`,
|
||||
},
|
||||
})
|
||||
.returning();
|
||||
@ -71,10 +80,8 @@ export async function enqueueImageArchiveTasks(params: {
|
||||
}) {
|
||||
const urls = Array.from(
|
||||
new Set(
|
||||
(params.urls || [])
|
||||
.map((url) => String(url || "").trim())
|
||||
.filter(isArchivableHttpUrl)
|
||||
)
|
||||
(params.urls || []).map((url) => String(url || "").trim()).filter(isGeneratedImageSource),
|
||||
),
|
||||
);
|
||||
|
||||
const tasks: ImageArchiveTaskRecord[] = [];
|
||||
@ -97,15 +104,15 @@ export async function getImageArchiveTasksToStore(params: { limit: number }) {
|
||||
and(
|
||||
inArray(imageArchiveTask.status, ["pending", "failed"]),
|
||||
lte(imageArchiveTask.archiveAfterAt, now),
|
||||
lt(imageArchiveTask.attemptCount, MAX_IMAGE_ARCHIVE_ATTEMPTS)
|
||||
)
|
||||
lt(imageArchiveTask.attemptCount, MAX_IMAGE_ARCHIVE_ATTEMPTS),
|
||||
),
|
||||
)
|
||||
.orderBy(asc(imageArchiveTask.createdAt), asc(imageArchiveTask.id))
|
||||
.limit(Math.max(1, Math.min(50, params.limit)));
|
||||
}
|
||||
|
||||
export async function markImageArchiveTaskArchiving(id: string) {
|
||||
await db
|
||||
const claimed = await db
|
||||
.update(imageArchiveTask)
|
||||
.set({
|
||||
status: "archiving",
|
||||
@ -113,7 +120,15 @@ export async function markImageArchiveTaskArchiving(id: string) {
|
||||
lastError: null,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(imageArchiveTask.id, id));
|
||||
.where(
|
||||
and(
|
||||
eq(imageArchiveTask.id, id),
|
||||
inArray(imageArchiveTask.status, ["pending", "failed"]),
|
||||
lte(imageArchiveTask.archiveAfterAt, new Date()),
|
||||
),
|
||||
)
|
||||
.returning({ id: imageArchiveTask.id });
|
||||
return claimed.length > 0;
|
||||
}
|
||||
|
||||
export async function markImageArchiveTaskStored(params: {
|
||||
@ -128,18 +143,16 @@ export async function markImageArchiveTaskStored(params: {
|
||||
targetUrl: params.targetUrl,
|
||||
targetStorageConfigId: params.targetStorageConfigId || null,
|
||||
archivedAt: new Date(),
|
||||
replaceAfterAt: new Date(),
|
||||
lastError: null,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(imageArchiveTask.id, params.id));
|
||||
}
|
||||
|
||||
export async function markImageArchiveTaskFailed(params: {
|
||||
id: string;
|
||||
error: unknown;
|
||||
}) {
|
||||
export async function markImageArchiveTaskFailed(params: { id: string; error: unknown }) {
|
||||
const message = String(
|
||||
params.error instanceof Error ? params.error.message : params.error || "Unknown error"
|
||||
params.error instanceof Error ? params.error.message : params.error || "Unknown error",
|
||||
).slice(0, 2000);
|
||||
const isPermanentSourceError = /\bHTTP (?:404|410)\b/i.test(message);
|
||||
|
||||
@ -160,29 +173,35 @@ export async function markImageArchiveTaskFailed(params: {
|
||||
.where(eq(imageArchiveTask.id, params.id));
|
||||
}
|
||||
|
||||
function isArchiveReplacementDue(now: Date) {
|
||||
return and(
|
||||
or(
|
||||
eq(imageArchiveTask.status, "stored"),
|
||||
and(
|
||||
eq(imageArchiveTask.status, "completed"),
|
||||
sql`${imageArchiveTask.metadata}->>'replacementPasses' = '1'`,
|
||||
),
|
||||
),
|
||||
lte(imageArchiveTask.replaceAfterAt, now),
|
||||
);
|
||||
}
|
||||
|
||||
export async function getImageArchiveTasksToReplace(params: { limit: number }) {
|
||||
const now = new Date();
|
||||
return await db
|
||||
.select()
|
||||
.from(imageArchiveTask)
|
||||
.where(
|
||||
and(
|
||||
eq(imageArchiveTask.status, "stored"),
|
||||
lte(imageArchiveTask.replaceAfterAt, now)
|
||||
)
|
||||
)
|
||||
.where(isArchiveReplacementDue(new Date()))
|
||||
.orderBy(asc(imageArchiveTask.createdAt), asc(imageArchiveTask.id))
|
||||
.limit(Math.max(1, Math.min(50, params.limit)));
|
||||
}
|
||||
|
||||
export async function markImageArchiveTaskReplacing(id: string) {
|
||||
await db
|
||||
const claimed = await db
|
||||
.update(imageArchiveTask)
|
||||
.set({
|
||||
status: "replacing",
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(imageArchiveTask.id, id));
|
||||
.set({ status: "replacing", updatedAt: new Date() })
|
||||
.where(and(eq(imageArchiveTask.id, id), isArchiveReplacementDue(new Date())))
|
||||
.returning({ id: imageArchiveTask.id });
|
||||
return claimed.length > 0;
|
||||
}
|
||||
|
||||
export async function markImageArchiveTaskCompleted(params: {
|
||||
@ -195,23 +214,20 @@ export async function markImageArchiveTaskCompleted(params: {
|
||||
status: "completed",
|
||||
replacedAt: new Date(),
|
||||
lastError: null,
|
||||
metadata:
|
||||
params.replacedCount === undefined
|
||||
? undefined
|
||||
: sql`coalesce(${imageArchiveTask.metadata}, '{}'::jsonb) || ${JSON.stringify({
|
||||
replacedCount: params.replacedCount,
|
||||
})}::jsonb`,
|
||||
// Reconcile once more after delivery has settled: asynchronous generation/task
|
||||
// writes may arrive just after the first replacement. The live URL is already ready.
|
||||
replaceAfterAt: new Date(Date.now() + 60_000),
|
||||
metadata: sql`(coalesce(${imageArchiveTask.metadata}, '{}'::jsonb) - 'inlineSource')
|
||||
|| jsonb_build_object('replacementPasses', coalesce((${imageArchiveTask.metadata}->>'replacementPasses')::int, 0) + 1)
|
||||
|| ${JSON.stringify(params.replacedCount === undefined ? {} : { replacedCount: params.replacedCount })}::jsonb`,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(imageArchiveTask.id, params.id));
|
||||
}
|
||||
|
||||
export async function markImageArchiveTaskReplaceFailed(params: {
|
||||
id: string;
|
||||
error: unknown;
|
||||
}) {
|
||||
export async function markImageArchiveTaskReplaceFailed(params: { id: string; error: unknown }) {
|
||||
const message = String(
|
||||
params.error instanceof Error ? params.error.message : params.error || "Unknown error"
|
||||
params.error instanceof Error ? params.error.message : params.error || "Unknown error",
|
||||
).slice(0, 2000);
|
||||
|
||||
await db
|
||||
@ -230,6 +246,7 @@ export async function getImageArchiveTaskStats() {
|
||||
.select({
|
||||
status: imageArchiveTask.status,
|
||||
count: sql<number>`count(*)::int`,
|
||||
reconciliationCount: sql<number>`count(*) filter (where ${imageArchiveTask.status} = 'completed' and ${imageArchiveTask.metadata}->>'replacementPasses' = '1')::int`,
|
||||
})
|
||||
.from(imageArchiveTask)
|
||||
.groupBy(imageArchiveTask.status);
|
||||
@ -237,9 +254,11 @@ export async function getImageArchiveTaskStats() {
|
||||
const pendingStore = grouped
|
||||
.filter((item) => item.status === "pending" || item.status === "failed")
|
||||
.reduce((sum, item) => sum + (item.count || 0), 0);
|
||||
const pendingReplace = grouped
|
||||
.filter((item) => item.status === "stored")
|
||||
.reduce((sum, item) => sum + (item.count || 0), 0);
|
||||
const pendingReplace =
|
||||
grouped
|
||||
.filter((item) => item.status === "stored")
|
||||
.reduce((sum, item) => sum + (item.count || 0), 0) +
|
||||
grouped.reduce((sum, item) => sum + (item.reconciliationCount || 0), 0);
|
||||
|
||||
return {
|
||||
grouped,
|
||||
@ -247,3 +266,50 @@ export async function getImageArchiveTaskStats() {
|
||||
pendingReplace,
|
||||
};
|
||||
}
|
||||
|
||||
// Interrupted workers can be claimed again after a lease expires. Failed uploads keep
|
||||
// their original source and retry time, so transient bucket errors cannot lose images.
|
||||
export async function recoverStaleImageArchiveTasks() {
|
||||
const staleBefore = new Date(Date.now() - ARCHIVE_LEASE_MS);
|
||||
await db
|
||||
.update(imageArchiveTask)
|
||||
.set({
|
||||
status: sql`case when ${imageArchiveTask.attemptCount} >= ${MAX_IMAGE_ARCHIVE_ATTEMPTS} then 'abandoned' else 'failed' end`,
|
||||
archiveAfterAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(eq(imageArchiveTask.status, "archiving"), lt(imageArchiveTask.updatedAt, staleBefore)),
|
||||
);
|
||||
await db
|
||||
.update(imageArchiveTask)
|
||||
.set({ status: "stored", replaceAfterAt: new Date(), updatedAt: new Date() })
|
||||
.where(
|
||||
and(eq(imageArchiveTask.status, "replacing"), lt(imageArchiveTask.updatedAt, staleBefore)),
|
||||
);
|
||||
}
|
||||
|
||||
export async function getArchivedImageUrlMappings(urls: string[]) {
|
||||
if (!urls.length) return {};
|
||||
const sources = [...new Set(urls.filter(isGeneratedImageSource))];
|
||||
const tasks = await db
|
||||
.select({ sourceUrl: imageArchiveTask.sourceUrl, targetUrl: imageArchiveTask.targetUrl })
|
||||
.from(imageArchiveTask)
|
||||
.where(inArray(imageArchiveTask.sourceUrl, sources.map(imageArchiveSourceKey)));
|
||||
const targets = new Map(
|
||||
tasks.filter((task) => task.targetUrl).map((task) => [task.sourceUrl, task.targetUrl!]),
|
||||
);
|
||||
const replacements: Record<string, string> = {};
|
||||
for (const source of sources) {
|
||||
const target = targets.get(imageArchiveSourceKey(source));
|
||||
if (target) replacements[source] = target;
|
||||
}
|
||||
return replacements;
|
||||
}
|
||||
|
||||
// Normalize late writes too: an archive can finish before a task/canvas save arrives.
|
||||
export async function resolveArchivedImageUrls<T>(value: T): Promise<T> {
|
||||
const sources = collectImageArchiveSources(value);
|
||||
if (!sources.length) return value;
|
||||
return replaceArchivedImageUrls(value, await getArchivedImageUrlMappings(sources));
|
||||
}
|
||||
|
||||
56
lib/storage/image-archive-urls.ts
Normal file
56
lib/storage/image-archive-urls.ts
Normal file
@ -0,0 +1,56 @@
|
||||
// Shared by the server and canvas; no storage or network work happens here.
|
||||
export type ArchivedImageUrlMap = Record<string, string>;
|
||||
|
||||
export function isGeneratedImageSource(value: string) {
|
||||
return (
|
||||
/^https?:\/\//i.test(value) || /^data:image\/[a-z0-9.+-]+;base64,[a-z0-9+/=]+$/i.test(value)
|
||||
);
|
||||
}
|
||||
|
||||
export function collectImageArchiveSources(value: unknown): string[] {
|
||||
const urls = new Set<string>();
|
||||
const visit = (item: unknown) => {
|
||||
if (typeof item === "string") {
|
||||
if (isGeneratedImageSource(item)) urls.add(item);
|
||||
else
|
||||
for (const match of item.matchAll(/!\[[^\]]*\]\(([^)]+)\)/g)) {
|
||||
if (isGeneratedImageSource(match[1])) urls.add(match[1]);
|
||||
}
|
||||
} else if (Array.isArray(item)) item.forEach(visit);
|
||||
else if (item && typeof item === "object" && !(item instanceof Date))
|
||||
Object.values(item).forEach(visit);
|
||||
};
|
||||
visit(value);
|
||||
return [...urls];
|
||||
}
|
||||
|
||||
// Structural sharing preserves unaffected nodes and Date values; no undo step is added.
|
||||
export function replaceArchivedImageUrls<T>(value: T, replacements: ArchivedImageUrlMap): T {
|
||||
if (typeof value === "string") {
|
||||
if (replacements[value]) return replacements[value] as T;
|
||||
// URLs inside generated-image Markdown carry both preview and original URLs.
|
||||
return value.replace(/!\[([^\]]*)\]\(([^)]+)\)/g, (_match, alt: string, url: string) => {
|
||||
const nextAlt = alt
|
||||
.split("|||")
|
||||
.map((part) => replacements[part] || part)
|
||||
.join("|||");
|
||||
return ``;
|
||||
}) as T;
|
||||
}
|
||||
if (!value || typeof value !== "object" || value instanceof Date) return value;
|
||||
let changed = false;
|
||||
if (Array.isArray(value)) {
|
||||
const next = value.map((item) => {
|
||||
const replaced = replaceArchivedImageUrls(item, replacements);
|
||||
if (replaced !== item) changed = true;
|
||||
return replaced;
|
||||
});
|
||||
return (changed ? next : value) as T;
|
||||
}
|
||||
const next: Record<string, unknown> = {};
|
||||
for (const [key, item] of Object.entries(value)) {
|
||||
next[key] = replaceArchivedImageUrls(item, replacements);
|
||||
if (next[key] !== item) changed = true;
|
||||
}
|
||||
return (changed ? next : value) as T;
|
||||
}
|
||||
66
lib/storage/image-archive-worker.ts
Normal file
66
lib/storage/image-archive-worker.ts
Normal file
@ -0,0 +1,66 @@
|
||||
import "server-only";
|
||||
import { after } from "next/server";
|
||||
|
||||
const workerKey = "__generated_image_archive_worker__";
|
||||
type Worker = {
|
||||
running: boolean;
|
||||
scheduled: boolean;
|
||||
requested: boolean;
|
||||
retryTimer?: ReturnType<typeof setTimeout>;
|
||||
};
|
||||
function getWorker(): Worker {
|
||||
const root = globalThis as typeof globalThis & { [workerKey]?: Worker };
|
||||
return (root[workerKey] ??= { running: false, scheduled: false, requested: false });
|
||||
}
|
||||
|
||||
export function scheduleImageArchiveWorker(deliveryConfirmed = false) {
|
||||
const worker = getWorker();
|
||||
worker.requested = true;
|
||||
if (worker.running || (worker.scheduled && !deliveryConfirmed)) return;
|
||||
worker.scheduled = true;
|
||||
if (worker.retryTimer) {
|
||||
clearTimeout(worker.retryTimer);
|
||||
worker.retryTimer = undefined;
|
||||
}
|
||||
const run = async () => {
|
||||
if (worker.running) return;
|
||||
worker.scheduled = false;
|
||||
worker.running = true;
|
||||
let retryNeeded = false;
|
||||
try {
|
||||
const { runImageArchiveJob } = await import("@/lib/jobs/image-archive-job");
|
||||
do {
|
||||
worker.requested = false;
|
||||
const result = await runImageArchiveJob();
|
||||
retryNeeded = result.stats.pendingStore > 0 || result.stats.pendingReplace > 0;
|
||||
// Stop when only backoff tasks remain; do not spin while storage is down.
|
||||
if (!worker.requested && result.store.total === 0 && result.replace.total === 0) break;
|
||||
} while (true);
|
||||
} catch (error) {
|
||||
retryNeeded = true;
|
||||
console.error("[ImageArchiveWorker] background archive failed", error);
|
||||
} finally {
|
||||
worker.running = false;
|
||||
if (worker.requested) scheduleImageArchiveWorker();
|
||||
else if (retryNeeded) {
|
||||
worker.retryTimer = setTimeout(scheduleImageArchiveWorker, 60_000);
|
||||
worker.retryTimer.unref?.();
|
||||
}
|
||||
}
|
||||
};
|
||||
// Detached generation tasks may inherit a request that has already closed. Keep
|
||||
// a Node-server fallback so archiving never depends on the user leaving a tab open.
|
||||
const deliveryFallback = setTimeout(() => {
|
||||
if (worker.scheduled && !worker.running) void run();
|
||||
}, 1_000);
|
||||
deliveryFallback.unref?.();
|
||||
try {
|
||||
// Request handlers deliver first. Durable task rows also allow cron/restart recovery.
|
||||
after(run);
|
||||
} catch {
|
||||
// Durable canvas tasks can run outside Next's request context on the Node server.
|
||||
setImmediate(() => {
|
||||
void run();
|
||||
});
|
||||
}
|
||||
}
|
||||
@ -2,10 +2,13 @@ import "server-only";
|
||||
|
||||
import { randomUUID } from "node:crypto";
|
||||
import sharp from "sharp";
|
||||
import { isGeneratedImageSource } from "./image-archive-urls";
|
||||
import { scheduleImageArchiveWorker } from "./image-archive-worker";
|
||||
import { createLogger } from "@/lib/observability/logger";
|
||||
import {
|
||||
enqueueImageArchiveTasks,
|
||||
getImageArchiveDisplayHoldMs,
|
||||
imageArchiveSourceKey,
|
||||
} from "@/lib/storage/image-archive-db";
|
||||
import { uploadFile } from "@/lib/storage/service";
|
||||
|
||||
@ -25,16 +28,12 @@ function getDownloadTimeoutMs() {
|
||||
return Math.max(5_000, Math.min(5 * 60_000, Math.round(parsed)));
|
||||
}
|
||||
|
||||
function isHttpUrl(value: string) {
|
||||
return /^https?:\/\//i.test(String(value || "").trim());
|
||||
}
|
||||
|
||||
export function isDeferredImageArchiveEnabled() {
|
||||
return process.env.AI_IMAGE_DEFERRED_ARCHIVE !== "false";
|
||||
return true;
|
||||
}
|
||||
|
||||
export function isImmediateImageArchiveEnabled() {
|
||||
return process.env.AI_IMAGE_IMMEDIATE_ARCHIVE !== "false";
|
||||
return false;
|
||||
}
|
||||
|
||||
export async function registerGeneratedImageTemporaryUrls(params: {
|
||||
@ -44,24 +43,24 @@ export async function registerGeneratedImageTemporaryUrls(params: {
|
||||
if (!isDeferredImageArchiveEnabled()) return [];
|
||||
const urls = Array.from(
|
||||
new Set(
|
||||
(params.urls || [])
|
||||
.map((url) => String(url || "").trim())
|
||||
.filter(isHttpUrl)
|
||||
)
|
||||
(params.urls || []).map((url) => String(url || "").trim()).filter(isGeneratedImageSource),
|
||||
),
|
||||
);
|
||||
if (urls.length === 0) return [];
|
||||
|
||||
try {
|
||||
return await enqueueImageArchiveTasks({
|
||||
const tasks = await enqueueImageArchiveTasks({
|
||||
urls,
|
||||
metadata: {
|
||||
...(params.metadata || {}),
|
||||
displayHoldMs: getImageArchiveDisplayHoldMs(),
|
||||
},
|
||||
});
|
||||
scheduleImageArchiveWorker();
|
||||
return tasks;
|
||||
} catch (error) {
|
||||
logger.warn("enqueue temporary image archive failed", {
|
||||
urls,
|
||||
imageCount: urls.length,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
return [];
|
||||
@ -77,10 +76,8 @@ export async function formatDeferredGeneratedImageMarkdown(params: {
|
||||
}) {
|
||||
const urls = Array.from(
|
||||
new Set(
|
||||
(params.urls || [])
|
||||
.map((url) => String(url || "").trim())
|
||||
.filter(isHttpUrl)
|
||||
)
|
||||
(params.urls || []).map((url) => String(url || "").trim()).filter(isGeneratedImageSource),
|
||||
),
|
||||
);
|
||||
if (urls.length === 0) {
|
||||
throw new Error("AI 未返回可用图片地址");
|
||||
@ -97,17 +94,17 @@ export async function formatDeferredGeneratedImageMarkdown(params: {
|
||||
return persisted
|
||||
.map(
|
||||
({ originalUrl, thumbnailUrl }) =>
|
||||
``
|
||||
``,
|
||||
)
|
||||
.join("\n\n");
|
||||
}
|
||||
|
||||
export async function createDeferredUploadedImage(
|
||||
sourceUrl: string,
|
||||
metadata?: DeferredImageArchiveMetadata | null
|
||||
metadata?: DeferredImageArchiveMetadata | null,
|
||||
) {
|
||||
const url = String(sourceUrl || "").trim();
|
||||
if (!isDeferredImageArchiveEnabled() || !isHttpUrl(url)) return null;
|
||||
if (!isDeferredImageArchiveEnabled() || !isGeneratedImageSource(url)) return null;
|
||||
|
||||
const [persisted] = await persistGeneratedImageUrls({
|
||||
urls: [url],
|
||||
@ -124,48 +121,20 @@ async function persistGeneratedImageUrls(params: {
|
||||
urls: string[];
|
||||
metadata?: DeferredImageArchiveMetadata | null;
|
||||
}) {
|
||||
if (!isImmediateImageArchiveEnabled()) {
|
||||
await registerGeneratedImageTemporaryUrls(params);
|
||||
return params.urls.map((url) => ({
|
||||
originalUrl: url,
|
||||
thumbnailUrl: url,
|
||||
}));
|
||||
}
|
||||
|
||||
return await Promise.all(
|
||||
params.urls.map(async (url) => {
|
||||
try {
|
||||
const archived = await archiveTemporaryImage(url);
|
||||
logger.info("generated image archived before response", {
|
||||
providerType: params.metadata?.providerType,
|
||||
modelId: params.metadata?.modelId,
|
||||
bytes: archived.bytes,
|
||||
});
|
||||
return {
|
||||
originalUrl: archived.url,
|
||||
thumbnailUrl: archived.thumbnailUrl || archived.url,
|
||||
};
|
||||
} catch (error) {
|
||||
logger.error("immediate generated image archive failed; queued for retry", {
|
||||
providerType: params.metadata?.providerType,
|
||||
modelId: params.metadata?.modelId,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
await registerGeneratedImageTemporaryUrls({
|
||||
urls: [url],
|
||||
metadata: params.metadata || null,
|
||||
});
|
||||
return {
|
||||
originalUrl: url,
|
||||
thumbnailUrl: url,
|
||||
};
|
||||
}
|
||||
})
|
||||
);
|
||||
// Only the durable queue insert is awaited. Downloading, thumbnailing and bucket
|
||||
// uploads run after delivery and never extend the provider response time.
|
||||
await registerGeneratedImageTemporaryUrls(params);
|
||||
return params.urls.map((url) => {
|
||||
const preview = imageArchiveSourceKey(url);
|
||||
return { originalUrl: preview, thumbnailUrl: preview };
|
||||
});
|
||||
}
|
||||
|
||||
function getImageExtension(contentType: string | null, sourceUrl: string) {
|
||||
const normalized = String(contentType || "").split(";")[0].trim().toLowerCase();
|
||||
const normalized = String(contentType || "")
|
||||
.split(";")[0]
|
||||
.trim()
|
||||
.toLowerCase();
|
||||
const byType: Record<string, string> = {
|
||||
"image/jpeg": "jpg",
|
||||
"image/jpg": "jpg",
|
||||
@ -178,7 +147,11 @@ function getImageExtension(contentType: string | null, sourceUrl: string) {
|
||||
|
||||
try {
|
||||
const pathname = new URL(sourceUrl).pathname.toLowerCase();
|
||||
const ext = pathname.split(".").pop()?.replace(/[^a-z0-9]+/g, "") || "";
|
||||
const ext =
|
||||
pathname
|
||||
.split(".")
|
||||
.pop()
|
||||
?.replace(/[^a-z0-9]+/g, "") || "";
|
||||
if (["jpg", "jpeg", "png", "webp", "gif", "svg"].includes(ext)) {
|
||||
return ext === "jpeg" ? "jpg" : ext;
|
||||
}
|
||||
@ -192,14 +165,20 @@ function getImageExtension(contentType: string | null, sourceUrl: string) {
|
||||
function toArrayBuffer(buffer: Buffer): ArrayBuffer {
|
||||
return buffer.buffer.slice(
|
||||
buffer.byteOffset,
|
||||
buffer.byteOffset + buffer.byteLength
|
||||
buffer.byteOffset + buffer.byteLength,
|
||||
) as ArrayBuffer;
|
||||
}
|
||||
|
||||
export async function downloadTemporaryImage(params: {
|
||||
sourceUrl: string;
|
||||
retries?: number;
|
||||
}) {
|
||||
export async function downloadTemporaryImage(params: { sourceUrl: string; retries?: number }) {
|
||||
if (params.sourceUrl.startsWith("data:")) {
|
||||
const match = params.sourceUrl.match(/^data:(image\/[a-z0-9.+-]+);base64,([a-z0-9+/=]+)$/i);
|
||||
if (!match) throw new Error("Invalid inline generated image");
|
||||
return {
|
||||
buffer: Buffer.from(match[2], "base64"),
|
||||
contentType: match[1],
|
||||
extension: getImageExtension(match[1], ""),
|
||||
};
|
||||
}
|
||||
const retries = Math.max(0, Math.min(5, params.retries ?? 2));
|
||||
|
||||
for (let attempt = 0; attempt <= retries; attempt += 1) {
|
||||
@ -249,7 +228,7 @@ export async function archiveTemporaryImage(sourceUrl: string) {
|
||||
const thumbnailUploaded = thumbnailBuffer
|
||||
? await uploadFile(
|
||||
`ai-generated-archive-thumb-${objectSuffix}.webp`,
|
||||
toArrayBuffer(thumbnailBuffer)
|
||||
toArrayBuffer(thumbnailBuffer),
|
||||
).catch((error) => {
|
||||
logger.warn("generated image thumbnail upload failed", {
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
|
||||
@ -6,6 +6,7 @@ import {
|
||||
assetMemorySnapshot,
|
||||
canvasDetailArchive,
|
||||
canvasNode,
|
||||
canvasImageTask,
|
||||
canvasProject,
|
||||
conversationWorkingState,
|
||||
externalImageApiRequestLog,
|
||||
@ -26,18 +27,14 @@ function normalizeMediaUrl(value: string): string {
|
||||
const parsed = new URL(canonical);
|
||||
return `${parsed.origin}${parsed.pathname}`.replace(/\/+$/, "").toLowerCase();
|
||||
} catch {
|
||||
return canonical
|
||||
.split("#")[0]
|
||||
.split("?")[0]
|
||||
.replace(/\/+$/, "")
|
||||
.toLowerCase();
|
||||
return canonical.split("#")[0].split("?")[0].replace(/\/+$/, "").toLowerCase();
|
||||
}
|
||||
}
|
||||
|
||||
export function replaceDeepUrls(
|
||||
value: unknown,
|
||||
sourceUrl: string,
|
||||
targetUrl: string
|
||||
targetUrl: string,
|
||||
): { value: unknown; changed: boolean } {
|
||||
if (typeof value === "string") {
|
||||
if (!value.includes(sourceUrl)) {
|
||||
@ -70,10 +67,7 @@ export function replaceDeepUrls(
|
||||
return { value, changed: false };
|
||||
}
|
||||
|
||||
export async function replaceImageUrlInDatabase(
|
||||
sourceUrl: string,
|
||||
targetUrl: string
|
||||
) {
|
||||
export async function replaceImageUrlInDatabase(sourceUrl: string, targetUrl: string) {
|
||||
if (!sourceUrl || !targetUrl || sourceUrl === targetUrl) {
|
||||
return { changed: 0 };
|
||||
}
|
||||
@ -106,8 +100,8 @@ export async function replaceImageUrlInDatabase(
|
||||
eq(mediaAsset.thumbnailUrl, sourceUrl),
|
||||
eq(mediaAsset.normalizedUrl, sourceNormalizedUrl),
|
||||
sql`${mediaAsset.metadata}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${mediaAsset.summary}::text like ${`%${sourceUrl}%`}`
|
||||
)
|
||||
sql`${mediaAsset.summary}::text like ${`%${sourceUrl}%`}`,
|
||||
),
|
||||
)
|
||||
.limit(500);
|
||||
|
||||
@ -117,8 +111,7 @@ export async function replaceImageUrlInDatabase(
|
||||
const thumbnail = replaceDeepUrls(item.thumbnailUrl, sourceUrl, targetUrl);
|
||||
const metadata = replaceDeepUrls(item.metadata, sourceUrl, targetUrl);
|
||||
const summary = replaceDeepUrls(item.summary, sourceUrl, targetUrl);
|
||||
const shouldNormalize =
|
||||
item.normalizedUrl === sourceNormalizedUrl || original.changed;
|
||||
const shouldNormalize = item.normalizedUrl === sourceNormalizedUrl || original.changed;
|
||||
const nextOriginal = (original.value as string) || item.originalUrl;
|
||||
|
||||
if (
|
||||
@ -139,8 +132,7 @@ export async function replaceImageUrlInDatabase(
|
||||
originalUrl: nextOriginal,
|
||||
normalizedUrl: shouldNormalize ? targetNormalizedUrl : item.normalizedUrl,
|
||||
previewUrl: (preview.value as string | null | undefined) ?? item.previewUrl,
|
||||
thumbnailUrl:
|
||||
(thumbnail.value as string | null | undefined) ?? item.thumbnailUrl,
|
||||
thumbnailUrl: (thumbnail.value as string | null | undefined) ?? item.thumbnailUrl,
|
||||
metadata: (metadata.value ?? item.metadata) as any,
|
||||
summary: (summary.value ?? item.summary) as any,
|
||||
updatedAt: new Date(),
|
||||
@ -152,8 +144,7 @@ export async function replaceImageUrlInDatabase(
|
||||
.set({
|
||||
originalUrl: nextOriginal,
|
||||
previewUrl: (preview.value as string | null | undefined) ?? item.previewUrl,
|
||||
thumbnailUrl:
|
||||
(thumbnail.value as string | null | undefined) ?? item.thumbnailUrl,
|
||||
thumbnailUrl: (thumbnail.value as string | null | undefined) ?? item.thumbnailUrl,
|
||||
metadata: (metadata.value ?? item.metadata) as any,
|
||||
summary: (summary.value ?? item.summary) as any,
|
||||
updatedAt: new Date(),
|
||||
@ -170,8 +161,8 @@ export async function replaceImageUrlInDatabase(
|
||||
or(
|
||||
eq(assetMemorySnapshot.normalizedUrl, sourceNormalizedUrl),
|
||||
sql`${assetMemorySnapshot.metadata}::text like ${`%${sourceUrl}%`}`,
|
||||
like(assetMemorySnapshot.summaryText, `%${sourceUrl}%`)
|
||||
)
|
||||
like(assetMemorySnapshot.summaryText, `%${sourceUrl}%`),
|
||||
),
|
||||
)
|
||||
.limit(500);
|
||||
|
||||
@ -203,8 +194,8 @@ export async function replaceImageUrlInDatabase(
|
||||
.where(
|
||||
or(
|
||||
sql`${message.parts}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${message.attachments}::text like ${`%${sourceUrl}%`}`
|
||||
)
|
||||
sql`${message.attachments}::text like ${`%${sourceUrl}%`}`,
|
||||
),
|
||||
)
|
||||
.limit(500);
|
||||
|
||||
@ -243,120 +234,67 @@ export async function replaceImageUrlInDatabase(
|
||||
changed += 1;
|
||||
}
|
||||
|
||||
const escapedSource = JSON.stringify(sourceUrl).slice(1, -1);
|
||||
const escapedTarget = JSON.stringify(targetUrl).slice(1, -1);
|
||||
const projects = await db
|
||||
.select({
|
||||
id: canvasProject.id,
|
||||
canvasData: canvasProject.canvasData,
|
||||
.update(canvasProject)
|
||||
.set({
|
||||
canvasData: sql`replace(${canvasProject.canvasData}::text, ${escapedSource}, ${escapedTarget})::jsonb`,
|
||||
thumbnail: sql`case when ${canvasProject.thumbnail} = ${sourceUrl} then ${targetUrl} else ${canvasProject.thumbnail} end`,
|
||||
// URL maintenance is not a content revision; keep autosave conflict detection valid.
|
||||
})
|
||||
.from(canvasProject)
|
||||
.where(sql`${canvasProject.canvasData}::text like ${`%${sourceUrl}%`}`)
|
||||
.limit(500);
|
||||
|
||||
for (const item of projects) {
|
||||
const replaced = replaceDeepUrls(item.canvasData, sourceUrl, targetUrl);
|
||||
if (!replaced.changed) continue;
|
||||
|
||||
await db
|
||||
.update(canvasProject)
|
||||
.set({
|
||||
canvasData: replaced.value as any,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(canvasProject.id, item.id));
|
||||
changed += 1;
|
||||
}
|
||||
|
||||
const nodes = await db
|
||||
.select({
|
||||
id: canvasNode.id,
|
||||
data: canvasNode.data,
|
||||
})
|
||||
.from(canvasNode)
|
||||
.where(sql`${canvasNode.data}::text like ${`%${sourceUrl}%`}`)
|
||||
.limit(500);
|
||||
|
||||
for (const item of nodes) {
|
||||
const replaced = replaceDeepUrls(item.data, sourceUrl, targetUrl);
|
||||
if (!replaced.changed) continue;
|
||||
|
||||
await db
|
||||
.update(canvasNode)
|
||||
.set({
|
||||
data: replaced.value as any,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(canvasNode.id, item.id));
|
||||
changed += 1;
|
||||
}
|
||||
|
||||
const workingStates = await db
|
||||
.select()
|
||||
.from(conversationWorkingState)
|
||||
.where(
|
||||
or(
|
||||
sql`${conversationWorkingState.currentFocusImageUrls}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${conversationWorkingState.selectedImageUrls}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${conversationWorkingState.lastUploadedImageUrls}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${conversationWorkingState.lastGeneratedImageUrls}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${conversationWorkingState.lastReferencedImageUrls}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${conversationWorkingState.lastResolvedTargetImageUrls}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${conversationWorkingState.lastCanvasSnapshot}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${conversationWorkingState.lastTurnSummary}::text like ${`%${sourceUrl}%`}`
|
||||
)
|
||||
sql`position(${escapedSource} in ${canvasProject.canvasData}::text) > 0`,
|
||||
eq(canvasProject.thumbnail, sourceUrl),
|
||||
),
|
||||
)
|
||||
.limit(500);
|
||||
.returning({ id: canvasProject.id });
|
||||
changed += projects.length;
|
||||
const nodes = await db
|
||||
.update(canvasNode)
|
||||
.set({
|
||||
data: sql`replace(${canvasNode.data}::text, ${escapedSource}, ${escapedTarget})::jsonb`,
|
||||
})
|
||||
.where(sql`position(${escapedSource} in ${canvasNode.data}::text) > 0`)
|
||||
.returning({ id: canvasNode.id });
|
||||
changed += nodes.length;
|
||||
const tasks = await db
|
||||
.update(canvasImageTask)
|
||||
.set({
|
||||
result: sql`replace(${canvasImageTask.result}::text, ${escapedSource}, ${escapedTarget})::jsonb`,
|
||||
})
|
||||
.where(sql`position(${escapedSource} in ${canvasImageTask.result}::text) > 0`)
|
||||
.returning({ id: canvasImageTask.id });
|
||||
changed += tasks.length;
|
||||
|
||||
for (const item of workingStates) {
|
||||
const currentFocus = replaceDeepUrls(
|
||||
item.currentFocusImageUrls,
|
||||
sourceUrl,
|
||||
targetUrl
|
||||
);
|
||||
const selected = replaceDeepUrls(item.selectedImageUrls, sourceUrl, targetUrl);
|
||||
const uploaded = replaceDeepUrls(item.lastUploadedImageUrls, sourceUrl, targetUrl);
|
||||
const generated = replaceDeepUrls(item.lastGeneratedImageUrls, sourceUrl, targetUrl);
|
||||
const referenced = replaceDeepUrls(item.lastReferencedImageUrls, sourceUrl, targetUrl);
|
||||
const resolved = replaceDeepUrls(
|
||||
item.lastResolvedTargetImageUrls,
|
||||
sourceUrl,
|
||||
targetUrl
|
||||
);
|
||||
const canvasSnapshot = replaceDeepUrls(
|
||||
item.lastCanvasSnapshot,
|
||||
sourceUrl,
|
||||
targetUrl
|
||||
);
|
||||
const turnSummary = replaceDeepUrls(item.lastTurnSummary, sourceUrl, targetUrl);
|
||||
|
||||
if (
|
||||
!currentFocus.changed &&
|
||||
!selected.changed &&
|
||||
!uploaded.changed &&
|
||||
!generated.changed &&
|
||||
!referenced.changed &&
|
||||
!resolved.changed &&
|
||||
!canvasSnapshot.changed &&
|
||||
!turnSummary.changed
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
|
||||
await db
|
||||
.update(conversationWorkingState)
|
||||
.set({
|
||||
currentFocusImageUrls: currentFocus.value as any,
|
||||
selectedImageUrls: selected.value as any,
|
||||
lastUploadedImageUrls: uploaded.value as any,
|
||||
lastGeneratedImageUrls: generated.value as any,
|
||||
lastReferencedImageUrls: referenced.value as any,
|
||||
lastResolvedTargetImageUrls: resolved.value as any,
|
||||
lastCanvasSnapshot: (canvasSnapshot.value ?? item.lastCanvasSnapshot) as any,
|
||||
lastTurnSummary: (turnSummary.value ?? item.lastTurnSummary) as any,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(conversationWorkingState.id, item.id));
|
||||
changed += 1;
|
||||
}
|
||||
const workingStates = await db
|
||||
.update(conversationWorkingState)
|
||||
.set({
|
||||
currentFocusImageUrls: sql`replace(${conversationWorkingState.currentFocusImageUrls}::text, ${escapedSource}, ${escapedTarget})::jsonb`,
|
||||
selectedImageUrls: sql`replace(${conversationWorkingState.selectedImageUrls}::text, ${escapedSource}, ${escapedTarget})::jsonb`,
|
||||
lastUploadedImageUrls: sql`replace(${conversationWorkingState.lastUploadedImageUrls}::text, ${escapedSource}, ${escapedTarget})::jsonb`,
|
||||
lastGeneratedImageUrls: sql`replace(${conversationWorkingState.lastGeneratedImageUrls}::text, ${escapedSource}, ${escapedTarget})::jsonb`,
|
||||
lastReferencedImageUrls: sql`replace(${conversationWorkingState.lastReferencedImageUrls}::text, ${escapedSource}, ${escapedTarget})::jsonb`,
|
||||
lastResolvedTargetImageUrls: sql`replace(${conversationWorkingState.lastResolvedTargetImageUrls}::text, ${escapedSource}, ${escapedTarget})::jsonb`,
|
||||
lastCanvasSnapshot: sql`replace(${conversationWorkingState.lastCanvasSnapshot}::text, ${escapedSource}, ${escapedTarget})::jsonb`,
|
||||
lastTurnSummary: sql`replace(${conversationWorkingState.lastTurnSummary}::text, ${escapedSource}, ${escapedTarget})::jsonb`,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
or(
|
||||
sql`position(${escapedSource} in ${conversationWorkingState.currentFocusImageUrls}::text) > 0`,
|
||||
sql`position(${escapedSource} in ${conversationWorkingState.selectedImageUrls}::text) > 0`,
|
||||
sql`position(${escapedSource} in ${conversationWorkingState.lastUploadedImageUrls}::text) > 0`,
|
||||
sql`position(${escapedSource} in ${conversationWorkingState.lastGeneratedImageUrls}::text) > 0`,
|
||||
sql`position(${escapedSource} in ${conversationWorkingState.lastReferencedImageUrls}::text) > 0`,
|
||||
sql`position(${escapedSource} in ${conversationWorkingState.lastResolvedTargetImageUrls}::text) > 0`,
|
||||
sql`position(${escapedSource} in ${conversationWorkingState.lastCanvasSnapshot}::text) > 0`,
|
||||
sql`position(${escapedSource} in ${conversationWorkingState.lastTurnSummary}::text) > 0`,
|
||||
),
|
||||
)
|
||||
.returning({ id: conversationWorkingState.id });
|
||||
changed += workingStates.length;
|
||||
|
||||
const detailArchives = await db
|
||||
.select({
|
||||
@ -374,8 +312,8 @@ export async function replaceImageUrlInDatabase(
|
||||
sql`${canvasDetailArchive.brief}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${canvasDetailArchive.slots}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${canvasDetailArchive.uiSurfaceSnapshot}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${canvasDetailArchive.uiSurfaceLatestReview}::text like ${`%${sourceUrl}%`}`
|
||||
)
|
||||
sql`${canvasDetailArchive.uiSurfaceLatestReview}::text like ${`%${sourceUrl}%`}`,
|
||||
),
|
||||
)
|
||||
.limit(500);
|
||||
|
||||
@ -416,8 +354,8 @@ export async function replaceImageUrlInDatabase(
|
||||
.where(
|
||||
or(
|
||||
like(model3dAsset.previewImageUrl, `%${sourceUrl}%`),
|
||||
sql`${model3dAsset.assetPayload}::text like ${`%${sourceUrl}%`}`
|
||||
)
|
||||
sql`${model3dAsset.assetPayload}::text like ${`%${sourceUrl}%`}`,
|
||||
),
|
||||
)
|
||||
.limit(500);
|
||||
|
||||
@ -429,8 +367,7 @@ export async function replaceImageUrlInDatabase(
|
||||
await db
|
||||
.update(model3dAsset)
|
||||
.set({
|
||||
previewImageUrl:
|
||||
(preview.value as string | null | undefined) ?? item.previewImageUrl,
|
||||
previewImageUrl: (preview.value as string | null | undefined) ?? item.previewImageUrl,
|
||||
assetPayload: (payload.value ?? item.assetPayload) as any,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
@ -444,22 +381,14 @@ export async function replaceImageUrlInDatabase(
|
||||
.where(
|
||||
or(
|
||||
sql`${externalImageApiRequestLog.responsePayload}::text like ${`%${sourceUrl}%`}`,
|
||||
sql`${externalImageApiRequestLog.generatedImageUrls}::text like ${`%${sourceUrl}%`}`
|
||||
)
|
||||
sql`${externalImageApiRequestLog.generatedImageUrls}::text like ${`%${sourceUrl}%`}`,
|
||||
),
|
||||
)
|
||||
.limit(500);
|
||||
|
||||
for (const item of apiLogs) {
|
||||
const responsePayload = replaceDeepUrls(
|
||||
item.responsePayload,
|
||||
sourceUrl,
|
||||
targetUrl
|
||||
);
|
||||
const generatedUrls = replaceDeepUrls(
|
||||
item.generatedImageUrls,
|
||||
sourceUrl,
|
||||
targetUrl
|
||||
);
|
||||
const responsePayload = replaceDeepUrls(item.responsePayload, sourceUrl, targetUrl);
|
||||
const generatedUrls = replaceDeepUrls(item.generatedImageUrls, sourceUrl, targetUrl);
|
||||
if (!responsePayload.changed && !generatedUrls.changed) continue;
|
||||
|
||||
await db
|
||||
|
||||
@ -1,3 +1,4 @@
|
||||
import { replaceArchivedImageUrls, type ArchivedImageUrlMap } from "@/lib/storage/image-archive-urls";
|
||||
import { create } from 'zustand';
|
||||
import { immer } from 'zustand/middleware/immer';
|
||||
import { devtools, persist } from 'zustand/middleware';
|
||||
@ -161,6 +162,7 @@ interface CanvasStore {
|
||||
// 节点操作
|
||||
addNode: (node: Omit<CanvasNode, 'id' | 'createdAt' | 'updatedAt'>) => string;
|
||||
updateNode: (id: string, updates: Partial<CanvasNode>) => void;
|
||||
replaceArchivedImageUrls: (replacements: ArchivedImageUrlMap) => void;
|
||||
deleteNode: (id: string) => void;
|
||||
deleteSelectedNodes: () => void;
|
||||
addSuppressedMediaKeys: (keys: string[]) => void;
|
||||
@ -410,6 +412,19 @@ export const useCanvasStore = create<CanvasStore>()(
|
||||
}
|
||||
}),
|
||||
|
||||
replaceArchivedImageUrls: (replacements) => set((state) => {
|
||||
state.nodes = replaceArchivedImageUrls(state.nodes, replacements);
|
||||
state.history = replaceArchivedImageUrls(state.history, replacements);
|
||||
state.styleProState = replaceArchivedImageUrls(state.styleProState, replacements);
|
||||
state.pendingChatAttachments = replaceArchivedImageUrls(state.pendingChatAttachments, replacements);
|
||||
state.pendingPickedObjects = replaceArchivedImageUrls(state.pendingPickedObjects, replacements);
|
||||
state.pickedObject = replaceArchivedImageUrls(state.pickedObject, replacements);
|
||||
state.imageNodeActionRequest = replaceArchivedImageUrls(state.imageNodeActionRequest, replacements);
|
||||
state.imageNodeActionQueue = replaceArchivedImageUrls(state.imageNodeActionQueue, replacements);
|
||||
state.canvasImageSelectionSession = replaceArchivedImageUrls(state.canvasImageSelectionSession, replacements);
|
||||
state.suppressedMediaKeys = replaceArchivedImageUrls(state.suppressedMediaKeys, replacements);
|
||||
}),
|
||||
|
||||
addSuppressedMediaKeys: (keys) => set((state) => {
|
||||
const normalized = Array.from(
|
||||
new Set((keys || []).map((item) => String(item || "").trim()).filter(Boolean))
|
||||
|
||||
8
tests/image-archive.config.ts
Normal file
8
tests/image-archive.config.ts
Normal file
@ -0,0 +1,8 @@
|
||||
import { defineConfig } from "@playwright/test";
|
||||
export default defineConfig({
|
||||
testDir: "./routes",
|
||||
testMatch: /image-archive-flow\.test\.ts/,
|
||||
workers: 1,
|
||||
reporter: "list",
|
||||
timeout: 30_000,
|
||||
});
|
||||
626
tests/routes/image-archive-flow.test.ts
Normal file
626
tests/routes/image-archive-flow.test.ts
Normal file
@ -0,0 +1,626 @@
|
||||
import { expect, test } from "@playwright/test";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { createRequire } from "node:module";
|
||||
import path from "node:path";
|
||||
import vm from "node:vm";
|
||||
import * as crypto from "node:crypto";
|
||||
import ts from "typescript";
|
||||
import {
|
||||
collectImageArchiveSources,
|
||||
replaceArchivedImageUrls,
|
||||
} from "@/lib/storage/image-archive-urls";
|
||||
import { normalizeCanvasImageTaskQueueConfig } from "@/lib/canvas/image-task-config";
|
||||
import type { ImageNodeData } from "@/types/canvas";
|
||||
import { useCanvasStore } from "@/store/canvas";
|
||||
|
||||
const root = path.resolve(__dirname, "../..");
|
||||
const png = Buffer.from(
|
||||
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+jRZkAAAAASUVORK5CYII=",
|
||||
"base64",
|
||||
);
|
||||
const raw = "https://provider.example/result.png?signature=keep-me";
|
||||
const permanent = "https://bucket.example/result.png";
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>((done) => {
|
||||
resolve = done;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
// Execute the actual server modules with isolated DB/storage adapters: no provider
|
||||
// requests, production DB connections, or paid generations are made by these tests.
|
||||
function serverLoader(mocks: Record<string, unknown>, env: Record<string, string> = {}) {
|
||||
const modules = new Map<string, { exports: any }>();
|
||||
const runtimeGlobal = {};
|
||||
const load = (relative: string): any => {
|
||||
const filename = path.resolve(root, relative);
|
||||
if (modules.has(filename)) return modules.get(filename)!.exports;
|
||||
const module = { exports: {} };
|
||||
modules.set(filename, module);
|
||||
const source = ts.transpileModule(readFileSync(filename, "utf8"), {
|
||||
compilerOptions: {
|
||||
module: ts.ModuleKind.CommonJS,
|
||||
target: ts.ScriptTarget.ES2022,
|
||||
esModuleInterop: true,
|
||||
},
|
||||
}).outputText;
|
||||
const nativeRequire = createRequire(filename);
|
||||
const requireMock = (name: string) => {
|
||||
if (name === "server-only") return {};
|
||||
if (Object.prototype.hasOwnProperty.call(mocks, name)) return mocks[name];
|
||||
if (name.startsWith("@/")) return load(`${name.slice(2)}.ts`);
|
||||
if (name.startsWith("."))
|
||||
return load(path.relative(root, path.resolve(path.dirname(filename), `${name}.ts`)));
|
||||
return nativeRequire(name);
|
||||
};
|
||||
const context = vm.createContext({
|
||||
module,
|
||||
exports: module.exports,
|
||||
require: requireMock,
|
||||
globalThis: runtimeGlobal,
|
||||
process: { env },
|
||||
Buffer,
|
||||
URL,
|
||||
Response,
|
||||
crypto,
|
||||
setTimeout,
|
||||
clearTimeout,
|
||||
setImmediate,
|
||||
AbortSignal,
|
||||
console: { log() {}, warn() {}, error() {}, info() {} },
|
||||
fetch:
|
||||
mocks.fetch ||
|
||||
(() => {
|
||||
throw new Error("Unexpected network call");
|
||||
}),
|
||||
});
|
||||
new vm.Script(source, { filename }).runInContext(context);
|
||||
return module.exports;
|
||||
};
|
||||
return load;
|
||||
}
|
||||
const loggerMock = { createLogger: () => ({ info() {}, warn() {}, error() {} }) };
|
||||
|
||||
test("160 requests from the same user and model start without admission waiting", async () => {
|
||||
const module = serverLoader({})("lib/canvas/image-concurrency.ts");
|
||||
let queued = 0;
|
||||
const slots = await Promise.all(
|
||||
Array.from({ length: 160 }, () =>
|
||||
module.acquireCanvasImageModelSlot({
|
||||
modelKey: "same-model",
|
||||
ownerKey: "same-user",
|
||||
onQueued: () => {
|
||||
queued += 1;
|
||||
},
|
||||
}),
|
||||
),
|
||||
);
|
||||
expect(module.getCanvasImageRunningCountForModel("same-model")).toBe(160);
|
||||
expect(queued).toBe(0);
|
||||
expect(slots.every((slot: any) => slot.waitMs === 0 && slot.maxConcurrent === 0)).toBe(true);
|
||||
await Promise.all(slots.map((slot: any) => slot.release()));
|
||||
await slots[0].release();
|
||||
expect(module.getCanvasImageRunningCountForModel("same-model")).toBe(0);
|
||||
expect(
|
||||
(await module.getCanvasImageModelQueueStatus({ modelKey: "same-model" })).estimatedWaitMs,
|
||||
).toBe(0);
|
||||
expect(
|
||||
normalizeCanvasImageTaskQueueConfig(
|
||||
{ concurrency: 3, batchDispatchIntervalMs: 1000 },
|
||||
{ envConcurrency: "2" },
|
||||
),
|
||||
).toEqual({ concurrency: 0, batchDispatchIntervalMs: 0 });
|
||||
});
|
||||
|
||||
test("provider preview resolves while a bucket upload is still blocked", async () => {
|
||||
const bucket = deferred<any>();
|
||||
let uploadCalls = 0;
|
||||
let downloadCalls = 0;
|
||||
let workerCalls = 0;
|
||||
const registered: string[] = [];
|
||||
const module = serverLoader(
|
||||
{
|
||||
"@/lib/observability/logger": loggerMock,
|
||||
"@/lib/storage/image-archive-db": {
|
||||
enqueueImageArchiveTasks: async ({ urls }: { urls: string[] }) => {
|
||||
registered.push(...urls);
|
||||
return [];
|
||||
},
|
||||
getImageArchiveDisplayHoldMs: () => 0,
|
||||
imageArchiveSourceKey: (url: string) => url,
|
||||
},
|
||||
"./image-archive-worker": {
|
||||
scheduleImageArchiveWorker: () => {
|
||||
workerCalls += 1;
|
||||
},
|
||||
},
|
||||
"@/lib/storage/service": {
|
||||
uploadFile: async () => {
|
||||
uploadCalls += 1;
|
||||
return bucket.promise;
|
||||
},
|
||||
},
|
||||
fetch: async () => {
|
||||
downloadCalls += 1;
|
||||
return new Response(png, { headers: { "content-type": "image/png" } });
|
||||
},
|
||||
},
|
||||
{ AI_IMAGE_IMMEDIATE_ARCHIVE: "true", AI_IMAGE_DEFERRED_ARCHIVE: "false" },
|
||||
)("lib/storage/image-archive.ts");
|
||||
const archive = module.archiveTemporaryImage(raw);
|
||||
await expect.poll(() => uploadCalls).toBe(1);
|
||||
const preview = await module.createDeferredUploadedImage(raw);
|
||||
expect(preview).toEqual({ original: raw, thumbnail: raw });
|
||||
expect(uploadCalls).toBe(1);
|
||||
expect(downloadCalls).toBe(1);
|
||||
expect(registered).toEqual([raw]);
|
||||
expect(workerCalls).toBe(1);
|
||||
bucket.resolve({ url: permanent, pathname: "result.png", contentType: "image/png" });
|
||||
expect((await archive).url).toBe(permanent);
|
||||
});
|
||||
|
||||
test("inline outputs have short, stable preview identities and durable queue payloads", async () => {
|
||||
let inserted: any;
|
||||
const db = {
|
||||
insert: () => ({
|
||||
values: (value: unknown) => {
|
||||
inserted = value;
|
||||
return { onConflictDoUpdate: () => ({ returning: async () => [inserted] }) };
|
||||
},
|
||||
}),
|
||||
};
|
||||
const module = serverLoader(
|
||||
{
|
||||
"@/lib/db": { db },
|
||||
"@/lib/db/schema": { imageArchiveTask: { sourceUrl: "sourceUrl" } },
|
||||
},
|
||||
{ NEXT_PUBLIC_APP_URL: "https://app.example/subpath", IMAGE_ARCHIVE_DISPLAY_HOLD_HOURS: "1" },
|
||||
)("lib/storage/image-archive-db.ts");
|
||||
const data = `data:image/png;base64,${png.toString("base64")}`;
|
||||
await module.enqueueImageArchiveTask({
|
||||
sourceUrl: data,
|
||||
metadata: { providerType: "gemini-image" },
|
||||
});
|
||||
expect(inserted.sourceUrl).toMatch(
|
||||
/^https:\/\/app\.example\/api\/generated-image-preview\/[a-f0-9]{64}$/,
|
||||
);
|
||||
expect(inserted.metadata.inlineSource).toBe(data);
|
||||
expect(module.imageArchiveSourceKey(data)).toBe(inserted.sourceUrl);
|
||||
expect(module.getImageArchiveDisplayHoldMs()).toBe(0);
|
||||
});
|
||||
|
||||
test("background storage starts after delivery and coalesces repeated wakeups", async () => {
|
||||
const callbacks: Array<() => Promise<void>> = [];
|
||||
let runs = 0;
|
||||
const module = serverLoader({
|
||||
"next/server": {
|
||||
after: (fn: () => Promise<void>) => {
|
||||
callbacks.push(fn);
|
||||
},
|
||||
},
|
||||
"@/lib/jobs/image-archive-job": {
|
||||
runImageArchiveJob: async () => {
|
||||
runs += 1;
|
||||
return {
|
||||
store: { total: runs === 1 ? 1 : 0 },
|
||||
replace: { total: 0 },
|
||||
stats: { pendingStore: 0, pendingReplace: 0 },
|
||||
};
|
||||
},
|
||||
},
|
||||
})("lib/storage/image-archive-worker.ts");
|
||||
module.scheduleImageArchiveWorker();
|
||||
module.scheduleImageArchiveWorker();
|
||||
expect(callbacks).toHaveLength(1);
|
||||
expect(runs).toBe(0);
|
||||
await callbacks[0]();
|
||||
expect(runs).toBe(2);
|
||||
});
|
||||
|
||||
test("bucket failure retains the provider URL and retry task", async () => {
|
||||
let failed = 0;
|
||||
let stored = 0;
|
||||
let replaced = 0;
|
||||
const module = serverLoader({
|
||||
"@/lib/storage/image-archive": {
|
||||
archiveTemporaryImage: async () => {
|
||||
throw new Error("Bucket unavailable");
|
||||
},
|
||||
},
|
||||
"@/lib/storage/image-archive-db": {
|
||||
recoverStaleImageArchiveTasks: async () => {},
|
||||
getImageArchiveTasksToStore: async () => [{ id: "task", sourceUrl: raw, metadata: null }],
|
||||
markImageArchiveTaskArchiving: async () => true,
|
||||
markImageArchiveTaskStored: async () => {
|
||||
stored += 1;
|
||||
},
|
||||
markImageArchiveTaskFailed: async () => {
|
||||
failed += 1;
|
||||
},
|
||||
getImageArchiveTasksToReplace: async () => [],
|
||||
getImageArchiveTaskStats: async () => ({ pendingStore: 1, pendingReplace: 0 }),
|
||||
},
|
||||
"@/lib/storage/image-url-replacement": {
|
||||
replaceImageUrlInDatabase: async () => {
|
||||
replaced += 1;
|
||||
},
|
||||
},
|
||||
})("lib/jobs/image-archive-job.ts");
|
||||
const result = await module.runImageArchiveJob();
|
||||
expect(result.store.failed).toBe(1);
|
||||
expect(failed).toBe(1);
|
||||
expect(stored).toBe(0);
|
||||
expect(replaced).toBe(0);
|
||||
});
|
||||
|
||||
test("URL replacement preserves canvas geometry, selection, dates and undo position", () => {
|
||||
const time = new Date("2026-10-11T01:00:00Z");
|
||||
const node = {
|
||||
id: "node",
|
||||
type: "image",
|
||||
position: { x: 300, y: 40 },
|
||||
size: { width: 300, height: 500 },
|
||||
data: { src: raw, originalUrl: raw },
|
||||
createdAt: time,
|
||||
};
|
||||
const unaffected = { id: "text", type: "text", data: { text: "unchanged" } };
|
||||
const nodes = [node, unaffected];
|
||||
useCanvasStore.setState({
|
||||
projectId: "project",
|
||||
nodes: nodes as any,
|
||||
history: [{ nodes } as any],
|
||||
historyIndex: 0,
|
||||
selectedNodeIds: ["node"],
|
||||
});
|
||||
useCanvasStore.getState().replaceArchivedImageUrls({ [raw]: permanent });
|
||||
const state = useCanvasStore.getState();
|
||||
expect((state.nodes[0].data as ImageNodeData).src).toBe(permanent);
|
||||
expect(state.nodes[0].position).toEqual(node.position);
|
||||
expect(state.nodes[0].size).toEqual(node.size);
|
||||
expect(state.nodes[0].createdAt).toBe(time);
|
||||
expect(state.nodes[1]).toBe(unaffected);
|
||||
expect((state.history[0].nodes[0].data as ImageNodeData).src).toBe(permanent);
|
||||
expect(state.historyIndex).toBe(0);
|
||||
expect(state.selectedNodeIds).toEqual(["node"]);
|
||||
const markdown = ``;
|
||||
expect(replaceArchivedImageUrls(markdown, { [raw]: permanent })).toBe(
|
||||
``,
|
||||
);
|
||||
expect(collectImageArchiveSources({ nodes, markdown })).toEqual([raw]);
|
||||
});
|
||||
|
||||
test("durable canvas queue starts all same-user jobs and claims each only once", async () => {
|
||||
const tasks = Array.from({ length: 20 }, (_, index) => ({
|
||||
id: `task-${index}`,
|
||||
userId: "one-user",
|
||||
status: "pending",
|
||||
attempts: 0,
|
||||
request: { model: "one-model", prompt: "test" },
|
||||
createdAt: new Date(),
|
||||
startedAt: null,
|
||||
})) as any[];
|
||||
const gate = deferred<any>();
|
||||
let started = 0;
|
||||
const projection = (rows: any[], shape?: Record<string, string>) =>
|
||||
shape
|
||||
? rows.map((row) =>
|
||||
Object.fromEntries(Object.entries(shape).map(([key, field]) => [key, row[field]])),
|
||||
)
|
||||
: rows;
|
||||
const query = (shape?: Record<string, string>, updates?: any) => {
|
||||
let predicate = (_row: any) => true;
|
||||
let limit = Infinity;
|
||||
const execute = () => {
|
||||
const rows = tasks.filter(predicate).slice(0, limit);
|
||||
if (updates)
|
||||
for (const row of rows)
|
||||
Object.assign(row, updates, {
|
||||
attempts: updates.status === "running" ? row.attempts + 1 : row.attempts,
|
||||
});
|
||||
return rows;
|
||||
};
|
||||
const builder: any = {
|
||||
from: () => builder,
|
||||
where: (condition: any) => {
|
||||
predicate = condition;
|
||||
return builder;
|
||||
},
|
||||
orderBy: () => builder,
|
||||
limit: (count: number) => {
|
||||
limit = count;
|
||||
return builder;
|
||||
},
|
||||
returning: async (returnShape: any) => projection(execute(), returnShape),
|
||||
then: (resolve: any, reject: any) =>
|
||||
Promise.resolve(projection(execute(), shape)).then(resolve, reject),
|
||||
};
|
||||
return builder;
|
||||
};
|
||||
const sqlMock = (_parts: TemplateStringsArray, ..._args: unknown[]) => ({});
|
||||
const taskTable = Object.fromEntries(
|
||||
[
|
||||
"id",
|
||||
"request",
|
||||
"userId",
|
||||
"status",
|
||||
"attempts",
|
||||
"createdAt",
|
||||
"startedAt",
|
||||
"completedAt",
|
||||
"result",
|
||||
"error",
|
||||
"updatedAt",
|
||||
].map((field) => [field, field]),
|
||||
);
|
||||
const module = serverLoader({
|
||||
"@/lib/db": {
|
||||
db: {
|
||||
select: (shape: any) => query(shape),
|
||||
update: () => ({ set: (updates: any) => query(undefined, updates) }),
|
||||
},
|
||||
},
|
||||
"@/lib/db/schema": { canvasImageTask: taskTable },
|
||||
"drizzle-orm": {
|
||||
eq: (field: string, value: unknown) => (row: any) => row[field] === value,
|
||||
and:
|
||||
(...predicates: any[]) =>
|
||||
(row: any) =>
|
||||
predicates.every((fn) => fn(row)),
|
||||
lt: (field: string, value: any) => (row: any) => row[field] != null && row[field] < value,
|
||||
asc: (field: unknown) => field,
|
||||
notInArray: (field: string, values: unknown[]) => (row: any) => !values.includes(row[field]),
|
||||
sql: sqlMock,
|
||||
},
|
||||
"./generate-image": {
|
||||
generateCanvasImage: async () => {
|
||||
started += 1;
|
||||
return gate.promise;
|
||||
},
|
||||
},
|
||||
"./scale-adjust-model-relight": {
|
||||
applyScaleAdjustModelRelight: async ({ initialResult }: any) => initialResult,
|
||||
},
|
||||
"@/lib/style-pro/task": {
|
||||
getStyleProTaskPrimaryModelKey: () => "style",
|
||||
executeStyleProTask: async () => {
|
||||
throw new Error("Unexpected style task");
|
||||
},
|
||||
},
|
||||
"@/lib/canvas/layout-replace-task": { isCanvasLayoutReplaceTaskRequest: () => false },
|
||||
"@/lib/canvas/layout-replace-batch": {},
|
||||
"@/lib/agents/agent-project-result-persistence": {},
|
||||
"@/lib/storage/image-archive-db": { resolveArchivedImageUrls: async (value: unknown) => value },
|
||||
})("lib/canvas/image-task-queue.ts");
|
||||
await module.enqueueCanvasImageTask(tasks[0].id);
|
||||
await expect.poll(() => started).toBe(20);
|
||||
await module.enqueueCanvasImageTask(tasks[0].id);
|
||||
expect(started).toBe(20);
|
||||
expect(tasks.every((task) => task.status === "running" && task.attempts === 1)).toBe(true);
|
||||
gate.resolve({
|
||||
images: [{ originalUrl: raw, thumbnailUrl: raw }],
|
||||
model: "one-model",
|
||||
prompt: "test",
|
||||
});
|
||||
await expect.poll(() => tasks.filter((task) => task.status === "succeeded").length).toBe(20);
|
||||
});
|
||||
|
||||
test("successful background archive replaces saved references immediately", async () => {
|
||||
let stored = false;
|
||||
let claimed = false;
|
||||
let completed = false;
|
||||
const calls: string[][] = [];
|
||||
const module = serverLoader({
|
||||
"@/lib/storage/image-archive": {
|
||||
archiveTemporaryImage: async () => ({ url: permanent, bytes: png.length }),
|
||||
},
|
||||
"@/lib/storage/image-archive-db": {
|
||||
recoverStaleImageArchiveTasks: async () => {},
|
||||
getImageArchiveTasksToStore: async () => [{ id: "task", sourceUrl: raw, metadata: null }],
|
||||
markImageArchiveTaskArchiving: async () => {
|
||||
if (claimed) return false;
|
||||
claimed = true;
|
||||
return true;
|
||||
},
|
||||
markImageArchiveTaskStored: async () => {
|
||||
stored = true;
|
||||
},
|
||||
markImageArchiveTaskFailed: async () => {
|
||||
throw new Error("Unexpected storage failure");
|
||||
},
|
||||
getImageArchiveTasksToReplace: async () =>
|
||||
stored && !completed ? [{ id: "task", sourceUrl: raw, targetUrl: permanent }] : [],
|
||||
markImageArchiveTaskReplacing: async () => true,
|
||||
markImageArchiveTaskCompleted: async () => {
|
||||
completed = true;
|
||||
},
|
||||
getImageArchiveTaskStats: async () => ({ pendingStore: 0, pendingReplace: 0 }),
|
||||
},
|
||||
"@/lib/storage/image-url-replacement": {
|
||||
replaceImageUrlInDatabase: async (...urls: string[]) => {
|
||||
calls.push(urls);
|
||||
return { changed: 2 };
|
||||
},
|
||||
},
|
||||
})("lib/jobs/image-archive-job.ts");
|
||||
const result = await module.runImageArchiveJob();
|
||||
expect(result.store.succeeded).toBe(1);
|
||||
expect(result.replace.succeeded).toBe(1);
|
||||
expect(calls).toEqual([[raw, permanent]]);
|
||||
await module.runImageArchiveJob();
|
||||
expect(calls).toHaveLength(1);
|
||||
});
|
||||
|
||||
test("inline preview serves exact image bytes then redirects to the stored object", async () => {
|
||||
const key = "a".repeat(64);
|
||||
let archived = false;
|
||||
const rows = () => [
|
||||
{
|
||||
targetUrl: archived ? permanent : null,
|
||||
metadata: { inlineSource: `data:image/png;base64,${png.toString("base64")}` },
|
||||
},
|
||||
];
|
||||
const module = serverLoader(
|
||||
{
|
||||
"@/lib/db": {
|
||||
db: { select: () => ({ from: () => ({ where: () => ({ limit: async () => rows() }) }) }) },
|
||||
},
|
||||
"@/lib/db/schema": { imageArchiveTask: { sourceUrl: "sourceUrl" } },
|
||||
"next/server": {
|
||||
NextResponse: class extends Response {
|
||||
static redirect(url: string, status: number) {
|
||||
return new Response(null, { status, headers: { Location: url } });
|
||||
}
|
||||
},
|
||||
},
|
||||
},
|
||||
{ NEXT_PUBLIC_APP_URL: "https://app.example" },
|
||||
)("app/api/generated-image-preview/[key]/route.ts");
|
||||
const response = await module.GET({}, { params: Promise.resolve({ key }) });
|
||||
expect(response.headers.get("Content-Type")).toBe("image/png");
|
||||
expect(Buffer.from(await response.arrayBuffer())).toEqual(png);
|
||||
archived = true;
|
||||
const redirect = await module.GET({}, { params: Promise.resolve({ key }) });
|
||||
expect(redirect.status).toBe(307);
|
||||
expect(redirect.headers.get("Location")).toBe(permanent);
|
||||
const invalid = await module.GET({}, { params: Promise.resolve({ key: "guess" }) });
|
||||
expect(invalid.status).toBe(404);
|
||||
});
|
||||
|
||||
test("archive lookup requires login and an owned canvas before reading image mappings", async () => {
|
||||
let signedIn = false;
|
||||
let reads = 0;
|
||||
const module = serverLoader({
|
||||
"@/app/(auth)/auth": { auth: async () => (signedIn ? { user: { id: "owner" } } : null) },
|
||||
"@/lib/db": {
|
||||
db: {
|
||||
select: () => {
|
||||
reads += 1;
|
||||
return { from: () => ({ where: () => ({ limit: async () => [] }) }) };
|
||||
},
|
||||
},
|
||||
},
|
||||
"@/lib/db/schema": {
|
||||
canvasProject: { id: "id", userId: "userId" },
|
||||
imageArchiveTask: {},
|
||||
generatedImage: {},
|
||||
canvasImageTask: {},
|
||||
},
|
||||
"@/lib/storage/image-archive-db": { imageArchiveSourceKey: (url: string) => url },
|
||||
"@/lib/storage/image-archive-worker": {
|
||||
scheduleImageArchiveWorker: () => {
|
||||
throw new Error("Unowned request must not start archiving");
|
||||
},
|
||||
},
|
||||
"next/server": {
|
||||
NextResponse: class extends Response {
|
||||
static json(body: unknown, init?: ResponseInit) {
|
||||
return new Response(JSON.stringify(body), init);
|
||||
}
|
||||
},
|
||||
},
|
||||
})("app/api/canvas/[id]/image-archives/route.ts");
|
||||
const request = { json: async () => ({ urls: [raw] }) };
|
||||
expect(
|
||||
(await module.POST(request, { params: Promise.resolve({ id: "other-project" }) })).status,
|
||||
).toBe(401);
|
||||
expect(reads).toBe(0);
|
||||
signedIn = true;
|
||||
expect(
|
||||
(await module.POST(request, { params: Promise.resolve({ id: "other-project" }) })).status,
|
||||
).toBe(404);
|
||||
expect(reads).toBe(1);
|
||||
});
|
||||
|
||||
test("replacement retries reconcile late writes once, without reclaiming completed jobs forever", async () => {
|
||||
const past = new Date(Date.now() - 1000);
|
||||
const future = new Date(Date.now() + 60_000);
|
||||
const rows = [
|
||||
{ id: "stored", status: "stored", replaceAfterAt: past },
|
||||
{ id: "settle", status: "completed", replaceAfterAt: past, metadata: { replacementPasses: 1 } },
|
||||
{
|
||||
id: "too-soon",
|
||||
status: "completed",
|
||||
replaceAfterAt: future,
|
||||
metadata: { replacementPasses: 1 },
|
||||
},
|
||||
{ id: "done", status: "completed", replaceAfterAt: past, metadata: { replacementPasses: 2 } },
|
||||
{ id: "legacy", status: "completed", replaceAfterAt: past },
|
||||
];
|
||||
const table = Object.fromEntries(
|
||||
["id", "status", "replaceAfterAt", "metadata", "createdAt", "updatedAt"].map((key) => [
|
||||
key,
|
||||
key,
|
||||
]),
|
||||
);
|
||||
const query = (updates?: any) => {
|
||||
let predicate = (_row: any) => true;
|
||||
const execute = () => {
|
||||
const found = rows.filter(predicate);
|
||||
if (updates) found.forEach((row) => Object.assign(row, updates));
|
||||
return found;
|
||||
};
|
||||
const builder: any = {
|
||||
from: () => builder,
|
||||
orderBy: () => builder,
|
||||
limit: () => builder,
|
||||
where: (value: any) => {
|
||||
predicate = value;
|
||||
return builder;
|
||||
},
|
||||
returning: async () => execute(),
|
||||
then: (resolve: any) => Promise.resolve(execute()).then(resolve),
|
||||
};
|
||||
return builder;
|
||||
};
|
||||
const module = serverLoader({
|
||||
"@/lib/db": {
|
||||
db: { select: () => query(), update: () => ({ set: (updates: any) => query(updates) }) },
|
||||
},
|
||||
"@/lib/db/schema": { imageArchiveTask: table },
|
||||
"drizzle-orm": {
|
||||
and:
|
||||
(...conditions: any[]) =>
|
||||
(row: any) =>
|
||||
conditions.every((fn) => fn(row)),
|
||||
or:
|
||||
(...conditions: any[]) =>
|
||||
(row: any) =>
|
||||
conditions.some((fn) => fn(row)),
|
||||
eq: (field: string, value: unknown) => (row: any) => row[field] === value,
|
||||
lte: (field: string, value: any) => (row: any) => row[field] <= value,
|
||||
asc: (field: unknown) => field,
|
||||
sql: (parts: TemplateStringsArray) => {
|
||||
if (!parts.join("").includes("replacementPasses")) throw new Error("Unexpected predicate");
|
||||
return (row: any) => row.metadata?.replacementPasses === 1;
|
||||
},
|
||||
},
|
||||
})("lib/storage/image-archive-db.ts");
|
||||
expect(
|
||||
(await module.getImageArchiveTasksToReplace({ limit: 50 })).map((row: any) => row.id),
|
||||
).toEqual(["stored", "settle"]);
|
||||
expect(await module.markImageArchiveTaskReplacing("settle")).toBe(true);
|
||||
expect(await module.markImageArchiveTaskReplacing("settle")).toBe(false);
|
||||
expect(await module.markImageArchiveTaskReplacing("done")).toBe(false);
|
||||
});
|
||||
|
||||
test("detached generation still archives when its original request has already closed", async () => {
|
||||
let runs = 0;
|
||||
const module = serverLoader({
|
||||
// A closed request may accept a callback that is never invoked.
|
||||
"next/server": { after: () => {} },
|
||||
"@/lib/jobs/image-archive-job": {
|
||||
runImageArchiveJob: async () => {
|
||||
runs += 1;
|
||||
return {
|
||||
store: { total: 0 },
|
||||
replace: { total: 0 },
|
||||
stats: { pendingStore: 0, pendingReplace: 0 },
|
||||
};
|
||||
},
|
||||
},
|
||||
})("lib/storage/image-archive-worker.ts");
|
||||
module.scheduleImageArchiveWorker();
|
||||
expect(runs).toBe(0);
|
||||
await expect.poll(() => runs, { timeout: 5_000 }).toBe(1);
|
||||
});
|
||||
Loading…
x
Reference in New Issue
Block a user