95 lines
4.6 KiB
TypeScript
95 lines
4.6 KiB
TypeScript
import "server-only";
|
|
import { randomUUID } from "node:crypto";
|
|
import { fetchImageBytes, cacheImageBytes } from "./image-byte-cache";
|
|
import { normalizeCanvasDownloadAsset } from "@/lib/canvas/photoshop-safe-download";
|
|
import { claimImageDownloadTask, saveImageDownloadAsset, getImageDownloadTasks, hasPendingImageDownloads } from "./image-archive-db";
|
|
import { uploadFile } from "./service";
|
|
import { getDownloadFailure } from "@/lib/canvas/download-error";
|
|
import { createLogger } from "@/lib/observability/logger";
|
|
import { trackRuntimeWork } from "@/lib/observability/active-work";
|
|
|
|
const logger = createLogger("canvas.download");
|
|
|
|
export async function prepareImageArchiveDownload(id: string) {
|
|
const task = await claimImageDownloadTask(id);
|
|
if (!task?.targetUrl) return;
|
|
let phase: "read" | "convert" | "store" = "read";
|
|
try {
|
|
const source = await fetchImageBytes(task.targetUrl);
|
|
phase = "convert";
|
|
const asset = await normalizeCanvasDownloadAsset({
|
|
item: { url: task.targetUrl }, index: 0, bytes: source.buffer,
|
|
contentType: source.contentType, includeSequencePrefix: false,
|
|
});
|
|
if (!asset.ok) throw new Error(asset.error);
|
|
// An already compatible PNG shares the original object; other formats get
|
|
// one prepared PNG, without changing the original dimensions or transparency.
|
|
phase = "store";
|
|
const url = source.buffer.equals(asset.bytes) && asset.contentType === "image/png"
|
|
? task.targetUrl
|
|
: (await uploadFile(`canvas-download-${randomUUID()}.${asset.filename.split(".").pop() || "png"}`, Uint8Array.from(asset.bytes).buffer)).url;
|
|
await saveImageDownloadAsset(id, { status: "ready", url, bytes: asset.bytes.length,
|
|
contentType: asset.contentType, preparedAt: new Date().toISOString() });
|
|
await cacheImageBytes(`source::${url}`, { buffer: asset.bytes, contentType: asset.contentType });
|
|
} catch (error) {
|
|
const attempts = Number((task.metadata as { downloadAsset?: { attempts?: number } } | null)?.downloadAsset?.attempts || 1);
|
|
const failure = phase === "store"
|
|
? getDownloadFailure("STORAGE_WRITE_FAILED")
|
|
: phase === "convert" && getDownloadFailure(error).code === "DOWNLOAD_FAILED"
|
|
? getDownloadFailure("图片转换失败") : getDownloadFailure(error);
|
|
logger.warn("archive_download_preparation_failed", error, { requestId: id, code: failure.code });
|
|
await saveImageDownloadAsset(id, { status: !failure.retryable || attempts >= 12 ? "abandoned" : "failed", attempts,
|
|
failure, retryAt: new Date(Date.now() + 60_000).toISOString() });
|
|
}
|
|
}
|
|
export async function preparePendingImageDownloads() {
|
|
const tasks = await getImageDownloadTasks();
|
|
let cursor = 0;
|
|
await Promise.all(Array.from({ length: Math.min(2, tasks.length) }, async () => {
|
|
while (cursor < tasks.length) await prepareImageArchiveDownload(tasks[cursor++].id);
|
|
}));
|
|
return tasks.length;
|
|
}
|
|
|
|
const state = globalThis as typeof globalThis & {
|
|
__imageDownloadWorker?: { running: boolean; scanPending: boolean; priority: Set<string>; retry?: ReturnType<typeof setTimeout> };
|
|
};
|
|
|
|
/** Download preparation runs independently of the archive URL replacement batch. */
|
|
export function scheduleImageDownloadPreparation(id?: string) {
|
|
const worker = state.__imageDownloadWorker ??= { running: false, scanPending: false, priority: new Set() };
|
|
if (id) worker.priority.add(id);
|
|
else worker.scanPending = true;
|
|
if (worker.running) return;
|
|
if (worker.retry) { clearTimeout(worker.retry); worker.retry = undefined; }
|
|
worker.running = true;
|
|
setImmediate(() => {
|
|
void trackRuntimeWork(async () => {
|
|
try {
|
|
for (;;) {
|
|
const priority = [...worker.priority].slice(0, 2);
|
|
for (const taskId of priority) worker.priority.delete(taskId);
|
|
if (priority.length) {
|
|
await Promise.all(priority.map(prepareImageArchiveDownload));
|
|
} else if (!worker.scanPending || await preparePendingImageDownloads() === 0) {
|
|
break;
|
|
}
|
|
}
|
|
} catch (error) {
|
|
logger.error("download_worker_failed", error);
|
|
} finally {
|
|
worker.running = false;
|
|
// Eligible retry times live in the database, including interrupted leases.
|
|
if (worker.priority.size) {
|
|
scheduleImageDownloadPreparation([...worker.priority][0]);
|
|
} else if (worker.scanPending && await hasPendingImageDownloads().catch(() => true)) {
|
|
worker.retry = setTimeout(() => scheduleImageDownloadPreparation(), 60_000);
|
|
worker.retry.unref?.();
|
|
} else {
|
|
worker.scanPending = false;
|
|
}
|
|
}
|
|
}, { durable: true });
|
|
});
|
|
}
|