feat(import): run catalog synchronizations as durable jobs

src/lib/furni/import-job.ts

@@ -1,3 +1,4 @@
+import type { SourceFurni } from "@/lib/services/clone-import";
 export interface ImportJobItem {
 	id: number;
 	classname: string;
@@ -7,9 +8,13 @@ export interface ImportJobItem {
 	revision: number;
 	category: string;
 	attachmentId?: string;
+	cloneSourceId?: string;
+	cloneEntry?: SourceFurni;
 }
 export interface ImportJob {
 	mode?: "repair";
+	syncKind?: "official" | "clone";
+	operationId?: string;
 	retryOf?: string;
 	id: string;
 	userId: number;

src/lib/services/furni-job-worker.test.ts

@@ -16,7 +16,7 @@ const mocks = vi.hoisted(() => ({
 }));
 vi.mock("@/lib/redis", () => ({ redis: { set: mocks.set, eval: mocks.eval } }));
 vi.mock("@/lib/server-log", () => ({ logServerError: vi.fn() }));
-vi.mock("./audit", () => ({ logAudit: vi.fn() }));
+vi.mock("./audit", () => ({ logAudit: vi.fn().mockResolvedValue(undefined) }));
 vi.mock("./catalog-git-queue", () => ({ withCatalogExport: mocks.export }));
 vi.mock("./clone-sources", () => ({ getSource: async () => null }));
 vi.mock("./furni-job-store", () => ({
@@ -42,6 +42,8 @@ vi.mock("./rcon", () => ({
 
 vi.mock("./catalog-git-export", () => ({ runCatalogExport: vi.fn() }));
 
+vi.mock("./furni-sync-item", () => ({ runSyncJobItem: mocks.import }));
+
 import { drainFurnitureImports } from "./furni-job-worker";
 
 let job: ImportJob;
@@ -223,3 +225,27 @@ it("does not save a translated outcome after ownership is lost during translatio
 	expect(snapshots.at(-1)?.items[0].state).toBe("running");
 	expect(mocks.updateCatalog).not.toHaveBeenCalled();
 });
+
+it("restores the original operation context while running a durable job", async () => {
+	const { getOperationContext } = await import(
+		"@/lib/foundation/request-context"
+	);
+	job.operationId = "original-operation";
+	mocks.import.mockImplementation(async () => {
+		expect(getOperationContext()).toEqual({
+			operationId: "original-operation",
+			userId: 5,
+		});
+		return { ok: true, itemId: 900, warnings: [] };
+	});
+	await drainFurnitureImports();
+	expect(getOperationContext()).toEqual({});
+});
+
+it("keeps successful imports complete if their audit write fails", async () => {
+	const { logAudit } = await import("./audit");
+	vi.mocked(logAudit).mockRejectedValueOnce(new Error("Audit unavailable"));
+	await drainFurnitureImports();
+	expect(job.items[0].state).toBe("done");
+	expect(job.items[0].warnings).toContain("Audit record could not be saved");
+});

src/lib/services/furni-job-worker.ts

@@ -1,4 +1,6 @@
 import { randomUUID } from "node:crypto";
+import { createStore, runWithStore } from "@/lib/foundation/request-context";
+import type { IpAddress, RequestId, UserId } from "@/lib/foundation/types";
 import type { ImportJob } from "@/lib/furni/import-job";
 import { redis } from "@/lib/redis";
 import { logServerError } from "@/lib/server-log";
@@ -10,6 +12,7 @@ import { readFurnitureAttachment } from "./furni-attachment";
 import { patchLocalizedFurniDataEntries } from "./furni-data-i18n";
 import { ensureDirectories, importSingleFurni } from "./furni-import";
 import { ImportJobStore } from "./furni-job-store";
+import { runSyncJobItem } from "./furni-sync-item";
 import { repairFurniture } from "./furniture-repair";
 import { rcon } from "./rcon";
 
@@ -100,87 +103,108 @@ async function drain() {
 			}
 			job.state = "running";
 			await saveOwned(job);
-			await withCatalogExport(async () => {
-				await ensureDirectories();
-				const source = job.sourceId ? await getSource(job.sourceId) : null;
-				for (const item of job.items) {
-					await requireLease();
-					if (item.state !== "pending") continue;
-					if (await store.isCancellationRequested(job.id)) {
-						for (const remaining of job.items)
-							if (remaining.state === "pending") remaining.state = "cancelled";
-						break;
-					}
-					item.state = "running";
-					await saveOwned(job);
-					try {
-						if (job.sourceId && !source)
-							throw Error("Import source no longer exists");
-						const providedNitro = item.attachmentId
-							? await readFurnitureAttachment(
-									item.attachmentId,
-									item.classname,
-									job.userId,
-								)
-							: undefined;
+			const operation = createStore("worker" as IpAddress);
+			operation.requestId = (job.operationId ?? job.id) as RequestId;
+			operation.userId = job.userId as UserId;
+			await runWithStore(operation, () =>
+				withCatalogExport(async () => {
+					await ensureDirectories();
+					const source = job.sourceId ? await getSource(job.sourceId) : null;
+					for (const item of job.items) {
 						await requireLease();
-						const result = await (job.mode === "repair"
-							? repairFurniture
-							: importSingleFurni)({
-							...item,
-							repairExisting: true,
-							providedNitro,
-							sourceSwfBaseUrl: source?.sourceSwfBaseUrl,
-							nitroBaseUrl: source?.nitroBaseUrl,
-							iconBaseUrl: source?.iconBaseUrl,
-						});
-						await requireLease();
-						item.warnings = result.warnings;
-						item.itemId = result.itemId;
-						if (!result.ok) throw Error(result.error || "Import failed");
-						item.state = "done";
-						logAudit({
-							userId: job.userId,
-							action: "furni_import",
-							target: "ItemsBase",
-							targetId: result.itemId ?? 0,
-							after: {
-								classname: item.classname,
+						if (item.state !== "pending") continue;
+						if (await store.isCancellationRequested(job.id)) {
+							for (const remaining of job.items)
+								if (remaining.state === "pending")
+									remaining.state = "cancelled";
+							break;
+						}
+						item.state = "running";
+						await saveOwned(job);
+						try {
+							if (job.sourceId && !source)
+								throw Error("Import source no longer exists");
+							const providedNitro = item.attachmentId
+								? await readFurnitureAttachment(
+										item.attachmentId,
+										item.classname,
+										job.userId,
+									)
+								: undefined;
+							await requireLease();
+							const result = job.syncKind
+								? await runSyncJobItem(item)
+								: await (job.mode === "repair"
+										? repairFurniture
+										: importSingleFurni)({
+										...item,
+										repairExisting: true,
+										providedNitro,
+										sourceSwfBaseUrl: source?.sourceSwfBaseUrl,
+										nitroBaseUrl: source?.nitroBaseUrl,
+										iconBaseUrl: source?.iconBaseUrl,
+									});
+							await requireLease();
+							item.warnings = result.warnings;
+							item.itemId = result.itemId;
+							if (!result.ok) throw Error(result.error || "Import failed");
+							item.state = "done";
+							await logAudit({
+								userId: job.userId,
+								action: "furni_import",
+								target: "ItemsBase",
+								targetId: result.itemId ?? 0,
+								after: {
+									classname: item.classname,
+									jobId: job.id,
+									operationId: job.operationId,
+									syncKind: job.syncKind,
+									repairExisting: true,
+								},
+							}).catch((auditError) => {
+								item.warnings ??= [];
+								item.warnings.push("Audit record could not be saved");
+								logServerError("furni.audit_failed", auditError, {
+									jobId: job.id,
+									classname: item.classname,
+								});
+							});
+							if (job.translate && job.mode !== "repair")
+								try {
+									await patchLocalizedFurniDataEntries([item], true, job.langs);
+								} catch {
+									item.warnings.push(
+										"Translation failed; furniture imported successfully",
+									);
+								}
+						} catch (error) {
+							if (error instanceof ImportLeaseLostError) throw error;
+							await requireLease();
+							logServerError("furni.item_failed", error, {
 								jobId: job.id,
-								repairExisting: true,
-							},
-						});
-						if (job.translate && job.mode !== "repair")
-							try {
-								await patchLocalizedFurniDataEntries([item], true, job.langs);
-							} catch {
-								item.warnings.push(
-									"Translation failed; furniture imported successfully",
-								);
-							}
+								classname: item.classname,
+							});
+							item.state = "failed";
+							item.error =
+								error instanceof Error ? error.message : "Import failed";
+						}
+						await saveOwned(job);
+					}
+					await requireLease();
+					try {
+						await rcon.updateCatalog();
+						await requireLease();
+						await rcon.updateItems();
 					} catch (error) {
 						if (error instanceof ImportLeaseLostError) throw error;
-						await requireLease();
-						item.state = "failed";
-						item.error =
-							error instanceof Error ? error.message : "Import failed";
+						for (const item of job.items)
+							if (item.state === "done") {
+								item.warnings ??= [];
+								item.warnings.push("Game cache refresh failed");
+							}
 					}
-					await saveOwned(job);
-				}
-				await requireLease();
-				try {
-					await rcon.updateCatalog();
-					await requireLease();
-					await rcon.updateItems();
-				} catch (error) {
-					if (error instanceof ImportLeaseLostError) throw error;
-					for (const item of job.items)
-						if (item.state === "done") {
-							item.warnings ??= [];
-							item.warnings.push("Game cache refresh failed");
-						}
-				}
-			});
+				}),
+			);
 			await requireLease();
 			job.state = job.items.some((item) => item.state === "cancelled")
 				? "cancelled"

src/lib/services/furni-sync-item.test.ts

@@ -0,0 +1,75 @@
+import { beforeEach, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+	existing: vi.fn(),
+	clone: vi.fn(),
+	source: vi.fn(),
+	import: vi.fn(),
+}));
+vi.mock("@/lib/db", async () => ({
+	...(await import("@/db/schema")),
+	db: {
+		select: () => ({
+			from: () => ({ where: () => ({ limit: mocks.existing }) }),
+		}),
+	},
+}));
+vi.mock("./clone-import", () => ({ cloneSingleFurni: mocks.clone }));
+vi.mock("./clone-sources", () => ({ getSource: mocks.source }));
+vi.mock("./furni-import", () => ({ importSingleFurni: mocks.import }));
+
+import { runSyncJobItem } from "./furni-sync-item";
+
+const item = {
+	id: 0,
+	classname: "chair",
+	name: "Chair",
+	description: "",
+	type: "flooritem",
+	revision: 1,
+	category: "other",
+};
+beforeEach(() => {
+	vi.clearAllMocks();
+	mocks.existing.mockResolvedValue([]);
+	mocks.import.mockResolvedValue({ ok: true, warnings: [] });
+});
+it("skips an item another queued request has already imported", async () => {
+	mocks.existing.mockResolvedValue([{ id: 3 }]);
+	expect(await runSyncJobItem(item)).toMatchObject({ ok: true, itemId: 3 });
+	expect(mocks.import).not.toHaveBeenCalled();
+	expect(mocks.clone).not.toHaveBeenCalled();
+});
+it("never repairs existing furniture during official sync", async () => {
+	await runSyncJobItem(item);
+	expect(mocks.import).toHaveBeenCalledWith({ ...item, repairExisting: false });
+});
+it("fails a deleted clone source without substituting official assets", async () => {
+	mocks.source.mockResolvedValue(null);
+	await expect(
+		runSyncJobItem({ ...item, cloneSourceId: "gone" }),
+	).rejects.toThrow("no longer available");
+	expect(mocks.import).not.toHaveBeenCalled();
+});
+
+it("returns the created clone item ID for audit correlation", async () => {
+	mocks.existing.mockResolvedValueOnce([]).mockResolvedValueOnce([{ id: 71 }]);
+	mocks.source.mockResolvedValue({ id: "source" });
+	mocks.clone.mockResolvedValue({ ok: true, warnings: [] });
+	const cloneEntry = {
+		id: 1,
+		classname: "chair",
+		name: "Chair",
+		description: "",
+		xdim: 1,
+		ydim: 1,
+		canstandon: false,
+		cansiton: true,
+		canlayon: false,
+		customparams: "",
+		itemType: "s" as const,
+	};
+	expect(
+		await runSyncJobItem({ ...item, cloneSourceId: "source", cloneEntry }),
+	).toMatchObject({ ok: true, itemId: 71 });
+});

src/lib/services/furni-sync-item.ts

@@ -0,0 +1,36 @@
+import { eq } from "drizzle-orm";
+import { db, ItemsBase } from "@/lib/db";
+import type { ImportJobItem } from "@/lib/furni/import-job";
+import { cloneSingleFurni } from "./clone-import";
+import { getSource } from "./clone-sources";
+import { importSingleFurni } from "./furni-import";
+
+/** Called only inside the shared import lease and catalog export lock. */
+export async function runSyncJobItem(item: ImportJobItem) {
+	const [existing] = await db
+		.select({ id: ItemsBase.id })
+		.from(ItemsBase)
+		.where(eq(ItemsBase.itemName, item.classname))
+		.limit(1);
+	if (existing)
+		return {
+			ok: true,
+			itemId: existing.id,
+			warnings: ["Already present; synchronization skipped this item."],
+		};
+	if (item.cloneSourceId) {
+		const source = await getSource(item.cloneSourceId);
+		if (!source || !item.cloneEntry)
+			throw Error("Synchronization source no longer available");
+		const result = await cloneSingleFurni({ source, entry: item.cloneEntry });
+		const [created] = result.ok
+			? await db
+					.select({ id: ItemsBase.id })
+					.from(ItemsBase)
+					.where(eq(ItemsBase.itemName, item.classname))
+					.limit(1)
+			: [];
+		return { ...result, itemId: created?.id };
+	}
+	return importSingleFurni({ ...item, repairExisting: false });
+}

src/lib/services/furni-sync-queue.test.ts

@@ -0,0 +1,113 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+	read: vi.fn(),
+	create: vi.fn(),
+	official: vi.fn(),
+	sources: vi.fn(),
+	fetch: vi.fn(),
+	ping: vi.fn(),
+	drain: vi.fn(),
+	after: vi.fn(),
+	existing: vi.fn(),
+}));
+vi.mock("next/server", () => ({ after: mocks.after, NextResponse: Response }));
+vi.mock("@/lib/db", () => ({
+	ItemsBase: { itemName: "item_name" },
+	db: { select: () => ({ from: mocks.existing }) },
+}));
+vi.mock("@/lib/redis", () => ({ redis: { ping: mocks.ping } }));
+vi.mock("./furni-job-store", () => ({
+	validJobId: (id: unknown) => typeof id === "string" && id.length === 36,
+	ImportJobStore: class {
+		read = mocks.read;
+		create = mocks.create;
+	},
+}));
+vi.mock("./furni-job-worker", () => ({ drainFurnitureImports: mocks.drain }));
+vi.mock("./clone-sources", () => ({ listSources: mocks.sources }));
+vi.mock("./clone-import", () => ({ fetchSourceFurnidata: mocks.fetch }));
+vi.mock("./habbo-furnidata-cache", () => ({
+	getOfficialHabboFurnidata: mocks.official,
+}));
+
+import { enqueueFurnitureSync } from "./furni-sync-queue";
+
+const id = "11111111-1111-4111-8111-111111111111";
+const request = (body: unknown = { id }) =>
+	new Request("https://example.test/api/sync", {
+		method: "POST",
+		body: JSON.stringify(body),
+	});
+beforeEach(() => {
+	vi.clearAllMocks();
+	mocks.read.mockRejectedValue({ code: "ENOENT" });
+	mocks.create.mockImplementation(async (job) => job);
+	mocks.ping.mockResolvedValue("PONG");
+	mocks.existing.mockResolvedValue([]);
+	mocks.official.mockResolvedValue(new Map());
+	mocks.sources.mockResolvedValue([]);
+});
+describe("durable synchronization enqueue", () => {
+	it("persists an empty completed job when already up to date", async () => {
+		const res = await enqueueFurnitureSync(request(), 7, "official");
+		expect(res.status).toBe(200);
+		expect(mocks.create).toHaveBeenCalledWith(
+			expect.objectContaining({ state: "completed", items: [], userId: 7 }),
+		);
+	});
+	it("reuses an existing request before source discovery", async () => {
+		const job = { id, userId: 7, syncKind: "official" };
+		mocks.read.mockResolvedValue(job);
+		expect((await enqueueFurnitureSync(request(), 7, "official")).status).toBe(
+			200,
+		);
+		expect(mocks.create).not.toHaveBeenCalled();
+		expect(mocks.official).not.toHaveBeenCalled();
+	});
+	it("rejects other owners and changed operation kinds", async () => {
+		mocks.read.mockResolvedValue({ id, userId: 8, syncKind: "official" });
+		expect((await enqueueFurnitureSync(request(), 7, "official")).status).toBe(
+			409,
+		);
+		mocks.read.mockResolvedValue({ id, userId: 7, syncKind: "clone" });
+		expect((await enqueueFurnitureSync(request(), 7, "official")).status).toBe(
+			409,
+		);
+	});
+	it("does not enqueue while redis is unavailable", async () => {
+		mocks.ping.mockRejectedValue(Error("offline"));
+		expect((await enqueueFurnitureSync(request(), 7, "official")).status).toBe(
+			503,
+		);
+		expect(mocks.create).not.toHaveBeenCalled();
+	});
+	it("deduplicates classnames across source snapshots", async () => {
+		mocks.sources.mockResolvedValue([
+			{ id: "a", furnidataUrl: "a" },
+			{ id: "b", furnidataUrl: "b" },
+		]);
+		mocks.fetch.mockResolvedValue([
+			{
+				id: 1,
+				classname: "chair",
+				name: "Chair",
+				description: "",
+				itemType: "s",
+			},
+		]);
+		await enqueueFurnitureSync(request(), 7, "clone");
+		const job = mocks.create.mock.calls[0][0];
+		expect(job.items).toHaveLength(1);
+		expect(job.items[0].cloneSourceId).toBe("a");
+		expect(job.items[0].cloneEntry.classname).toBe("chair");
+	});
+	it("never queues partial discovery after a source failure", async () => {
+		mocks.sources.mockResolvedValue([{ id: "a", furnidataUrl: "a" }]);
+		mocks.fetch.mockRejectedValue(Error("unavailable"));
+		await expect(enqueueFurnitureSync(request(), 7, "clone")).rejects.toThrow(
+			"unavailable",
+		);
+		expect(mocks.create).not.toHaveBeenCalled();
+	});
+});

src/lib/services/furni-sync-queue.ts

@@ -0,0 +1,114 @@
+import { after } from "next/server";
+import { apiError, apiOk } from "@/lib/api";
+import { db, ItemsBase } from "@/lib/db";
+import { getRequestId } from "@/lib/foundation/request-context";
+import type { ImportJob, ImportJobItem } from "@/lib/furni/import-job";
+import { validateClassnames } from "@/lib/furni/studio-inspection";
+import { redis } from "@/lib/redis";
+import { fetchSourceFurnidata } from "./clone-import";
+import { listSources } from "./clone-sources";
+import { ImportJobStore, validJobId } from "./furni-job-store";
+import { drainFurnitureImports } from "./furni-job-worker";
+import { getOfficialHabboFurnidata } from "./habbo-furnidata-cache";
+
+export async function enqueueFurnitureSync(
+	request: Request,
+	userId: number,
+	kind: "official" | "clone",
+) {
+	const body = await request.json().catch(() => null);
+	if (
+		!validJobId(body?.id) ||
+		(kind === "official" && body.sourceId !== undefined) ||
+		(body.sourceId !== undefined &&
+			(typeof body.sourceId !== "string" || body.sourceId.length > 100))
+	)
+		return apiError("Invalid synchronization request", 400);
+	const store = new ImportJobStore();
+	const existing = await store.read(body.id).catch((error) => {
+		if (error.code === "ENOENT") return null;
+		throw error;
+	});
+	if (existing) {
+		if (
+			existing.userId !== userId ||
+			existing.syncKind !== kind ||
+			existing.sourceId !== body.sourceId
+		)
+			return apiError("Request ID already used", 409);
+		after(drainFurnitureImports);
+		return apiOk({ job: existing });
+	}
+	if (!redis || (await redis.ping().catch(() => null)) !== "PONG")
+		return apiError("Background import queue is temporarily unavailable", 503);
+	const have = new Set(
+		(await db.select({ name: ItemsBase.itemName }).from(ItemsBase)).map(
+			(row) => row.name,
+		),
+	);
+	const items = new Map<string, ImportJobItem>();
+	if (kind === "official") {
+		for (const entry of (await getOfficialHabboFurnidata()).values()) {
+			if (have.has(entry.classname)) continue;
+			items.set(entry.classname, {
+				id: 0,
+				classname: entry.classname,
+				name: entry.name,
+				description: entry.description ?? "",
+				type: entry.category === "wallitem" ? "wallitem" : "flooritem",
+				revision: entry.revision ?? 0,
+				category: entry.category ?? "unknown",
+			});
+		}
+	} else {
+		const sources = (await listSources()).filter(
+			(source) => !body.sourceId || source.id === body.sourceId,
+		);
+		if (!sources.length) return apiError("No sources configured", 400);
+		// Discovery is read-only. Persist one complete snapshot, never a partial request.
+		for (const source of sources)
+			for (const entry of await fetchSourceFurnidata(source.furnidataUrl)) {
+				if (have.has(entry.classname) || items.has(entry.classname)) continue;
+				items.set(entry.classname, {
+					id: entry.id,
+					classname: entry.classname,
+					name: entry.name,
+					description: entry.description,
+					type: entry.itemType === "i" ? "wallitem" : "flooritem",
+					revision: Number(entry.revision) || 0,
+					category: String(entry.category ?? "unknown"),
+					cloneSourceId: source.id,
+					cloneEntry: entry,
+				});
+			}
+	}
+	if (items.size > (kind === "official" ? 500 : 2000))
+		return apiError(
+			"Too many missing items. Import a smaller selection from Studio.",
+			400,
+		);
+	if ([...items.keys()].some((classname) => !validateClassnames([classname])))
+		return apiError("Source contains invalid furniture identifiers", 400);
+	const now = new Date().toISOString();
+	const job: ImportJob = {
+		id: body.id,
+		userId,
+		syncKind: kind,
+		operationId: getRequestId(),
+		sourceId: body.sourceId,
+		translate: false,
+		createdAt: now,
+		updatedAt: now,
+		state: items.size ? "queued" : "completed",
+		items: [...items.values()].map((item) => ({ ...item, state: "pending" })),
+	};
+	const saved = await store.create(job);
+	if (
+		saved.userId !== userId ||
+		saved.syncKind !== kind ||
+		saved.sourceId !== body.sourceId
+	)
+		return apiError("Request ID already used", 409);
+	after(drainFurnitureImports);
+	return apiOk({ job: saved });
+}