diff --git a/.env.example b/.env.example index e6705c96f..948807853 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/app/(admin)/settings/_components/local-ai-sections/runtime-sections.tsx b/app/(admin)/settings/_components/local-ai-sections/runtime-sections.tsx index b9da6a47c..397c31689 100644 --- a/app/(admin)/settings/_components/local-ai-sections/runtime-sections.tsx +++ b/app/(admin)/settings/_components/local-ai-sections/runtime-sections.tsx @@ -1078,139 +1078,15 @@ export function LocalAiRuntimeSections({ - {/* 全站生图单模型并发 */} - 控制全站生图链路中，同一个模型允许同时运行的任务数，以及排队后下一批任务的发起间隔。配置保存到数据库，异步队列和直调生成都会自动读取，无需重启服务。 - - } + description="生图任务直接发起，图片先显示，后台转存到存储桶。" expanded={isLocalAiSectionExpanded("canvas-image-task-queue")} id="local-ai-section-canvas-image-task-queue" onToggle={() => toggleLocalAiSection("canvas-image-task-queue")} - summary={ - - } - title={ - - - 全站生图单模型并发与批次间隔 - - } + summary={} + title={生图任务执行} > -
-
-
生效范围
-
-

- • - 影响画布异步生图、智能体直调生图、视频分镜高清预览等全站生图入口。 -

-

- • - 多用户共享同一套并发限制，但会按模型分别计数，不同模型之间不再互相阻塞。 -

-

- • 当前配置的是单模型并发，不是全站总并发，也不是单用户并发。 -

-

- • - 批次间隔只控制排队补位的“下一批”发起节奏，不影响首批任务即时启动。 -

-

- • 数值过高可能触发上游模型 429、超时或 504，建议配合模型 Key - 池继续细分。 -

-
-
- -
-
- - - setCanvasImageTaskQueueConfig( - normalizeCanvasImageTaskQueueConfig({ - ...canvasImageTaskQueueConfig, - concurrency: e.target.value, - }) - ) - } - type="number" - value={canvasImageTaskQueueConfig.concurrency} - /> -
- 可配置范围 {MIN_CANVAS_IMAGE_TASK_CONCURRENCY}- - {MAX_CANVAS_IMAGE_TASK_CONCURRENCY}，当前为{" "} - {canvasImageTaskQueueConfig.concurrency} 并发，默认值为{" "} - {DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY}。 -
-
- -
- - - setCanvasImageTaskQueueConfig( - normalizeCanvasImageTaskQueueConfig({ - ...canvasImageTaskQueueConfig, - batchDispatchIntervalMs: Math.round( - Number.parseFloat(e.target.value || "0") * 1000 - ), - }) - ) - } - step="0.5" - type="number" - value={String( - canvasImageTaskQueueConfig.batchDispatchIntervalMs / 1000 - )} - /> -
- 可配置范围 {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)}{" "} - 秒。 -
-
-
-
- -
- -
+

