feat(jobs): add cancellation and queued source transfers

apps/server/src/application/services/directory-sync-service.ts

@@ -50,6 +50,38 @@ function publishSyncStatus(
 	});
 }
 
+async function persistSyncStatus(
+	mediaSourceId: string,
+	status: MediaSourceSyncState,
+	message?: string,
+): Promise<void> {
+	const now = new Date();
+	try {
+		await sourceRepo.update(mediaSourceId, {
+			syncStatus: status,
+			...(status === "syncing"
+				? { lastSyncStartedAt: now, lastSyncError: null }
+				: status === "idle"
+					? { lastSyncCompletedAt: now, lastSyncError: null }
+					: { lastSyncError: message ?? "Directory sync failed" }),
+		});
+	} catch (error) {
+		logger.error(
+			{ err: error, mediaSourceId, status },
+			"Failed to persist media source sync status",
+		);
+	}
+}
+
+async function setSyncStatus(
+	mediaSourceId: string,
+	status: MediaSourceSyncState,
+	message?: string,
+): Promise<void> {
+	await persistSyncStatus(mediaSourceId, status, message);
+	publishSyncStatus(mediaSourceId, status, message);
+}
+
 export function getSourceSyncState(
 	mediaSourceId: string,
 ): MediaSourceSyncState {
@@ -173,15 +205,15 @@ export const DirectorySyncService = {
 				added: 0,
 				deleted: 0,
 			};
-			publishSyncStatus(mediaSourceId, "syncing");
+			await setSyncStatus(mediaSourceId, "syncing");
 			try {
 				const source = await sourceRepo.findById(mediaSourceId);
 				if (source?.type !== "local") {
 					logger.info(
 						{ mediaSourceId },
 						"Skipping sync for non-local or missing source",
 					);
-					publishSyncStatus(
+					await setSyncStatus(
 						mediaSourceId,
 						"idle",
 						source
@@ -200,7 +232,7 @@ export const DirectorySyncService = {
 						{ mediaSourceId, basePath },
 						"Base path does not exist or is not accessible during sync",
 					);
-					publishSyncStatus(
+					await setSyncStatus(
 						mediaSourceId,
 						"error",
 						"Source path is not accessible",
@@ -268,14 +300,14 @@ export const DirectorySyncService = {
 					{ mediaSourceId, syncResult: result },
 					"Directory sync completed successfully",
 				);
-				publishSyncStatus(mediaSourceId, "idle");
+				await setSyncStatus(mediaSourceId, "idle");
 				return result;
 			} catch (error) {
 				logger.error(
 					{ err: error, mediaSourceId },
 					"Error during directory sync",
 				);
-				publishSyncStatus(
+				await setSyncStatus(
 					mediaSourceId,
 					"error",
 					PublicDirectorySyncFailureMessage,

apps/server/src/application/services/job-dispatch-service.ts

@@ -1,6 +1,10 @@
 import type { DeferredActions } from "@solid-imager/application/ports/media-service";
 import type { Job } from "@solid-imager/core/domain/repositories/job-repository";
 import { services } from "~/application/registry";
+import {
+	processSourceExportJob,
+	processSourceRestoreJob,
+} from "~/application/services/source-transfer-job-service";
 import { RealtimeEventBus } from "~/infrastructure/events/realtime-event-bus";
 import {
 	processAutoTaggingJob,
@@ -65,6 +69,10 @@ export async function processJob(job: Job) {
 		await processBatchCcipDispatchJob(job);
 	} else if (job.type === "generate_thumbnail") {
 		await getThumbnailJobHandlers().processThumbnailGenerationJob(job);
+	} else if (job.type === "source_export") {
+		await processSourceExportJob(job);
+	} else if (job.type === "source_restore") {
+		return processSourceRestoreJob(job);
 	} else if (job.type === "sync_lancedb" || job.type === "sync_lancedb_full") {
 		if (!mediaSourceId) {
 			throw new Error(`Job ${job.id} missing mediaSourceId`);

apps/server/src/application/services/job-transfer-storage.ts

@@ -0,0 +1,140 @@
+import type { Dirent } from "node:fs";
+import { createWriteStream } from "node:fs";
+import fs from "node:fs/promises";
+import path from "node:path";
+import { pipeline } from "node:stream/promises";
+import { webReadableToNodeStream } from "~/infrastructure/utils/stream-utils";
+
+const JobTransferDirectory = path.resolve(
+	process.cwd(),
+	".cache",
+	"job-transfers",
+);
+const JobArtifactTtlMs = 24 * 60 * 60 * 1000;
+
+export type JobTransferMode = "json" | "zip" | "lancedb";
+
+export type JobArtifact = {
+	path: string;
+	fileName: string;
+	contentType: string;
+	size: number;
+	expiresAt: Date;
+};
+
+export function getJobTransferRoot(): string {
+	return JobTransferDirectory;
+}
+
+function getModeExtension(mode: JobTransferMode): string {
+	return mode === "json" ? "ndjson" : "tar";
+}
+
+export function getInputPath(jobId: string, mode: JobTransferMode): string {
+	return path.join(
+		JobTransferDirectory,
+		"inputs",
+		`${jobId}.${getModeExtension(mode)}`,
+	);
+}
+
+export function getArtifactPath(jobId: string, mode: JobTransferMode): string {
+	return path.join(
+		JobTransferDirectory,
+		"artifacts",
+		`${jobId}.${getModeExtension(mode)}`,
+	);
+}
+
+export function isJobTransferPath(jobId: string, targetPath: string): boolean {
+	const resolvedTarget = path.resolve(targetPath);
+	const resolvedRoot = path.resolve(JobTransferDirectory);
+	return (
+		resolvedTarget.startsWith(`${resolvedRoot}${path.sep}`) &&
+		path.basename(resolvedTarget).startsWith(jobId)
+	);
+}
+
+export function getArtifactMetadata(
+	jobId: string,
+	mediaSourceId: string,
+	mode: JobTransferMode,
+): Omit<JobArtifact, "size"> {
+	if (mode === "json") {
+		return {
+			path: getArtifactPath(jobId, mode),
+			fileName: `source-${mediaSourceId}-dump.ndjson`,
+			contentType: "application/x-ndjson",
+			expiresAt: new Date(Date.now() + JobArtifactTtlMs),
+		};
+	}
+
+	return {
+		path: getArtifactPath(jobId, mode),
+		fileName:
+			mode === "lancedb"
+				? `source-${mediaSourceId}-dump-lancedb.tar`
+				: `source-${mediaSourceId}-dump.tar`,
+		contentType: "application/x-tar",
+		expiresAt: new Date(Date.now() + JobArtifactTtlMs),
+	};
+}
+
+export async function persistJobInput(
+	jobId: string,
+	mode: JobTransferMode,
+	file: File,
+): Promise<string> {
+	const inputPath = getInputPath(jobId, mode);
+	await fs.mkdir(path.dirname(inputPath), { recursive: true });
+	await pipeline(
+		webReadableToNodeStream(file.stream()),
+		createWriteStream(inputPath),
+	);
+	return inputPath;
+}
+
+export async function removeJobTransferFile(targetPath: string): Promise<void> {
+	if (!targetPath.startsWith(`${JobTransferDirectory}${path.sep}`)) {
+		return;
+	}
+	await fs.rm(targetPath, { force: true }).catch(() => {});
+}
+
+function isNodeErrorCode(error: unknown, code: string): boolean {
+	return (
+		typeof error === "object" &&
+		error !== null &&
+		"code" in error &&
+		error.code === code
+	);
+}
+
+export async function cleanupExpiredJobTransferFiles(
+	now = Date.now(),
+): Promise<void> {
+	const expirationTime = now - JobArtifactTtlMs;
+	for (const directoryName of ["inputs", "artifacts"] as const) {
+		const directoryPath = path.join(JobTransferDirectory, directoryName);
+		let entries: Dirent[];
+		try {
+			entries = await fs.readdir(directoryPath, { withFileTypes: true });
+		} catch (error) {
+			if (isNodeErrorCode(error, "ENOENT")) {
+				continue;
+			}
+			throw error;
+		}
+
+		for (const entry of entries) {
+			if (!entry.isFile()) {
+				continue;
+			}
+			const targetPath = path.join(directoryPath, entry.name);
+			const stat = await fs.stat(targetPath);
+			if (stat.mtimeMs <= expirationTime) {
+				await removeJobTransferFile(targetPath);
+			}
+		}
+	}
+}

apps/server/src/application/services/source-transfer-job-service.ts

@@ -0,0 +1,84 @@
+import { createWriteStream } from "node:fs";
+import fs from "node:fs/promises";
+import path from "node:path";
+import { pipeline } from "node:stream/promises";
+import {
+	sourceExportJobPayloadSchema,
+	sourceRestoreJobPayloadSchema,
+} from "@solid-imager/core/domain/jobs/schemas";
+import type { Job } from "@solid-imager/core/domain/repositories/job-repository";
+import { services } from "~/application/registry";
+import { BackupService } from "~/application/services/backup-service";
+import {
+	getArtifactMetadata,
+	isJobTransferPath,
+	removeJobTransferFile,
+} from "~/application/services/job-transfer-storage";
+import {
+	asDumpStream,
+	webReadableToNodeStream,
+} from "~/infrastructure/utils/stream-utils";
+
+export async function processSourceExportJob(job: Job): Promise<void> {
+	if (!job.mediaSourceId) {
+		throw new Error(`Job ${job.id} missing mediaSourceId`);
+	}
+
+	const payload = sourceExportJobPayloadSchema.parse(job.payload);
+	const dump = await BackupService.createDump(job.mediaSourceId, payload.mode, {
+		includeImages: payload.includeImages,
+	});
+	const artifact = getArtifactMetadata(job.id, job.mediaSourceId, payload.mode);
+
+	await fs.mkdir(path.dirname(artifact.path), { recursive: true });
+	await pipeline(
+		webReadableToNodeStream(asDumpStream(dump)),
+		createWriteStream(artifact.path),
+	);
+	const stat = await fs.stat(artifact.path);
+	await services.getJobRepository().setArtifact(job.id, {
+		...artifact,
+		size: stat.size,
+	});
+}
+
+export async function processSourceRestoreJob(job: Job): Promise<unknown> {
+	if (!job.mediaSourceId) {
+		throw new Error(`Job ${job.id} missing mediaSourceId`);
+	}
+
+	const payload = sourceRestoreJobPayloadSchema.parse(job.payload);
+	if (!isJobTransferPath(job.id, payload.inputPath)) {
+		throw new Error("Invalid source restore input path");
+	}
+
+	let completed = false;
+	try {
+		if (payload.mode === "json") {
+			const result = await BackupService.importSourceNdjson(
+				job.mediaSourceId,
+				payload.inputPath,
+			);
+			completed = true;
+			return result;
+		}
+		if (payload.mode === "lancedb") {
+			const result = await BackupService.importLanceDB(
+				job.mediaSourceId,
+				payload.inputPath,
+			);
+			completed = true;
+			return result;
+		}
+		const result = await BackupService.importSourceTar(
+			job.mediaSourceId,
+			payload.inputPath,
+		);
+		completed = true;
+		return result;
+	} finally {
+		if (completed) {
+			await removeJobTransferFile(payload.inputPath);
+		}
+	}
+}