89 lines
3.7 KiB
TypeScript
89 lines
3.7 KiB
TypeScript
import "server-only";
|
|
import { after } from "next/server";
|
|
|
|
const workerKey = "__generated_image_archive_worker_v2__";
|
|
type Worker = {
|
|
active: number;
|
|
scheduled: boolean;
|
|
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] ??= { 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();
|
|
if (preferredTaskId) worker.priority.add(preferredTaskId);
|
|
if (worker.scheduled && !deliveryConfirmed) return;
|
|
worker.scheduled = true;
|
|
const run = async () => {
|
|
if (!worker.scheduled) return;
|
|
worker.scheduled = false;
|
|
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));
|
|
});
|
|
}
|
|
};
|
|
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); }
|
|
}
|