已取消模型及单用户并发限制，任务无需等待其他图片生成完成。任务状态、超时处理和失败重试继续保留。

 {/* 批量处理配置 */} diff --git a/app/(admin)/settings/admin-settings-client.tsx b/app/(admin)/settings/admin-settings-client.tsx index ec4273759..28f553080 100644 --- a/app/(admin)/settings/admin-settings-client.tsx +++ b/app/(admin)/settings/admin-settings-client.tsx @@ -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, diff --git a/app/api/canvas/[id]/image-archives/route.ts b/app/api/canvas/[id]/image-archives/route.ts new file mode 100644 index 000000000..68df25d96 --- /dev/null +++ b/app/api/canvas/[id]/image-archives/route.ts @@ -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(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 = {}; + 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" } }, + ); +} diff --git a/app/api/canvas/[id]/route.ts b/app/api/canvas/[id]/route.ts index a6aac174e..f2e7338b8 100644 --- a/app/api/canvas/[id]/route.ts +++ b/app/api/canvas/[id]/route.ts @@ -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 = diff --git a/app/api/generated-image-preview/[key]/route.ts b/app/api/generated-image-preview/[key]/route.ts new file mode 100644 index 000000000..58cb50330 --- /dev/null +++ b/app/api/generated-image-preview/[key]/route.ts @@ -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", + }, + }); +} diff --git a/app/canvas/[id]/CanvasEditorLegacyPageClient.tsx b/app/canvas/[id]/CanvasEditorLegacyPageClient.tsx index 75a2b3bf8..91a8c114d 100644 --- a/app/canvas/[id]/CanvasEditorLegacyPageClient.tsx +++ b/app/canvas/[id]/CanvasEditorLegacyPageClient.tsx @@ -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); diff --git a/hooks/use-canvas-image-archives.ts b/hooks/use-canvas-image-archives.ts new file mode 100644 index 000000000..e62d9b310 --- /dev/null +++ b/hooks/use-canvas-image-archives.ts @@ -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 { + 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(); + const unknownChecks = new Map(); + let timer: ReturnType | 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]); +} diff --git a/lib/ai/providers/atlascloud.ts b/lib/ai/providers/atlascloud.ts index 6921c42cc..21473dc6e 100644 --- a/lib/ai/providers/atlascloud.ts +++ b/lib/ai/providers/atlascloud.ts @@ -357,14 +357,12 @@ async function uploadImageBuffer(buffer: Buffer, mimeType: string) { } async function archiveOutputImage(imageUrl: string): Promise { + 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", diff --git a/lib/ai/providers/bananaimage.ts b/lib/ai/providers/bananaimage.ts index 1eb6ce9a5..5538df3fd 100644 --- a/lib/ai/providers/bananaimage.ts +++ b/lib/ai/providers/bananaimage.ts @@ -329,15 +329,13 @@ async function uploadImageBuffer( } async function archiveOutputImage(imageUrl: string): Promise { + 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", diff --git a/lib/ai/providers/gemini-image.ts b/lib/ai/providers/gemini-image.ts index b507a3940..b17f40b97 100644 --- a/lib/ai/providers/gemini-image.ts +++ b/lib/ai/providers/gemini-image.ts @@ -119,6 +119,9 @@ async function inferImageAspectRatioFromUrl(url: string): Promise * 将 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"); diff --git a/lib/ai/providers/huanke-gemini.ts b/lib/ai/providers/huanke-gemini.ts index aa11bee94..684318152 100644 --- a/lib/ai/providers/huanke-gemini.ts +++ b/lib/ai/providers/huanke-gemini.ts @@ -62,6 +62,9 @@ async function createThumbnail(buffer: Buffer, mimeType: string): Promise { + 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"); diff --git a/lib/ai/providers/openai-image.ts b/lib/ai/providers/openai-image.ts index afcf8fd82..0cbea2626 100644 --- a/lib/ai/providers/openai-image.ts +++ b/lib/ai/providers/openai-image.ts @@ -430,6 +430,9 @@ async function downloadAndUploadImage(imageUrl: string): Promise } 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( diff --git a/lib/ai/providers/snapgen-image.ts b/lib/ai/providers/snapgen-image.ts index 481f7ed3a..6d061157d 100644 --- a/lib/ai/providers/snapgen-image.ts +++ b/lib/ai/providers/snapgen-image.ts @@ -665,6 +665,9 @@ function generateFilename(prefix: string, extension: string) { } async function uploadImageBuffer(buffer: Buffer, extension: string): Promise { + 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); diff --git a/lib/ai/providers/stepfun-image.ts b/lib/ai/providers/stepfun-image.ts index 59cc55c36..06cd710f5 100644 --- a/lib/ai/providers/stepfun-image.ts +++ b/lib/ai/providers/stepfun-image.ts @@ -177,6 +177,9 @@ async function downloadAndUploadImage(imageUrl: string): Promise } 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( diff --git a/lib/ai/providers/vip98k.ts b/lib/ai/providers/vip98k.ts index f4c1b0a41..bcbd5ac83 100644 --- a/lib/ai/providers/vip98k.ts +++ b/lib/ai/providers/vip98k.ts @@ -511,14 +511,12 @@ async function uploadBuffer(buffer: Buffer, mimeType: string) { } async function archiveOutputImage(imageUrl: string): Promise { + 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", diff --git a/lib/ai/providers/visionary.ts b/lib/ai/providers/visionary.ts index 2e08dfd48..bd8005e90 100644 --- a/lib/ai/providers/visionary.ts +++ b/lib/ai/providers/visionary.ts @@ -269,15 +269,13 @@ async function uploadImageBuffer( } async function archiveOutputImage(imageUrl: string): Promise { + 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", diff --git a/lib/ai/providers/volcano-vision.ts b/lib/ai/providers/volcano-vision.ts index d4f5e60c6..2e565a789 100644 --- a/lib/ai/providers/volcano-vision.ts +++ b/lib/ai/providers/volcano-vision.ts @@ -428,9 +428,9 @@ async function createThumbnail(buffer: Buffer, mimeType: string): Promise { +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 { diff --git a/lib/ai/runtime/generated-image-persistence.ts b/lib/ai/runtime/generated-image-persistence.ts index 449ed76f0..3174b678c 100644 --- a/lib/ai/runtime/generated-image-persistence.ts +++ b/lib/ai/runtime/generated-image-persistence.ts @@ -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; diff --git a/lib/canvas/image-concurrency.ts b/lib/canvas/image-concurrency.ts index 1fe03a218..d41803e27 100644 --- a/lib/canvas/image-concurrency.ts +++ b/lib/canvas/image-concurrency.ts @@ -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; - waiters: ModelWaiter[]; - dispatchTimer: ReturnType | null; - delayedUntil: number | null; - recentExecutionDurationsMs: number[]; -} - -interface CanvasImageConcurrencyState { - models: Map; -} - -let cachedConfig: - | { - expiresAt: number; - value: CanvasImageTaskQueueConfig; - } - | null = null; - -function getGlobalConcurrencyState(): CanvasImageConcurrencyState { +const stateKey = "__unrestricted_canvas_image_models__"; +type ModelState = { running: Set; durations: number[] }; +function getModels() { const globalState = globalThis as typeof globalThis & { - [concurrencyStateKey]?: CanvasImageConcurrencyState; + [stateKey]?: Map; }; - - if (!globalState[concurrencyStateKey]) { - globalState[concurrencyStateKey] = { - models: new Map(), - }; - } - - 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(), - 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 { - 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 { - return (await getCanvasImageTaskQueueConfig()).concurrency; -} - -export async function getCanvasImageBatchDispatchIntervalMs(): Promise { - 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 { - const normalizedModelKey = normalizeCanvasImageModelKey(params.modelKey); - const { concurrency: maxConcurrent } = await getCanvasImageTaskQueueConfig(); - const modelState = - getGlobalConcurrencyState().models.get(normalizedModelKey) || { - running: new Map(), - 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; onQueued?: (status: CanvasImageModelQueueStatus) => void | Promise; }) { - 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((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, }; } diff --git a/lib/canvas/image-task-config.ts b/lib/canvas/image-task-config.ts index 6a1aa4ab6..066df0806 100644 --- a/lib/canvas/image-task-config.ts +++ b/lib/canvas/image-task-config.ts @@ -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) - : {}; - - 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 }; } diff --git a/lib/canvas/image-task-queue.ts b/lib/canvas/image-task-queue.ts index b2c67f6ae..2a3f84868 100644 --- a/lib/canvas/image-task-queue.ts +++ b/lib/canvas/image-task-queue.ts @@ -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; running: Map; runningByModel: Map>; - runningByModelUser: Map>; - nextBatchEligibleAt: Map; - nextBatchTimers: Map>; - awaitingNextBatch: Set; 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(), running: new Map(), runningByModel: new Map>(), - runningByModelUser: new Map>(), - nextBatchEligibleAt: new Map(), - nextBatchTimers: new Map>(), - awaitingNextBatch: new Set(), 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(); tasks.add(taskId); state.runningByModel.set(modelKey, tasks); - - const modelUserKey = getModelUserKey(modelKey, userId); - const userTasks = state.runningByModelUser.get(modelUserKey) || new Set(); - 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 { +async function claimNextPendingTask(): Promise { 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, }; } diff --git a/lib/db/queries.ts b/lib/db/queries.ts index dfe2c7ef1..e01dc109a 100644 --- a/lib/db/queries.ts +++ b/lib/db/queries.ts @@ -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 = db) { + params = { ...params, canvasData: await resolveArchivedImageUrls(params.canvasData) }; const nodes = Array.isArray(params.canvasData?.nodes) ? params.canvasData?.nodes : []; const assetUrls = Array.from( new Set( diff --git a/lib/jobs/image-archive-job.ts b/lib/jobs/image-archive-job.ts index 3f6589d8d..dcaffe682 100644 --- a/lib/jobs/image-archive-job.ts +++ b/lib/jobs/image-archive-job.ts @@ -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, diff --git a/lib/storage/image-archive-db.ts b/lib/storage/image-archive-db.ts index bfef74a20..25afc457d 100644 --- a/lib/storage/image-archive-db.ts +++ b/lib/storage/image-archive-db.ts @@ -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 | 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`count(*)::int`, + reconciliationCount: sql`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 = {}; + 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(value: T): Promise { + const sources = collectImageArchiveSources(value); + if (!sources.length) return value; + return replaceArchivedImageUrls(value, await getArchivedImageUrlMappings(sources)); +} diff --git a/lib/storage/image-archive-urls.ts b/lib/storage/image-archive-urls.ts new file mode 100644 index 000000000..344623a18 --- /dev/null +++ b/lib/storage/image-archive-urls.ts @@ -0,0 +1,56 @@ +// Shared by the server and canvas; no storage or network work happens here. +export type ArchivedImageUrlMap = Record; + +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(); + 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(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 `![${nextAlt}](${replacements[url] || url})`; + }) 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 = {}; + 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; +} diff --git a/lib/storage/image-archive-worker.ts b/lib/storage/image-archive-worker.ts new file mode 100644 index 000000000..7568148dd --- /dev/null +++ b/lib/storage/image-archive-worker.ts @@ -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; +}; +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(); + }); + } +} diff --git a/lib/storage/image-archive.ts b/lib/storage/image-archive.ts index b0ea3416f..0cb533936 100644 --- a/lib/storage/image-archive.ts +++ b/lib/storage/image-archive.ts @@ -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 }) => - `![thumb|||${originalUrl}|||${params.imageSize}](${thumbnailUrl})` + `![thumb|||${originalUrl}|||${params.imageSize}](${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 = { "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), diff --git a/lib/storage/image-url-replacement.ts b/lib/storage/image-url-replacement.ts index aad3500b4..240d856e2 100644 --- a/lib/storage/image-url-replacement.ts +++ b/lib/storage/image-url-replacement.ts @@ -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 diff --git a/store/canvas/index.ts b/store/canvas/index.ts index ce3ffd4f3..3640e4f08 100644 --- a/store/canvas/index.ts +++ b/store/canvas/index.ts @@ -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) => string; updateNode: (id: string, updates: Partial) => void; + replaceArchivedImageUrls: (replacements: ArchivedImageUrlMap) => void; deleteNode: (id: string) => void; deleteSelectedNodes: () => void; addSuppressedMediaKeys: (keys: string[]) => void; @@ -410,6 +412,19 @@ export const useCanvasStore = create()( } }), + 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)) diff --git a/tests/image-archive.config.ts b/tests/image-archive.config.ts new file mode 100644 index 000000000..b514ad791 --- /dev/null +++ b/tests/image-archive.config.ts @@ -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, +}); diff --git a/tests/routes/image-archive-flow.test.ts b/tests/routes/image-archive-flow.test.ts new file mode 100644 index 000000000..a3ca837b1 --- /dev/null +++ b/tests/routes/image-archive-flow.test.ts @@ -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() { + let resolve!: (value: T) => void; + const promise = new Promise((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, env: Record = {}) { + const modules = new Map(); + 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(); + 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> = []; + let runs = 0; + const module = serverLoader({ + "next/server": { + after: (fn: () => Promise) => { + 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 = `![thumb|||${raw}|||2K](${raw})`; + expect(replaceArchivedImageUrls(markdown, { [raw]: permanent })).toBe( + `![thumb|||${permanent}|||2K](${permanent})`, + ); + 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(); + let started = 0; + const projection = (rows: any[], shape?: Record) => + shape + ? rows.map((row) => + Object.fromEntries(Object.entries(shape).map(([key, field]) => [key, row[field]])), + ) + : rows; + const query = (shape?: Record, 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); +});