Merge pull request '原图下载等待七牛准备并拆分图片存储队列' (#86) from codex/direct-download-archive-20261113 into main
This commit is contained in:
commit
17075e38bd
@ -26,9 +26,10 @@ function failed(failure: DownloadFailure, requestId: string) {
|
||||
});
|
||||
}
|
||||
|
||||
function preparing(requestId: string) {
|
||||
function preparing(requestId: string, converting = false) {
|
||||
return NextResponse.json({ delivery: "preparing", retryAfter: 2, requestId,
|
||||
message: "正在准备原图，完成后将自动开始下载" }, {
|
||||
phase: converting ? "converting" : "preparing",
|
||||
message: converting ? "正在转换为兼容 PNG 并保存，完成后自动下载" : "正在准备原图，完成后将自动开始下载" }, {
|
||||
status: 202, headers: { "Retry-After": "2", "Cache-Control": "no-store" },
|
||||
});
|
||||
}
|
||||
@ -119,23 +120,23 @@ export async function POST(request: NextRequest) {
|
||||
|
||||
const name = sanitizeCanvasDownloadFileName(String(body.name || ""), "canvas-image");
|
||||
if (body.preferDirect === true) {
|
||||
const { task, direct, canPrepareDirect } = await resolveCanvasCdnDownload(urls, `${name}.png`).catch((error) => {
|
||||
logger.warn("cdn_lookup_failed", error, { requestId });
|
||||
return { task: null, direct: null, canPrepareDirect: false };
|
||||
});
|
||||
const { task, direct, canPrepareDirect } = await resolveCanvasCdnDownload(urls, `${name}.png`);
|
||||
if (direct) return NextResponse.json(direct, { headers: { "Cache-Control": "no-store" } });
|
||||
if (task && canPrepareDirect && body.preparation === "async") {
|
||||
if (task && canPrepareDirect) {
|
||||
const asset = (task.metadata as { downloadAsset?: { status?: string; failure?: DownloadFailure; retryAt?: string } } | null)?.downloadAsset;
|
||||
if (asset?.status === "abandoned" || (asset?.status === "failed" && Date.parse(asset.retryAt || "") > Date.now())) {
|
||||
return failed(asset.failure || getDownloadFailure(null), task.id);
|
||||
}
|
||||
scheduleImageDownloadPreparation(task.id);
|
||||
return preparing(task.id);
|
||||
return preparing(task.id, true);
|
||||
}
|
||||
if (task?.targetUrl) {
|
||||
if (!urls.includes(task.targetUrl)) urls.unshift(task.targetUrl);
|
||||
scheduleImageDownloadPreparation(task.id);
|
||||
if (task && !task.targetUrl) {
|
||||
const { originalImagePreparing, prepareOriginalImageArchive } = await import("@/lib/storage/original-image-download");
|
||||
const pending = await prepareOriginalImageArchive(task.sourceUrl);
|
||||
return NextResponse.json(originalImagePreparing(pending), { status: 202,
|
||||
headers: { "Retry-After": "3", "Cache-Control": "no-store" } });
|
||||
}
|
||||
throw new Error("下载文件暂时无法从存储桶直接下载");
|
||||
}
|
||||
const cookie = request.headers.get("cookie") || "";
|
||||
if (body.preparation === "async") {
|
||||
|
||||
@ -367,7 +367,9 @@ export async function GET(request: NextRequest) {
|
||||
buildImageProxyUrl(target.toString(), { delivery: "original" });
|
||||
const { resolveOriginalCdnDownload } = await import("@/lib/storage/original-image-download");
|
||||
const direct = await resolveOriginalCdnDownload(target.toString(), searchParams.get("name") || "", request.signal);
|
||||
return NextResponse.json(direct || { delivery: "proxy" }, { headers: { "Cache-Control": "no-store" } });
|
||||
if (!direct) throw new Error("下载文件暂时无法从存储桶直接下载");
|
||||
return NextResponse.json(direct, { status: direct.delivery === "preparing" ? 202 : 200,
|
||||
headers: { "Cache-Control": "no-store", ...(direct.delivery === "preparing" ? { "Retry-After": "3" } : {}) } });
|
||||
} catch (error) {
|
||||
return downloadFailure(error);
|
||||
}
|
||||
|
||||
@ -201,7 +201,8 @@ async function downloadOriginalImageFile(params: DownloadCanvasImageParams): Pro
|
||||
const { url, name, signal, onProgress } = params;
|
||||
const source = String(url || "").trim();
|
||||
if (!source) throw new Error("图片地址无效");
|
||||
const proxy = buildImageProxyUrl(source, { delivery: "original" });
|
||||
const sourceUrl = new URL(source, window.location.origin);
|
||||
const proxy = buildImageProxyUrl(/^https?:$/.test(sourceUrl.protocol) ? sourceUrl.toString() : source, { delivery: "original" });
|
||||
const filenameBase = sanitizeDownloadName(name || inferDownloadNameFromUrl(source), "image");
|
||||
const target = new URL(proxy, window.location.origin);
|
||||
onProgress?.({ phase: "preparing", message: "正在准备原图..." });
|
||||
@ -223,8 +224,20 @@ async function downloadOriginalImageFile(params: DownloadCanvasImageParams): Pro
|
||||
if (target.origin === window.location.origin && target.pathname === "/api/image-proxy") {
|
||||
target.searchParams.set("download", "resolve");
|
||||
target.searchParams.set("name", filenameBase);
|
||||
const metadata = await checkedFetch(target.pathname + target.search);
|
||||
const delivery = await metadata.json();
|
||||
const deadline = Date.now() + 180_000;
|
||||
let delivery: any;
|
||||
for (let poll = 0; ; poll++) {
|
||||
signal?.throwIfAborted();
|
||||
const metadata = await checkedFetch(target.pathname + target.search);
|
||||
delivery = await metadata.json();
|
||||
if (metadata.status !== 202) break;
|
||||
if (delivery.delivery !== "preparing") throw new Error("下载准备状态无效");
|
||||
if (Date.now() >= deadline || poll >= 60) {
|
||||
throw new CanvasImageDownloadError("原图仍在准备中", { code: "preparation_timeout", requestId: delivery.requestId });
|
||||
}
|
||||
onProgress?.({ phase: "preparing", message: delivery.message || "正在准备原图，完成后自动下载" });
|
||||
await waitForRetry(Math.max(1, Math.min(5, Number(delivery.retryAfter) || 3)) * 1000, signal);
|
||||
}
|
||||
if (delivery.delivery === "cdn") {
|
||||
// The resolver names the attachment using the stored MIME type.
|
||||
const extension = delivery.filename.split(".").pop();
|
||||
@ -235,7 +248,7 @@ async function downloadOriginalImageFile(params: DownloadCanvasImageParams): Pro
|
||||
onProgress?.({ phase: "downloading", message: "已开始下载，可在浏览器下载列表查看进度" });
|
||||
return result;
|
||||
}
|
||||
if (delivery.delivery !== "proxy") throw new Error("下载链接无效");
|
||||
throw new CanvasImageDownloadError("原图尚未准备完成", { code: "preparation_timeout", requestId: delivery.requestId });
|
||||
}
|
||||
const response = await checkedFetch(proxy);
|
||||
const contentType = response.headers.get("content-type") || "";
|
||||
@ -377,8 +390,8 @@ async function downloadPngFile({
|
||||
throw new CanvasImageDownloadError("原图仍在准备中", { code: "preparation_timeout", status: 202, requestId: preparing.requestId });
|
||||
}
|
||||
onProgress?.({
|
||||
phase: preparationPhase,
|
||||
message: isConverting ? conversionMessage : preparing.message || "正在准备原图，完成后将自动下载",
|
||||
phase: preparing.phase || preparationPhase,
|
||||
message: preparing.message || (isConverting ? conversionMessage : "正在准备原图，完成后将自动下载"),
|
||||
});
|
||||
const delaySeconds = Math.max(preparationPolls > 5 ? 5 : 1, Math.min(5, Number(preparing.retryAfter) || 3));
|
||||
await waitForRetry(delaySeconds * 1000, signal);
|
||||
|
||||
@ -15,11 +15,56 @@ import {
|
||||
markImageArchiveTaskReplaceFailed,
|
||||
markImageArchiveTaskReplacing,
|
||||
markImageArchiveTaskStored,
|
||||
claimNextImageArchiveTask,
|
||||
type ImageArchiveTaskRecord,
|
||||
} from "@/lib/storage/image-archive-db";
|
||||
import { replaceImageUrlInDatabase } from "@/lib/storage/image-url-replacement";
|
||||
|
||||
const DEFAULT_BATCH_SIZE = 10;
|
||||
|
||||
async function storeClaimedImage(task: ImageArchiveTaskRecord) {
|
||||
try {
|
||||
const metadata = task.metadata as { inlineSource?: string } | null;
|
||||
const archived = await archiveTemporaryImage(metadata?.inlineSource || task.sourceUrl, task.id);
|
||||
await markImageArchiveTaskStored({ id: task.id, targetUrl: archived.url, downloadAsset: archived.downloadAsset });
|
||||
if (!archived.downloadAsset) scheduleImageDownloadPreparation(task.id);
|
||||
console.log("[ImageArchiveJob] stored", { taskId: task.id, bytes: archived.bytes, targetUrl: archived.url });
|
||||
return true;
|
||||
} catch (error) {
|
||||
await markImageArchiveTaskFailed({ id: task.id, error });
|
||||
console.error("[ImageArchiveJob] store failed", { taskId: task.id, error: String(error) });
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export async function storeNextImageArchive(preferredTaskId?: string) {
|
||||
return trackRuntimeWork(async () => {
|
||||
const task = await claimNextImageArchiveTask(preferredTaskId);
|
||||
if (!task) return false;
|
||||
await storeClaimedImage(task);
|
||||
return true;
|
||||
}, { durable: true });
|
||||
}
|
||||
|
||||
export async function replacePendingImageArchives() {
|
||||
return trackRuntimeWork(async () => {
|
||||
const tasks = await getImageArchiveTasksToReplace({ limit: 10 });
|
||||
let claimed = 0;
|
||||
for (const task of tasks) {
|
||||
if (!(await markImageArchiveTaskReplacing(task.id))) continue;
|
||||
claimed++;
|
||||
try {
|
||||
if (!task.targetUrl) throw new Error("Missing targetUrl");
|
||||
const result = await replaceImageUrlInDatabase(task.sourceUrl, task.targetUrl);
|
||||
await markImageArchiveTaskCompleted({ id: task.id, replacedCount: result.changed });
|
||||
} catch (error) {
|
||||
await markImageArchiveTaskReplaceFailed({ id: task.id, error });
|
||||
}
|
||||
}
|
||||
return claimed;
|
||||
}, { durable: true });
|
||||
}
|
||||
|
||||
function getBatchSize() {
|
||||
const parsed = Number(process.env.IMAGE_ARCHIVE_BATCH_SIZE || DEFAULT_BATCH_SIZE);
|
||||
if (!Number.isFinite(parsed)) return DEFAULT_BATCH_SIZE;
|
||||
@ -38,32 +83,9 @@ export async function runImageArchiveJob({ preferredTaskId }: { preferredTaskId?
|
||||
let storeFailed = 0;
|
||||
|
||||
const storeOne = async (task: (typeof storeTasks)[number]) => {
|
||||
try {
|
||||
if (!(await markImageArchiveTaskArchiving(task.id))) return;
|
||||
const metadata = task.metadata as { inlineSource?: string } | null;
|
||||
const archived = await archiveTemporaryImage(metadata?.inlineSource || task.sourceUrl, task.id);
|
||||
await markImageArchiveTaskStored({
|
||||
id: task.id,
|
||||
targetUrl: archived.url,
|
||||
downloadAsset: archived.downloadAsset,
|
||||
});
|
||||
if (!archived.downloadAsset) scheduleImageDownloadPreparation(task.id);
|
||||
storeSucceeded += 1;
|
||||
console.log("[ImageArchiveJob] stored", {
|
||||
taskId: task.id,
|
||||
bytes: archived.bytes,
|
||||
sourceUrl: task.sourceUrl,
|
||||
targetUrl: archived.url,
|
||||
});
|
||||
} catch (error) {
|
||||
storeFailed += 1;
|
||||
await markImageArchiveTaskFailed({ id: task.id, error });
|
||||
console.error("[ImageArchiveJob] store failed", {
|
||||
taskId: task.id,
|
||||
sourceUrl: task.sourceUrl,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
}
|
||||
if (!(await markImageArchiveTaskArchiving(task.id))) return;
|
||||
if (await storeClaimedImage(task)) storeSucceeded++;
|
||||
else storeFailed++;
|
||||
};
|
||||
|
||||
// Bucket work has a small worker pool independent of unrestricted model generation.
|
||||
|
||||
@ -2,6 +2,7 @@ import "server-only";
|
||||
import { getDefaultStorageConfig } from "@/lib/db/queries";
|
||||
import { ensureStoredImageDownloadTask, findImageArchiveDownloadTask } from "./image-archive-db";
|
||||
import { extractImageProxySourceUrl, inferOriginalImageUrl } from "@/lib/image-url";
|
||||
import { prepareOriginalImageArchive } from "./original-image-download";
|
||||
|
||||
export async function resolveCanvasCdnDownload(urls: string[], filename: string) {
|
||||
const sources = Array.from(new Set(urls.flatMap(url => {
|
||||
@ -27,6 +28,7 @@ export async function resolveCanvasCdnDownload(urls: string[], filename: string)
|
||||
}
|
||||
}
|
||||
}
|
||||
if (!task) task = await prepareOriginalImageArchive(sources[0]);
|
||||
const asset = (task?.metadata as { downloadAsset?: { status?: string; url?: string; contentType?: string; bytes?: number } } | null)?.downloadAsset;
|
||||
const ready = asset?.status === "ready" && asset.url && asset.contentType === "image/png";
|
||||
const source = ready ? asset.url : task?.targetUrl;
|
||||
|
||||
@ -137,6 +137,32 @@ export async function markImageArchiveTaskArchiving(id: string) {
|
||||
return claimed.length > 0;
|
||||
}
|
||||
|
||||
export async function claimNextImageArchiveTask(preferredTaskId?: string) {
|
||||
return db.transaction(async (tx) => {
|
||||
const priority = preferredTaskId
|
||||
? [asc(sql`case when ${imageArchiveTask.id} = ${preferredTaskId} then 0 else 1 end`)] : [];
|
||||
const [task] = await tx.select().from(imageArchiveTask).where(and(
|
||||
inArray(imageArchiveTask.status, ["pending", "failed"]),
|
||||
lte(imageArchiveTask.archiveAfterAt, new Date()),
|
||||
lt(imageArchiveTask.attemptCount, MAX_IMAGE_ARCHIVE_ATTEMPTS),
|
||||
)).orderBy(...priority, asc(imageArchiveTask.createdAt), asc(imageArchiveTask.id))
|
||||
.limit(1).for("update", { skipLocked: true });
|
||||
if (!task) return null;
|
||||
await tx.update(imageArchiveTask).set({ status: "archiving",
|
||||
attemptCount: sql`${imageArchiveTask.attemptCount} + 1`, lastError: null, updatedAt: new Date(),
|
||||
}).where(eq(imageArchiveTask.id, task.id));
|
||||
return task;
|
||||
});
|
||||
}
|
||||
|
||||
export async function hasPendingImageArchives() {
|
||||
const [task] = await db.select({ id: imageArchiveTask.id }).from(imageArchiveTask).where(or(
|
||||
inArray(imageArchiveTask.status, ["pending", "failed", "archiving", "stored", "replacing"]),
|
||||
and(eq(imageArchiveTask.status, "completed"), sql`${imageArchiveTask.metadata}->>'replacementPasses' = '1'`),
|
||||
)).limit(1);
|
||||
return Boolean(task);
|
||||
}
|
||||
|
||||
export async function markImageArchiveTaskStored(params: {
|
||||
id: string;
|
||||
targetUrl: string;
|
||||
|
||||
@ -1,70 +1,88 @@
|
||||
import "server-only";
|
||||
import { after } from "next/server";
|
||||
|
||||
const workerKey = "__generated_image_archive_worker__";
|
||||
const workerKey = "__generated_image_archive_worker_v2__";
|
||||
type Worker = {
|
||||
running: boolean;
|
||||
active: number;
|
||||
scheduled: boolean;
|
||||
requested: boolean;
|
||||
preferredTaskId?: string;
|
||||
priority: Set<string>;
|
||||
replacing: boolean;
|
||||
replaceRequested: boolean;
|
||||
retryTimer?: ReturnType<typeof setTimeout>;
|
||||
};
|
||||
function getWorker(): Worker {
|
||||
const root = globalThis as typeof globalThis & { [workerKey]?: Worker };
|
||||
return (root[workerKey] ??= { running: false, scheduled: false, requested: false });
|
||||
return (root[workerKey] ??= { active: 0, scheduled: false, priority: new Set(), replacing: false, replaceRequested: false });
|
||||
}
|
||||
|
||||
async function retryIfNeeded() {
|
||||
const worker = getWorker();
|
||||
if (worker.active || worker.replacing || worker.retryTimer) return;
|
||||
const { hasPendingImageArchives, hasPendingImageDownloads } = await import("./image-archive-db");
|
||||
if (!(await hasPendingImageArchives()) && !(await hasPendingImageDownloads())) return;
|
||||
if (worker.retryTimer) return;
|
||||
worker.retryTimer = setTimeout(() => {
|
||||
worker.retryTimer = undefined;
|
||||
scheduleImageArchiveWorker();
|
||||
}, 60_000);
|
||||
worker.retryTimer.unref?.();
|
||||
}
|
||||
|
||||
function scheduleReplacement() {
|
||||
const worker = getWorker();
|
||||
worker.replaceRequested = true;
|
||||
if (worker.replacing) return;
|
||||
worker.replacing = true;
|
||||
void (async () => {
|
||||
const { replacePendingImageArchives } = await import("@/lib/jobs/image-archive-job");
|
||||
do {
|
||||
worker.replaceRequested = false;
|
||||
while (await replacePendingImageArchives()) { /* Drain references independently of uploads. */ }
|
||||
} while (worker.replaceRequested);
|
||||
})().catch(error => console.error("[ImageArchiveWorker] replacement failed", error)).finally(() => {
|
||||
worker.replacing = false;
|
||||
void retryIfNeeded().catch(error => console.error("[ImageArchiveWorker] retry failed", error));
|
||||
});
|
||||
}
|
||||
|
||||
export function scheduleImageArchiveWorker(deliveryConfirmed = false, preferredTaskId?: string) {
|
||||
const worker = getWorker();
|
||||
worker.preferredTaskId ??= preferredTaskId;
|
||||
worker.requested = true;
|
||||
if (worker.running || (worker.scheduled && !deliveryConfirmed)) return;
|
||||
if (preferredTaskId) worker.priority.add(preferredTaskId);
|
||||
if (worker.scheduled && !deliveryConfirmed) return;
|
||||
worker.scheduled = true;
|
||||
if (worker.retryTimer) {
|
||||
clearTimeout(worker.retryTimer);
|
||||
worker.retryTimer = undefined;
|
||||
}
|
||||
const run = async () => {
|
||||
if (worker.running) return;
|
||||
if (!worker.scheduled) 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 nextTaskId = worker.preferredTaskId;
|
||||
worker.preferredTaskId = undefined;
|
||||
const result = await runImageArchiveJob({ preferredTaskId: nextTaskId });
|
||||
retryNeeded = result.stats.pendingStore > 0 || result.stats.pendingReplace > 0 || result.pendingDownloads === true;
|
||||
// Stop when only backoff tasks remain; do not spin while storage is down.
|
||||
if (!worker.requested && result.store.total === 0 && result.replace.total === 0 && result.downloadTasks === 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?.();
|
||||
}
|
||||
if (worker.retryTimer) { clearTimeout(worker.retryTimer); worker.retryTimer = undefined; }
|
||||
const { storeNextImageArchive } = await import("@/lib/jobs/image-archive-job");
|
||||
const { recoverStaleImageArchiveTasks } = await import("./image-archive-db");
|
||||
const { scheduleImageDownloadPreparation } = await import("./image-download-preparation");
|
||||
await recoverStaleImageArchiveTasks();
|
||||
scheduleImageDownloadPreparation();
|
||||
scheduleReplacement();
|
||||
// Free slots take the next file immediately; slow files and reference updates
|
||||
// cannot hold an entire batch or delay newly generated images.
|
||||
while (worker.active < 4) {
|
||||
worker.active++;
|
||||
void (async () => {
|
||||
for (;;) {
|
||||
const id = worker.priority.values().next().value;
|
||||
if (id) worker.priority.delete(id);
|
||||
if (!(await storeNextImageArchive(id))) break;
|
||||
scheduleReplacement();
|
||||
}
|
||||
})().catch(error => console.error("[ImageArchiveWorker] storage failed", error)).finally(() => {
|
||||
worker.active--;
|
||||
if (worker.priority.size) scheduleImageArchiveWorker();
|
||||
else void retryIfNeeded().catch(error => console.error("[ImageArchiveWorker] retry failed", error));
|
||||
});
|
||||
}
|
||||
};
|
||||
// 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();
|
||||
});
|
||||
}
|
||||
const safeRun = () => run().catch(error => {
|
||||
console.error("[ImageArchiveWorker] scheduling failed", error);
|
||||
void retryIfNeeded().catch(error => console.error("[ImageArchiveWorker] retry failed", error));
|
||||
});
|
||||
const fallback = setTimeout(() => { if (worker.scheduled) void safeRun(); }, 1_000);
|
||||
fallback.unref?.();
|
||||
try { after(safeRun); } catch { setImmediate(safeRun); }
|
||||
}
|
||||
|
||||
@ -58,7 +58,8 @@ export async function registerGeneratedImageTemporaryUrls(params: {
|
||||
displayHoldMs: getImageArchiveDisplayHoldMs(),
|
||||
},
|
||||
});
|
||||
scheduleImageArchiveWorker();
|
||||
for (const task of tasks) scheduleImageArchiveWorker(false, task.id);
|
||||
if (!tasks.length) scheduleImageArchiveWorker();
|
||||
return tasks;
|
||||
} catch (error) {
|
||||
logger.warn("enqueue temporary image archive failed", {
|
||||
@ -199,10 +200,10 @@ export async function archiveTemporaryImage(sourceUrl: string, archiveId?: strin
|
||||
});
|
||||
if (direct) return direct;
|
||||
}
|
||||
const downloaded = await downloadTemporaryImage({ sourceUrl });
|
||||
const downloaded = await downloadTemporaryImage({ sourceUrl, retries: 0 });
|
||||
const objectSuffix = `${Date.now()}-${randomUUID()}`;
|
||||
const filename = `ai-generated-archive-${objectSuffix}.${downloaded.extension}`;
|
||||
const uploaded = await uploadFile(filename, toArrayBuffer(downloaded.buffer));
|
||||
const uploaded = await uploadFile(filename, toArrayBuffer(downloaded.buffer), { preferQiniuForm: true });
|
||||
await cacheImageBytes(`source::${uploaded.url}`, downloaded);
|
||||
return {
|
||||
url: uploaded.url,
|
||||
|
||||
@ -1,11 +1,31 @@
|
||||
import "server-only";
|
||||
import { getDefaultStorageConfig } from "@/lib/db/queries";
|
||||
import { findImageArchiveDownloadTask } from "./image-archive-db";
|
||||
import { enqueueImageArchiveTask, findImageArchiveDownloadTask } from "./image-archive-db";
|
||||
import { inferOriginalImageUrl } from "@/lib/image-url";
|
||||
import { getImageDownloadExtension } from "@/lib/image-delivery";
|
||||
import { sanitizeCanvasDownloadFileName } from "@/lib/canvas/photoshop-safe-download";
|
||||
import { HeadObjectCommand } from "@aws-sdk/client-s3";
|
||||
|
||||
export async function prepareOriginalImageArchive(source: string) {
|
||||
const original = inferOriginalImageUrl(source) || source;
|
||||
const task = await findImageArchiveDownloadTask([...new Set([source, original])])
|
||||
|| await enqueueImageArchiveTask({ sourceUrl: original, metadata: { requestedDownload: true } });
|
||||
if (!task) throw new Error("下载文件尚未保存到存储桶");
|
||||
if (!task.targetUrl) {
|
||||
if (task.status === "abandoned" || (task.status === "failed" && task.archiveAfterAt > new Date())) {
|
||||
throw new Error(task.lastError || "下载文件保存到存储桶失败");
|
||||
}
|
||||
const { scheduleImageArchiveWorker } = await import("./image-archive-worker");
|
||||
scheduleImageArchiveWorker(false, task.id);
|
||||
}
|
||||
return task;
|
||||
}
|
||||
|
||||
export function originalImagePreparing(task: { id: string; status: string }) {
|
||||
return { delivery: "preparing" as const, phase: "preparing" as const, retryAfter: 3, requestId: task.id,
|
||||
message: task.status === "archiving" ? "正在保存原图，完成后自动下载" : "正在排队准备原图，完成后自动下载" };
|
||||
}
|
||||
|
||||
/** Resolve the archived original without waiting for a separate PNG conversion. */
|
||||
export async function resolveOriginalCdnDownload(source: string, name: string, signal: AbortSignal) {
|
||||
const config = await getDefaultStorageConfig();
|
||||
@ -14,12 +34,9 @@ export async function resolveOriginalCdnDownload(source: string, name: string, s
|
||||
const origin = process.env.CANVAS_QINIU_ORIGIN_URL;
|
||||
let target = new URL(source);
|
||||
if (target.origin !== base.origin && (!origin || target.origin !== new URL(origin).origin)) {
|
||||
const task = await findImageArchiveDownloadTask([...new Set([source, inferOriginalImageUrl(source)])]);
|
||||
if (task?.status === "pending" && !task.targetUrl) {
|
||||
const { scheduleImageArchiveWorker } = await import("./image-archive-worker");
|
||||
scheduleImageArchiveWorker(false, task.id);
|
||||
}
|
||||
target = new URL(task?.targetUrl || source);
|
||||
const task = await prepareOriginalImageArchive(source);
|
||||
if (!task.targetUrl) return originalImagePreparing(task);
|
||||
target = new URL(task.targetUrl);
|
||||
}
|
||||
if (origin && target.origin === new URL(origin).origin && !target.searchParams.has("token")) {
|
||||
target.host = base.host;
|
||||
|
||||
@ -12,6 +12,7 @@ export default defineConfig({
|
||||
/routes\/canvas-download-rate-limit\.test\.ts/,
|
||||
/routes\/canvas-zip-download\.test\.ts/,
|
||||
/routes\/image-archive-flow\.test\.ts/,
|
||||
/routes\/image-archive-lanes\.test\.ts/,
|
||||
/routes\/runtime-active-work\.test\.ts/,
|
||||
/routes\/canvas-capability-task-recovery\.test\.ts/,
|
||||
/routes\/canvas-toolbar-capability\.test\.ts/,
|
||||
|
||||
@ -244,21 +244,12 @@ test("background storage starts after delivery and coalesces repeated wakeups",
|
||||
const callbacks: Array<() => Promise<void>> = [];
|
||||
const preferences: Array<string | undefined> = [];
|
||||
const module = serverLoader({
|
||||
"next/server": {
|
||||
after: (fn: () => Promise<void>) => {
|
||||
callbacks.push(fn);
|
||||
},
|
||||
},
|
||||
"next/server": { after: (fn: () => Promise<void>) => callbacks.push(fn) },
|
||||
"./image-archive-db": { recoverStaleImageArchiveTasks: async () => {}, hasPendingImageArchives: async () => false, hasPendingImageDownloads: async () => false },
|
||||
"./image-download-preparation": { scheduleImageDownloadPreparation() {} },
|
||||
"@/lib/jobs/image-archive-job": {
|
||||
runImageArchiveJob: async ({ preferredTaskId }: { preferredTaskId?: string }) => {
|
||||
preferences.push(preferredTaskId);
|
||||
return {
|
||||
store: { total: preferences.length === 1 ? 1 : 0 },
|
||||
replace: { total: 0 },
|
||||
stats: { pendingStore: 0, pendingReplace: 0 },
|
||||
downloadTasks: 0,
|
||||
};
|
||||
},
|
||||
storeNextImageArchive: async (id?: string) => { preferences.push(id); return false; },
|
||||
replacePendingImageArchives: async () => 0,
|
||||
},
|
||||
})("lib/storage/image-archive-worker.ts");
|
||||
module.scheduleImageArchiveWorker();
|
||||
@ -266,7 +257,8 @@ test("background storage starts after delivery and coalesces repeated wakeups",
|
||||
expect(callbacks).toHaveLength(1);
|
||||
expect(preferences).toEqual([]);
|
||||
await callbacks[0]();
|
||||
expect(preferences).toEqual(["requested-image", undefined]);
|
||||
expect(preferences[0]).toBe("requested-image");
|
||||
expect(preferences).toHaveLength(4);
|
||||
});
|
||||
|
||||
test("bucket failure retains the provider URL and retry task", async () => {
|
||||
@ -662,21 +654,15 @@ test("replacement retries reconcile late writes once, without reclaiming complet
|
||||
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: () => {} },
|
||||
"./image-archive-db": { recoverStaleImageArchiveTasks: async () => {}, hasPendingImageArchives: async () => false, hasPendingImageDownloads: async () => false },
|
||||
"./image-download-preparation": { scheduleImageDownloadPreparation() {} },
|
||||
"@/lib/jobs/image-archive-job": {
|
||||
runImageArchiveJob: async () => {
|
||||
runs += 1;
|
||||
return {
|
||||
store: { total: 0 },
|
||||
replace: { total: 0 },
|
||||
stats: { pendingStore: 0, pendingReplace: 0 },
|
||||
downloadTasks: 0,
|
||||
};
|
||||
},
|
||||
storeNextImageArchive: async () => { runs++; return false; },
|
||||
replacePendingImageArchives: async () => 0,
|
||||
},
|
||||
})("lib/storage/image-archive-worker.ts");
|
||||
module.scheduleImageArchiveWorker();
|
||||
expect(runs).toBe(0);
|
||||
await expect.poll(() => runs, { timeout: 5_000 }).toBe(1);
|
||||
await expect.poll(() => runs, { timeout: 5_000 }).toBe(4);
|
||||
});
|
||||
|
||||
67
tests/routes/image-archive-lanes.test.ts
Normal file
67
tests/routes/image-archive-lanes.test.ts
Normal file
@ -0,0 +1,67 @@
|
||||
import { test, expect } from "@playwright/test";
|
||||
import { serverLoader } from "../helpers/canvas-download-server-loader";
|
||||
|
||||
test("a slow original and slow reference updates do not block later files or new downloads", async () => {
|
||||
let finishSlow!: () => void;
|
||||
let finishReplacement!: () => void;
|
||||
const slow = new Promise<void>(resolve => { finishSlow = resolve; });
|
||||
const replacement = new Promise<void>(resolve => { finishReplacement = resolve; });
|
||||
const pending = Array.from({ length: 25 }, (_, i) => `image-${i}`);
|
||||
const completed: string[] = [];
|
||||
const callbacks: Array<() => Promise<void>> = [];
|
||||
let active = 0;
|
||||
let peak = 0;
|
||||
let replacing = false;
|
||||
const module = serverLoader({
|
||||
"next/server": { after: (callback: () => Promise<void>) => callbacks.push(callback) },
|
||||
"./image-archive-db": { recoverStaleImageArchiveTasks: async () => {}, hasPendingImageArchives: async () => false, hasPendingImageDownloads: async () => false },
|
||||
"./image-download-preparation": { scheduleImageDownloadPreparation() {} },
|
||||
"@/lib/jobs/image-archive-job": {
|
||||
storeNextImageArchive: async (preferred?: string) => {
|
||||
const index = preferred ? pending.indexOf(preferred) : -1;
|
||||
const id = pending.splice(index < 0 ? 0 : index, 1)[0];
|
||||
if (!id) return false;
|
||||
active++; peak = Math.max(peak, active);
|
||||
if (id === "image-0") await slow;
|
||||
else await new Promise(resolve => setTimeout(resolve, 1));
|
||||
completed.push(id); active--;
|
||||
return true;
|
||||
},
|
||||
replacePendingImageArchives: async () => { replacing = true; await replacement; return 0; },
|
||||
},
|
||||
})("lib/storage/image-archive-worker.ts");
|
||||
module.scheduleImageArchiveWorker();
|
||||
await callbacks.shift()!();
|
||||
await expect.poll(() => completed.length).toBe(24);
|
||||
expect(completed).not.toContain("image-0");
|
||||
expect(replacing).toBe(true);
|
||||
pending.push("new-download");
|
||||
module.scheduleImageArchiveWorker(false, "new-download");
|
||||
await callbacks.shift()!();
|
||||
await expect.poll(() => completed.includes("new-download")).toBe(true);
|
||||
expect(peak).toBeLessThanOrEqual(4);
|
||||
finishSlow(); finishReplacement();
|
||||
await expect.poll(() => completed.length).toBe(26);
|
||||
expect(new Set(completed).size).toBe(26);
|
||||
});
|
||||
|
||||
test("conversion requests wait for the original archive and never invoke the relay preparation queue", async () => {
|
||||
const pending = { id: "queued", sourceUrl: "https://model.example/image", status: "pending" };
|
||||
const route = serverLoader({
|
||||
"next/server": { NextResponse: Response },
|
||||
"@/app/(auth)/auth": { auth: async () => ({ user: { id: "test" } }) },
|
||||
"@/lib/storage/image-download-preparation": { scheduleImageDownloadPreparation() {} },
|
||||
"@/lib/storage/canvas-cdn-download": { resolveCanvasCdnDownload: async () => ({ task: pending, direct: null, canPrepareDirect: false }) },
|
||||
"@/lib/storage/original-image-download": {
|
||||
prepareOriginalImageArchive: async () => pending,
|
||||
originalImagePreparing: () => ({ delivery: "preparing", phase: "preparing", requestId: pending.id }),
|
||||
},
|
||||
"@/lib/canvas/download-preparation": { getDownloadPreparationQueue() { throw new Error("Must not relay image bytes"); } },
|
||||
})("app/api/canvas/download-image/route.ts");
|
||||
const response = await route.POST(new Request("https://app.example/api/canvas/download-image", {
|
||||
method: "POST", headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ url: pending.sourceUrl, preferDirect: true, preparation: "async" }),
|
||||
}));
|
||||
expect(response.status).toBe(202);
|
||||
expect(await response.json()).toMatchObject({ delivery: "preparing", phase: "preparing" });
|
||||
});
|
||||
@ -24,7 +24,7 @@ test("original downloads use the archived JPEG rather than the prepared PNG and
|
||||
expect(new URL(result.downloadUrl).pathname).toBe("/original.jpg");
|
||||
});
|
||||
|
||||
test("signed original URLs retain the byte-preserving proxy without changing their signature", async () => {
|
||||
test("signed original URLs cannot silently fall back to the application server", async () => {
|
||||
const module = serverLoader({
|
||||
"@/lib/db/queries": { getDefaultStorageConfig: async () => config },
|
||||
"./image-archive-db": { findImageArchiveDownloadTask: async () => null },
|
||||
@ -45,16 +45,17 @@ test("existing CDN objects bypass the archive scan and retain the MIME-derived f
|
||||
expect(new URL(result.downloadUrl).pathname).toBe("/source.png");
|
||||
});
|
||||
|
||||
test("temporary third-party URLs are never checked with HEAD when no archive is ready", async () => {
|
||||
test("temporary third-party originals enqueue storage without reading their body", async () => {
|
||||
const module = serverLoader({
|
||||
"@/lib/db/queries": { getDefaultStorageConfig: async () => config },
|
||||
"./image-archive-db": { findImageArchiveDownloadTask: async () => null },
|
||||
"./image-archive-db": { findImageArchiveDownloadTask: async () => null, enqueueImageArchiveTask: async () => ({ id: "new", status: "pending" }) },
|
||||
"./image-archive-worker": { scheduleImageArchiveWorker() {} },
|
||||
fetch() { throw new Error("The upstream file service does not support HEAD"); },
|
||||
})("lib/storage/original-image-download.ts");
|
||||
expect(await module.resolveOriginalCdnDownload("https://upstream.example/image.png", "image", new AbortController().signal)).toBeNull();
|
||||
expect(await module.resolveOriginalCdnDownload("https://upstream.example/image.png", "image", new AbortController().signal)).toMatchObject({ delivery: "preparing" });
|
||||
});
|
||||
|
||||
test("downloading a pending original prioritizes its existing archive task and keeps the proxy available", async () => {
|
||||
test("downloading a pending original prioritizes storage and reports preparation", async () => {
|
||||
const preferred: string[] = [];
|
||||
const module = serverLoader({
|
||||
"@/lib/db/queries": { getDefaultStorageConfig: async () => config },
|
||||
@ -62,7 +63,7 @@ test("downloading a pending original prioritizes its existing archive task and k
|
||||
"./image-archive-worker": { scheduleImageArchiveWorker: (_delivery: boolean, id: string) => preferred.push(id) },
|
||||
fetch() { throw new Error("The original must remain available through the proxy while it is queued"); },
|
||||
})("lib/storage/original-image-download.ts");
|
||||
expect(await module.resolveOriginalCdnDownload("https://upstream.example/queued.png", "image", new AbortController().signal)).toBeNull();
|
||||
expect(await module.resolveOriginalCdnDownload("https://upstream.example/queued.png", "image", new AbortController().signal)).toMatchObject({ delivery: "preparing" });
|
||||
expect(preferred).toEqual(["pending-image"]);
|
||||
});
|
||||
|
||||
@ -151,14 +152,13 @@ test("chat original downloads use a native CDN download without fetching the ima
|
||||
expect(progress.map(event => event.phase)).toEqual(["preparing", "downloading"]);
|
||||
});
|
||||
|
||||
test("original fallback preserves JPEG bytes and the MIME-derived filename", async () => {
|
||||
const dom = browserMocks(); const bytes = Buffer.from([255, 216, 255, 224, 0, 16, 74, 70, 73, 70, 255, 217]); let calls = 0;
|
||||
const client = serverLoader({ ...dom, fetch: async () => ++calls === 1 ? Response.json({ delivery: "proxy" }) : new Response(bytes, { headers: { "content-type": "image/jpeg" } }) })("lib/client/canvas-image-download.ts");
|
||||
expect(await client.downloadOriginalImage({ url: "https://external.example/image.png", name: "original.png" })).toEqual({ filename: "original.jpg", bytes: bytes.length });
|
||||
expect(calls).toBe(2);
|
||||
const actual = await fetch(dom.links[0].href).then(r=>r.arrayBuffer());
|
||||
expect(Buffer.from(actual).equals(bytes)).toBe(true);
|
||||
URL.revokeObjectURL(dom.links[0].href);
|
||||
test("a legacy proxy response does not start a slow original download", async () => {
|
||||
const dom = browserMocks(); let calls = 0;
|
||||
const client = serverLoader({ ...dom, fetch: async () => { calls++; return Response.json({ delivery: "proxy" }); } })("lib/client/canvas-image-download.ts");
|
||||
await expect(client.downloadOriginalImage({ url: "https://external.example/image.png" })).rejects.toThrow("原图尚未准备完成");
|
||||
expect(calls).toBe(1);
|
||||
expect(dom.links).toHaveLength(0);
|
||||
expect(dom.frames).toHaveLength(0);
|
||||
});
|
||||
|
||||
test("one click prepares a compatible PNG when the CDN original is WebP", async () => {
|
||||
@ -174,7 +174,7 @@ test("one click prepares a compatible PNG when the CDN original is WebP", async
|
||||
expect(init?.method).toBe("POST");
|
||||
expect(JSON.parse(String(init?.body))).toMatchObject({ url, name: "商品", preparation: "async" });
|
||||
if (calls.length === 2) {
|
||||
return Response.json({ delivery: "preparing", retryAfter: 1, message: "正在准备原图" }, { status: 202 });
|
||||
return Response.json({ delivery: "preparing", phase: "converting", retryAfter: 1, message: "正在转换为兼容 PNG，完成后自动下载" }, { status: 202 });
|
||||
}
|
||||
return Response.json({ delivery: "cdn", downloadUrl: "https://cdn.example/prepared.png", filename: "商品.png", bytes: 900 });
|
||||
} })("lib/client/canvas-image-download.ts");
|
||||
@ -190,31 +190,19 @@ test("one click prepares a compatible PNG when the CDN original is WebP", async
|
||||
expect(dom.links).toHaveLength(0);
|
||||
});
|
||||
|
||||
test("one click uses the PNG preparation route without buffering an incompatible proxy response", async () => {
|
||||
const dom = browserMocks();
|
||||
const calls: string[] = [];
|
||||
const progress: CanvasImageDownloadProgress[] = [];
|
||||
let cancelled = false;
|
||||
const bytes = Uint8Array.of(137, 80, 78, 71, 13, 10, 26, 10);
|
||||
const client = serverLoader({ ...dom, fetch: async (path: string) => {
|
||||
calls.push(path);
|
||||
if (calls.length === 1) return Response.json({ delivery: "proxy" });
|
||||
if (calls.length === 2) {
|
||||
return new Response(new ReadableStream({ cancel() { cancelled = true; } }),
|
||||
{ headers: { "content-type": "image/avif" } });
|
||||
}
|
||||
return new Response(bytes, { headers: { "content-type": "image/png", "content-length": String(bytes.length) } });
|
||||
test("queued originals show preparation then download directly without a proxy body", async () => {
|
||||
const dom = browserMocks(); const calls: string[] = []; const progress: CanvasImageDownloadProgress[] = [];
|
||||
const client = serverLoader({ ...dom, fetch: async (url: string) => {
|
||||
calls.push(url);
|
||||
if (calls.length === 1) return Response.json({ delivery: "preparing", retryAfter: 1, message: "正在保存原图，完成后自动下载" }, { status: 202 });
|
||||
return Response.json({ delivery: "cdn", downloadUrl: "https://cdn.example/image.png", filename: "image.png", bytes: 900 });
|
||||
} })("lib/client/canvas-image-download.ts");
|
||||
expect(await client.downloadOriginalImage({
|
||||
url: "https://external.example/image", name: "商品", onProgress: (event: CanvasImageDownloadProgress) => progress.push(event),
|
||||
})).toEqual({ filename: "商品.png", bytes: bytes.length });
|
||||
expect(cancelled).toBe(true);
|
||||
expect(calls[2]).toBe("/api/canvas/download-image");
|
||||
expect(progress).toContainEqual({
|
||||
phase: "downloading", message: "正在下载 PNG...", loadedBytes: bytes.length, totalBytes: bytes.length, percent: 99,
|
||||
});
|
||||
expect(dom.links).toHaveLength(1);
|
||||
URL.revokeObjectURL(dom.links[0].href);
|
||||
await client.downloadOriginalImage({ url: "https://external.example/image", onProgress: (event: CanvasImageDownloadProgress) => progress.push(event) });
|
||||
expect(calls).toHaveLength(2);
|
||||
expect(calls.every(url => url.includes("download=resolve"))).toBe(true);
|
||||
expect(progress.map(p => p.phase)).toEqual(["preparing", "preparing", "downloading"]);
|
||||
expect(dom.frames[0].src).toBe("https://cdn.example/image.png");
|
||||
expect(dom.links).toHaveLength(0);
|
||||
});
|
||||
|
||||
test("original downloads reject processed references before network access", async () => {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user