diff --git a/apps/server/public/openapi.json b/apps/server/public/openapi.json index e0822273..ee99b511 100644 --- a/apps/server/public/openapi.json +++ b/apps/server/public/openapi.json @@ -582,6 +582,274 @@ } } }, + "/sources/enqueueExport": { + "post": { + "operationId": "sources.enqueueExport", + "summary": "enqueueExport", + "tags": [ + "Media Sources" + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "string", + "format": "uuid" + }, + "mode": { + "enum": [ + "json", + "zip", + "lancedb" + ], + "type": "string", + "default": "json" + }, + "includeImages": { + "type": "boolean", + "default": false + } + }, + "required": [ + "id" + ] + } + } + } + }, + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "string", + "format": "uuid" + }, + "type": { + "type": "string" + }, + "mediaSourceId": { + "anyOf": [ + { + "type": "string", + "format": "uuid" + }, + { + "type": "null" + } + ] + }, + "status": { + "enum": [ + "pending", + "in_progress", + "completed", + "failed", + "cancelled" + ], + "type": "string" + }, + "createdAt": { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + "updatedAt": { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + "parentId": { + "anyOf": [ + { + "type": "string", + "format": "uuid" + }, + { + "type": "null" + } + ] + }, + "error": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "cancelRequestedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "cancelledAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "attemptCount": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "startedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "finishedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "targetMediaId": { + "anyOf": [ + { + "type": "string", + "format": "uuid" + }, + { + "type": "null" + } + ] + }, + "progress": { + "anyOf": [ + { + "type": "object", + "properties": { + "processed": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "failed": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "total": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + } + }, + "required": [ + "processed", + "failed", + "total" + ] + }, + { + "type": "null" + } + ] + }, + "artifact": { + "anyOf": [ + { + "type": "object", + "properties": { + "fileName": { + "type": "string" + }, + "contentType": { + "type": "string" + }, + "size": { + "anyOf": [ + { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + { + "type": "null" + } + ] + }, + "downloadUrl": { + "type": "string", + "minLength": 1 + } + }, + "required": [ + "fileName", + "contentType", + "size", + "downloadUrl" + ] + }, + { + "type": "null" + } + ] + } + }, + "required": [ + "id", + "type", + "mediaSourceId", + "status", + "createdAt", + "updatedAt", + "parentId", + "error", + "cancelRequestedAt", + "cancelledAt", + "attemptCount", + "startedAt", + "finishedAt", + "targetMediaId", + "progress", + "artifact" + ] + } + } + } + } + } + } + }, "/sources/restore": { "post": { "operationId": "sources.restore", @@ -681,6 +949,274 @@ } } }, + "/sources/enqueueImport": { + "post": { + "operationId": "sources.enqueueImport", + "summary": "enqueueImport", + "tags": [ + "Media Sources" + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "string", + "format": "uuid" + }, + "mode": { + "enum": [ + "json", + "zip", + "lancedb" + ], + "type": "string" + }, + "file": { + "not": {} + } + }, + "required": [ + "id", + "mode", + "file" + ] + } + } + } + }, + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "string", + "format": "uuid" + }, + "type": { + "type": "string" + }, + "mediaSourceId": { + "anyOf": [ + { + "type": "string", + "format": "uuid" + }, + { + "type": "null" + } + ] + }, + "status": { + "enum": [ + "pending", + "in_progress", + "completed", + "failed", + "cancelled" + ], + "type": "string" + }, + "createdAt": { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + "updatedAt": { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + "parentId": { + "anyOf": [ + { + "type": "string", + "format": "uuid" + }, + { + "type": "null" + } + ] + }, + "error": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "cancelRequestedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "cancelledAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "attemptCount": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "startedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "finishedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "targetMediaId": { + "anyOf": [ + { + "type": "string", + "format": "uuid" + }, + { + "type": "null" + } + ] + }, + "progress": { + "anyOf": [ + { + "type": "object", + "properties": { + "processed": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "failed": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "total": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + } + }, + "required": [ + "processed", + "failed", + "total" + ] + }, + { + "type": "null" + } + ] + }, + "artifact": { + "anyOf": [ + { + "type": "object", + "properties": { + "fileName": { + "type": "string" + }, + "contentType": { + "type": "string" + }, + "size": { + "anyOf": [ + { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + { + "type": "null" + } + ] + }, + "downloadUrl": { + "type": "string", + "minLength": 1 + } + }, + "required": [ + "fileName", + "contentType", + "size", + "downloadUrl" + ] + }, + { + "type": "null" + } + ] + } + }, + "required": [ + "id", + "type", + "mediaSourceId", + "status", + "createdAt", + "updatedAt", + "parentId", + "error", + "cancelRequestedAt", + "cancelledAt", + "attemptCount", + "startedAt", + "finishedAt", + "targetMediaId", + "progress", + "artifact" + ] + } + } + } + } + } + } + }, "/sources/importNdjson": { "post": { "operationId": "sources.importNdjson", @@ -6796,7 +7332,8 @@ "pending", "in_progress", "completed", - "failed" + "failed", + "cancelled" ], "type": "string" }, @@ -6853,7 +7390,8 @@ "pending", "in_progress", "completed", - "failed" + "failed", + "cancelled" ], "type": "string" }, @@ -6888,6 +7426,59 @@ } ] }, + "cancelRequestedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "cancelledAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "attemptCount": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "startedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "finishedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, "targetMediaId": { "anyOf": [ { @@ -6930,31 +7521,329 @@ "type": "null" } ] + }, + "artifact": { + "anyOf": [ + { + "type": "object", + "properties": { + "fileName": { + "type": "string" + }, + "contentType": { + "type": "string" + }, + "size": { + "anyOf": [ + { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + { + "type": "null" + } + ] + }, + "downloadUrl": { + "type": "string", + "minLength": 1 + } + }, + "required": [ + "fileName", + "contentType", + "size", + "downloadUrl" + ] + }, + { + "type": "null" + } + ] } }, - "required": [ - "id", - "type", - "mediaSourceId", - "status", - "createdAt", - "updatedAt", - "parentId", - "error", - "targetMediaId", - "progress" - ] - } + "required": [ + "id", + "type", + "mediaSourceId", + "status", + "createdAt", + "updatedAt", + "parentId", + "error", + "cancelRequestedAt", + "cancelledAt", + "attemptCount", + "startedAt", + "finishedAt", + "targetMediaId", + "progress", + "artifact" + ] + } + }, + "total": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + } + }, + "required": [ + "items", + "total" + ] + } + } + } + } + } + } + }, + "/jobs/get": { + "post": { + "operationId": "jobs.get", + "summary": "get", + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "string", + "format": "uuid" + } + }, + "required": [ + "id" + ] + } + } + } + }, + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": { + "type": "string", + "format": "uuid" + }, + "type": { + "type": "string" + }, + "mediaSourceId": { + "anyOf": [ + { + "type": "string", + "format": "uuid" + }, + { + "type": "null" + } + ] + }, + "status": { + "enum": [ + "pending", + "in_progress", + "completed", + "failed", + "cancelled" + ], + "type": "string" + }, + "createdAt": { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + "updatedAt": { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + "parentId": { + "anyOf": [ + { + "type": "string", + "format": "uuid" + }, + { + "type": "null" + } + ] + }, + "error": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "cancelRequestedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "cancelledAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "attemptCount": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "startedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "finishedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "targetMediaId": { + "anyOf": [ + { + "type": "string", + "format": "uuid" + }, + { + "type": "null" + } + ] + }, + "progress": { + "anyOf": [ + { + "type": "object", + "properties": { + "processed": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "failed": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "total": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + } + }, + "required": [ + "processed", + "failed", + "total" + ] + }, + { + "type": "null" + } + ] }, - "total": { - "type": "integer", - "minimum": 0, - "maximum": 9007199254740991 + "artifact": { + "anyOf": [ + { + "type": "object", + "properties": { + "fileName": { + "type": "string" + }, + "contentType": { + "type": "string" + }, + "size": { + "anyOf": [ + { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + { + "type": "null" + } + ] + }, + "downloadUrl": { + "type": "string", + "minLength": 1 + } + }, + "required": [ + "fileName", + "contentType", + "size", + "downloadUrl" + ] + }, + { + "type": "null" + } + ] } }, "required": [ - "items", - "total" + "id", + "type", + "mediaSourceId", + "status", + "createdAt", + "updatedAt", + "parentId", + "error", + "cancelRequestedAt", + "cancelledAt", + "attemptCount", + "startedAt", + "finishedAt", + "targetMediaId", + "progress", + "artifact" ] } } @@ -6963,10 +7852,10 @@ } } }, - "/jobs/get": { + "/jobs/retry": { "post": { - "operationId": "jobs.get", - "summary": "get", + "operationId": "jobs.retry", + "summary": "retry", "requestBody": { "required": true, "content": { @@ -7017,7 +7906,8 @@ "pending", "in_progress", "completed", - "failed" + "failed", + "cancelled" ], "type": "string" }, @@ -7052,6 +7942,59 @@ } ] }, + "cancelRequestedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "cancelledAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "attemptCount": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "startedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "finishedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, "targetMediaId": { "anyOf": [ { @@ -7094,6 +8037,46 @@ "type": "null" } ] + }, + "artifact": { + "anyOf": [ + { + "type": "object", + "properties": { + "fileName": { + "type": "string" + }, + "contentType": { + "type": "string" + }, + "size": { + "anyOf": [ + { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + { + "type": "null" + } + ] + }, + "downloadUrl": { + "type": "string", + "minLength": 1 + } + }, + "required": [ + "fileName", + "contentType", + "size", + "downloadUrl" + ] + }, + { + "type": "null" + } + ] } }, "required": [ @@ -7105,8 +8088,14 @@ "updatedAt", "parentId", "error", + "cancelRequestedAt", + "cancelledAt", + "attemptCount", + "startedAt", + "finishedAt", "targetMediaId", - "progress" + "progress", + "artifact" ] } } @@ -7115,10 +8104,10 @@ } } }, - "/jobs/retry": { + "/jobs/cancel": { "post": { - "operationId": "jobs.retry", - "summary": "retry", + "operationId": "jobs.cancel", + "summary": "cancel", "requestBody": { "required": true, "content": { @@ -7169,7 +8158,8 @@ "pending", "in_progress", "completed", - "failed" + "failed", + "cancelled" ], "type": "string" }, @@ -7204,6 +8194,59 @@ } ] }, + "cancelRequestedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "cancelledAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "attemptCount": { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + "startedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, + "finishedAt": { + "anyOf": [ + { + "type": "string", + "format": "date-time", + "x-native-type": "date" + }, + { + "type": "null" + } + ] + }, "targetMediaId": { "anyOf": [ { @@ -7246,6 +8289,46 @@ "type": "null" } ] + }, + "artifact": { + "anyOf": [ + { + "type": "object", + "properties": { + "fileName": { + "type": "string" + }, + "contentType": { + "type": "string" + }, + "size": { + "anyOf": [ + { + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + { + "type": "null" + } + ] + }, + "downloadUrl": { + "type": "string", + "minLength": 1 + } + }, + "required": [ + "fileName", + "contentType", + "size", + "downloadUrl" + ] + }, + { + "type": "null" + } + ] } }, "required": [ @@ -7257,8 +8340,14 @@ "updatedAt", "parentId", "error", + "cancelRequestedAt", + "cancelledAt", + "attemptCount", + "startedAt", + "finishedAt", "targetMediaId", - "progress" + "progress", + "artifact" ] } } @@ -7384,6 +8473,29 @@ "event", "data" ] + }, + { + "type": "object", + "properties": { + "event": { + "const": "job-cancelled" + }, + "data": { + "type": "object", + "properties": { + "jobId": { + "type": "string" + }, + "message": { + "type": "string" + } + } + } + }, + "required": [ + "event", + "data" + ] } ] }, diff --git a/apps/server/src/application/services/directory-sync-service.ts b/apps/server/src/application/services/directory-sync-service.ts index 498c9223..69f507ae 100644 --- a/apps/server/src/application/services/directory-sync-service.ts +++ b/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 { + 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 { + await persistSyncStatus(mediaSourceId, status, message); + publishSyncStatus(mediaSourceId, status, message); +} + export function getSourceSyncState( mediaSourceId: string, ): MediaSourceSyncState { @@ -173,7 +205,7 @@ 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") { @@ -181,7 +213,7 @@ export const DirectorySyncService = { { 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, diff --git a/apps/server/src/application/services/job-dispatch-service.ts b/apps/server/src/application/services/job-dispatch-service.ts index 7a5be13a..2c41d864 100644 --- a/apps/server/src/application/services/job-dispatch-service.ts +++ b/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`); diff --git a/apps/server/src/application/services/job-transfer-storage.ts b/apps/server/src/application/services/job-transfer-storage.ts new file mode 100644 index 00000000..b8cf9db3 --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + 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); + } + } + } +} diff --git a/apps/server/src/application/services/source-transfer-job-service.ts b/apps/server/src/application/services/source-transfer-job-service.ts new file mode 100644 index 00000000..75684682 --- /dev/null +++ b/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 { + 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 { + 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); + } + } +} diff --git a/apps/server/src/infrastructure/api/routers/jobs-router.ts b/apps/server/src/infrastructure/api/routers/jobs-router.ts index 0f47de8b..65511597 100644 --- a/apps/server/src/infrastructure/api/routers/jobs-router.ts +++ b/apps/server/src/infrastructure/api/routers/jobs-router.ts @@ -1,11 +1,13 @@ import { eventIterator, ORPCError, os } from "@orpc/server"; import { + isBatchParentJobType, jobDtoSchema, jobIdRequestSchema, jobListRequestSchema, jobListResponseSchema, jobStatusSchema, } from "@solid-imager/core/domain/jobs/schemas"; +import type { Job } from "@solid-imager/core/domain/repositories/job-repository"; import { type JobEvent, jobEventSchema, @@ -15,8 +17,8 @@ import { z } from "zod"; import { db } from "~/infrastructure/db"; import { jobs } from "~/infrastructure/db/schema"; import { RealtimeEventBus } from "~/infrastructure/events/realtime-event-bus"; +import { JobRepository } from "~/infrastructure/repositories/job-repository"; -type JobRow = typeof jobs.$inferSelect; const PublicJobFailureMessage = "Job failed"; function isRecord(value: unknown): value is Record { @@ -53,7 +55,7 @@ function readProgress(payload: unknown) { }; } -function toJobDto(job: JobRow) { +export function toJobDto(job: Job) { return { id: job.id, type: job.type, @@ -63,11 +65,11 @@ function toJobDto(job: JobRow) { updatedAt: job.updatedAt, parentId: job.parentId, error: job.error === null ? null : PublicJobFailureMessage, - cancelRequestedAt: job.cancelRequestedAt, - cancelledAt: job.cancelledAt, - attemptCount: job.attemptCount, - startedAt: job.startedAt, - finishedAt: job.finishedAt, + cancelRequestedAt: job.cancelRequestedAt ?? null, + cancelledAt: job.cancelledAt ?? null, + attemptCount: job.attemptCount ?? 0, + startedAt: job.startedAt ?? null, + finishedAt: job.finishedAt ?? null, targetMediaId: readTargetMediaId(job.payload), progress: readProgress(job.payload), artifact: @@ -75,7 +77,7 @@ function toJobDto(job: JobRow) { ? { fileName: job.artifactFileName, contentType: job.artifactContentType, - size: job.artifactSize, + size: job.artifactSize ?? null, downloadUrl: `/api/jobs/${job.id}/artifact`, } : null, @@ -130,6 +132,15 @@ export const jobsRouter = { status: "pending", error: null, result: null, + cancelRequestedAt: null, + cancelledAt: null, + startedAt: null, + finishedAt: null, + artifactPath: null, + artifactFileName: null, + artifactContentType: null, + artifactSize: null, + artifactExpiresAt: null, updatedAt: new Date(), }) .where(and(eq(jobs.id, input.id), eq(jobs.status, "failed"))) @@ -153,6 +164,40 @@ export const jobsRouter = { return toJobDto(requeued); }), + cancel: os + .input(jobIdRequestSchema) + .output(jobDtoSchema) + .handler(async ({ input }) => { + const job = await JobRepository.findById(input.id); + if (!job) { + throw new ORPCError("NOT_FOUND", { message: "Job not found" }); + } + if (job.status !== "pending" && job.status !== "in_progress") { + throw new ORPCError("BAD_REQUEST", { + message: "Only pending or in-progress jobs can be cancelled", + }); + } + if (isBatchParentJobType(job.type)) { + throw new ORPCError("BAD_REQUEST", { + message: "Batch parent jobs cannot be cancelled", + }); + } + + await JobRepository.requestCancellation(input.id); + const cancelled = await JobRepository.findById(input.id); + if (!cancelled) { + throw new ORPCError("NOT_FOUND", { message: "Job not found" }); + } + RealtimeEventBus.publishJob("job-cancelled", { + jobId: input.id, + message: + cancelled.status === "in_progress" + ? "Cancellation requested" + : "Job cancelled", + }); + return toJobDto(cancelled); + }), + events: os.output(eventIterator(jobEventSchema)).handler(async function* ({ signal, }) { diff --git a/apps/server/src/infrastructure/api/routers/sources-router.ts b/apps/server/src/infrastructure/api/routers/sources-router.ts index 6b9061c4..1b0f7e6f 100644 --- a/apps/server/src/infrastructure/api/routers/sources-router.ts +++ b/apps/server/src/infrastructure/api/routers/sources-router.ts @@ -1,4 +1,5 @@ -import { eventIterator, os } from "@orpc/server"; +import { eventIterator, ORPCError, os } from "@orpc/server"; +import { jobDtoSchema } from "@solid-imager/core/domain/jobs/schemas"; import type { MediaSource } from "@solid-imager/core/domain/repositories/source-repository"; import { type SourceEvent, @@ -16,13 +17,13 @@ import { asyncPool } from "@solid-imager/core/utils/async-pool"; import { isRecord } from "@solid-imager/core/utils/type-guards"; import { count, inArray } from "drizzle-orm"; import { z } from "zod"; +import { services } from "~/application/registry"; import { BackupService } from "~/application/services/backup-service"; -import { - DirectorySyncService, - getSourceSyncState, -} from "~/application/services/directory-sync-service"; +import { DirectorySyncService } from "~/application/services/directory-sync-service"; +import { persistJobInput } from "~/application/services/job-transfer-storage"; import { MediaService } from "~/application/services/media-service"; import { MediaSourceService } from "~/application/services/media-source-service"; +import { toJobDto } from "~/infrastructure/api/routers/jobs-router"; import { db } from "~/infrastructure/db"; import { medias } from "~/infrastructure/db/schema"; import { RealtimeEventBus } from "~/infrastructure/events/realtime-event-bus"; @@ -104,7 +105,7 @@ function addSourceSummary( return { ...source, mediaCount: mediaCounts.get(source.id) ?? 0, - syncStatus: getSourceSyncState(source.id), + syncStatus: source.syncStatus ?? "idle", }; } @@ -349,6 +350,34 @@ export const sourcesRouter = { }, }); }), + + enqueueExport: os + .input( + z.object({ + id: z.string().uuid(), + mode: z.enum(["json", "zip", "lancedb"]).default("json"), + includeImages: z.boolean().default(false), + }), + ) + .output(jobDtoSchema) + .handler(async ({ input }) => { + const [source] = await MediaSourceService.fetchSourceById(input.id); + if (!source) { + throw new ORPCError("NOT_FOUND", { + message: `Source not found: ${input.id}`, + }); + } + + const job = await services.getJobRepository().create({ + type: "source_export", + mediaSourceId: input.id, + payload: { + mode: input.mode, + includeImages: input.includeImages, + }, + }); + return toJobDto(job); + }), restore: os .meta({ openapi: { @@ -413,6 +442,44 @@ export const sourcesRouter = { } }), + enqueueImport: os + .input( + z.object({ + id: z.string().uuid(), + mode: z.enum(["json", "zip", "lancedb"]), + file: z.instanceof(File), + }), + ) + .output(jobDtoSchema) + .handler(async ({ input }) => { + const [source] = await MediaSourceService.fetchSourceById(input.id); + if (!source) { + throw new ORPCError("NOT_FOUND", { + message: `Source not found: ${input.id}`, + }); + } + + const { randomUUID } = await import("node:crypto"); + const jobId = randomUUID(); + const inputPath = await persistJobInput(jobId, input.mode, input.file); + try { + const job = await services.getJobRepository().create({ + id: jobId, + type: "source_restore", + mediaSourceId: input.id, + payload: { + mode: input.mode, + inputPath, + }, + }); + return toJobDto(job); + } catch (error) { + const fs = await import("node:fs/promises"); + await fs.rm(inputPath, { force: true }).catch(() => {}); + throw error; + } + }), + /** * Imports a media source from a streaming NDJSON file */ diff --git a/apps/server/src/infrastructure/jobs/job-worker.ts b/apps/server/src/infrastructure/jobs/job-worker.ts index accec26b..be24f0ed 100644 --- a/apps/server/src/infrastructure/jobs/job-worker.ts +++ b/apps/server/src/infrastructure/jobs/job-worker.ts @@ -1,6 +1,8 @@ import type { AppConfig } from "@solid-imager/core/domain/config/config-schema"; +import { cleanupExpiredJobTransferFiles } from "~/application/services/job-transfer-storage"; import type { IJobRepository } from "~/domain/repositories/job-repository"; import type { Job } from "~/infrastructure/db/schema"; +import { RealtimeEventBus } from "~/infrastructure/events/realtime-event-bus"; import { logger } from "~/infrastructure/logger"; type JsonSafeValue = @@ -38,6 +40,8 @@ function toJsonSafeValue(value: unknown): JsonSafeValue { } const StaleInProgressJobMs = 60 * 60 * 1000; +const JobHeartbeatMs = 5 * 60 * 1000; +const PublicJobFailureMessage = "Job failed"; export class JobWorker { private isRunning = false; @@ -211,6 +215,16 @@ export class JobWorker { if (isThumbnailJob) { this.activeThumbnailJobs++; } + const heartbeatId = setInterval(() => { + void Promise.resolve( + this.jobRepo.update(job.id, { updatedAt: new Date() }), + ).catch((error) => { + logger.error( + { err: error, jobId: job.id }, + "Failed to update job heartbeat", + ); + }); + }, JobHeartbeatMs); logger.info( { @@ -223,10 +237,29 @@ export class JobWorker { "Job started", ); try { + if (await this.jobRepo.isCancellationRequested(job.id)) { + await this.markCancelled(job); + return; + } + const result = await this.processor(job); const safeResult = result !== undefined ? toJsonSafeValue(result) : { success: true }; - await this.jobRepo.markAsCompleted(job.id, safeResult); + if (await this.jobRepo.isCancellationRequested(job.id)) { + await this.markCancelled(job); + return; + } else { + await this.jobRepo.markAsCompleted(job.id, safeResult); + const latestJob = await this.jobRepo.findById(job.id); + if (latestJob?.cancelRequestedAt || latestJob?.status === "cancelled") { + await this.markCancelled(job); + return; + } + RealtimeEventBus.publishJob("job-completed", { + jobId: job.id, + message: "Job completed", + }); + } logger.info( { jobId: job.id, @@ -240,6 +273,11 @@ export class JobWorker { } catch (error) { const errorMessage = error instanceof Error ? error.message : String(error); + if (await this.jobRepo.isCancellationRequested(job.id)) { + await this.markCancelled(job); + return; + } + logger.error( { err: error, @@ -252,7 +290,17 @@ export class JobWorker { "Job failed", ); await this.jobRepo.markAsFailed(job.id, errorMessage); + const latestJob = await this.jobRepo.findById(job.id); + if (latestJob?.cancelRequestedAt || latestJob?.status === "cancelled") { + await this.markCancelled(job); + return; + } + RealtimeEventBus.publishJob("job-failed", { + jobId: job.id, + error: PublicJobFailureMessage, + }); } finally { + clearInterval(heartbeatId); this.activeJobs--; if (isAiJob) { this.activeAiJobs--; @@ -265,6 +313,23 @@ export class JobWorker { } } } + + private async markCancelled(job: Job): Promise { + await this.jobRepo.markAsCancelled(job.id); + RealtimeEventBus.publishJob("job-cancelled", { + jobId: job.id, + message: "Job cancelled", + }); + logger.info( + { + jobId: job.id, + type: job.type, + mediaSourceId: job.mediaSourceId, + parentId: job.parentId, + }, + "Job cancelled", + ); + } private async recoverStaleJobs() { const olderThan = new Date(Date.now() - StaleInProgressJobMs); try { @@ -272,8 +337,12 @@ export class JobWorker { if (count > 0) { logger.warn({ count, olderThan }, "Requeued stale in-progress jobs"); } + await cleanupExpiredJobTransferFiles(); } catch (error) { - logger.error({ err: error }, "Failed to requeue stale in-progress jobs"); + logger.error( + { err: error }, + "Failed to recover stale jobs or clean up transfer files", + ); } } } diff --git a/apps/server/src/routes/api/jobs.$jobId.artifact.ts b/apps/server/src/routes/api/jobs.$jobId.artifact.ts new file mode 100644 index 00000000..18567726 --- /dev/null +++ b/apps/server/src/routes/api/jobs.$jobId.artifact.ts @@ -0,0 +1,51 @@ +import { createReadStream } from "node:fs"; +import fs from "node:fs/promises"; +import { createFileRoute } from "@tanstack/solid-router"; +import { services } from "~/application/registry"; +import { isJobTransferPath } from "~/application/services/job-transfer-storage"; +import type { ServerRouteContext } from "~/infrastructure/router/route-types"; +import { bootstrapServerRoute } from "~/infrastructure/server-route-bootstrap"; +import { nodeStreamToWebReadable } from "~/infrastructure/utils/stream-utils"; + +export const Route = createFileRoute("/api/jobs/$jobId/artifact")({ + server: { + handlers: { + GET: async ({ params }: ServerRouteContext<{ jobId: string }>) => { + bootstrapServerRoute(); + const job = await services.getJobRepository().findById(params.jobId); + if ( + job?.status !== "completed" || + !job.artifactPath || + !job.artifactFileName || + !job.artifactContentType || + !isJobTransferPath(job.id, job.artifactPath) + ) { + return new Response("Artifact not found", { status: 404 }); + } + + if (job.artifactExpiresAt && job.artifactExpiresAt <= new Date()) { + return new Response("Artifact expired", { status: 410 }); + } + + let stat: Awaited>; + try { + stat = await fs.stat(job.artifactPath); + } catch { + return new Response("Artifact not found", { status: 404 }); + } + + return new Response( + nodeStreamToWebReadable(createReadStream(job.artifactPath)), + { + headers: { + "Cache-Control": "private, max-age=3600", + "Content-Length": String(stat.size), + "Content-Type": job.artifactContentType, + "Content-Disposition": `attachment; filename="${encodeURIComponent(job.artifactFileName)}"`, + }, + }, + ); + }, + }, + }, +}); diff --git a/apps/server/src/tests/unit/application/services/directory-sync-service.test.ts b/apps/server/src/tests/unit/application/services/directory-sync-service.test.ts index 698bd468..acd1a23a 100644 --- a/apps/server/src/tests/unit/application/services/directory-sync-service.test.ts +++ b/apps/server/src/tests/unit/application/services/directory-sync-service.test.ts @@ -49,6 +49,7 @@ vi.mock("~/infrastructure/repositories/source-repository", () => ({ connectionInfo: { path: "/fake/path" }, }), findAll: vi.fn().mockResolvedValue([]), + update: vi.fn().mockResolvedValue(undefined), }, })); vi.mock("~/application/services/media-processing-service", () => ({ diff --git a/apps/server/src/tests/unit/application/services/media-service.test.ts b/apps/server/src/tests/unit/application/services/media-service.test.ts index bbb3a379..6a680039 100644 --- a/apps/server/src/tests/unit/application/services/media-service.test.ts +++ b/apps/server/src/tests/unit/application/services/media-service.test.ts @@ -105,6 +105,10 @@ describe("MediaService Unit Tests", () => { markAsInProgress: vi.fn(), markAsCompleted: vi.fn(), markAsFailed: vi.fn(), + requestCancellation: vi.fn(), + markAsCancelled: vi.fn(), + isCancellationRequested: vi.fn(), + setArtifact: vi.fn(), update: vi.fn(), incrementProgress: vi.fn(), incrementFailedCount: vi.fn(), diff --git a/apps/server/src/tests/unit/infrastructure/jobs/ccip-jobs.test.ts b/apps/server/src/tests/unit/infrastructure/jobs/ccip-jobs.test.ts index 46e1e6c8..a61c9e96 100644 --- a/apps/server/src/tests/unit/infrastructure/jobs/ccip-jobs.test.ts +++ b/apps/server/src/tests/unit/infrastructure/jobs/ccip-jobs.test.ts @@ -21,6 +21,10 @@ const jobRepository: IJobRepository = { markAsInProgress: vi.fn(), markAsCompleted: vi.fn(), markAsFailed: vi.fn(), + requestCancellation: vi.fn(), + markAsCancelled: vi.fn(), + isCancellationRequested: vi.fn(), + setArtifact: vi.fn(), update: (...args: Parameters) => update(...args), incrementProgress: (...args: Parameters) => incrementProgress(...args), diff --git a/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts b/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts index 81e7a44e..ec17b396 100644 --- a/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts +++ b/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts @@ -39,6 +39,10 @@ describe("JobWorker", () => { markAsInProgress: vi.fn().mockResolvedValue(undefined), markAsCompleted: vi.fn().mockResolvedValue(undefined), markAsFailed: vi.fn().mockResolvedValue(undefined), + requestCancellation: vi.fn().mockResolvedValue(undefined), + markAsCancelled: vi.fn().mockResolvedValue(undefined), + isCancellationRequested: vi.fn().mockResolvedValue(false), + setArtifact: vi.fn().mockResolvedValue(undefined), update: vi.fn(), incrementProgress: vi.fn(), incrementFailedCount: vi.fn(), diff --git a/apps/server/src/tests/unit/infrastructure/jobs/tagging-jobs.test.ts b/apps/server/src/tests/unit/infrastructure/jobs/tagging-jobs.test.ts index a8c54a48..7a848538 100644 --- a/apps/server/src/tests/unit/infrastructure/jobs/tagging-jobs.test.ts +++ b/apps/server/src/tests/unit/infrastructure/jobs/tagging-jobs.test.ts @@ -22,6 +22,10 @@ const jobRepository: IJobRepository = { markAsInProgress: vi.fn(), markAsCompleted: vi.fn(), markAsFailed: vi.fn(), + requestCancellation: vi.fn(), + markAsCancelled: vi.fn(), + isCancellationRequested: vi.fn(), + setArtifact: vi.fn(), update: (...args: Parameters) => update(...args), incrementProgress: (...args: Parameters) => incrementProgress(...args), diff --git a/packages/core/src/domain/contract/jobs.contract.ts b/packages/core/src/domain/contract/jobs.contract.ts index f5946059..1ff0ca9b 100644 --- a/packages/core/src/domain/contract/jobs.contract.ts +++ b/packages/core/src/domain/contract/jobs.contract.ts @@ -11,5 +11,6 @@ export const jobsContract = { list: oc.input(jobListRequestSchema).output(jobListResponseSchema), get: oc.input(jobIdRequestSchema).output(jobDtoSchema), retry: oc.input(jobIdRequestSchema).output(jobDtoSchema), + cancel: oc.input(jobIdRequestSchema).output(jobDtoSchema), events: oc.output(eventIterator(jobEventSchema)), }; diff --git a/packages/core/src/domain/contract/sources.contract.ts b/packages/core/src/domain/contract/sources.contract.ts index 704a8946..51f77017 100644 --- a/packages/core/src/domain/contract/sources.contract.ts +++ b/packages/core/src/domain/contract/sources.contract.ts @@ -1,5 +1,6 @@ import { eventIterator, oc } from "@orpc/contract"; import { z } from "zod"; +import { jobDtoSchema } from "../jobs/schemas"; import { sourceEventSchema } from "../sources/events"; import { mediaSourceInfoSchema, @@ -63,6 +64,16 @@ export const sourcesContract = { }), ), + enqueueExport: oc + .input( + z.object({ + id: z.string().uuid(), + mode: z.enum(["json", "zip", "lancedb"]).default("json"), + includeImages: z.boolean().default(false), + }), + ) + .output(jobDtoSchema), + restore: oc .input( z.object({ @@ -106,6 +117,16 @@ export const sourcesContract = { ) .output(importResultSchema), + enqueueImport: oc + .input( + z.object({ + id: z.string().uuid(), + mode: z.enum(["json", "zip", "lancedb"]), + file: z.instanceof(File), + }), + ) + .output(jobDtoSchema), + status: oc .input(z.object({ id: z.string().uuid() })) .output(mediaSourceStatusSchema), diff --git a/packages/core/src/domain/jobs/schemas.ts b/packages/core/src/domain/jobs/schemas.ts index 08bb276c..e3fd94aa 100644 --- a/packages/core/src/domain/jobs/schemas.ts +++ b/packages/core/src/domain/jobs/schemas.ts @@ -9,6 +9,16 @@ export const jobStatusSchema = z.enum([ ]); export type JobStatus = z.infer; +export const batchParentJobTypes = [ + "batch_ccip_parent", + "bulk_tagging_parent", + "thumbnail_generation_parent", +] as const; + +export function isBatchParentJobType(type: string): boolean { + return batchParentJobTypes.some((parentType) => parentType === type); +} + export const jobProgressSchema = z .object({ processed: z.number().int().nonnegative(), @@ -65,3 +75,19 @@ export const jobIdRequestSchema = z.object({ id: z.string().uuid(), }); export type JobIdRequest = z.infer; + +export const sourceExportJobPayloadSchema = z.object({ + mode: z.enum(["json", "zip", "lancedb"]), + includeImages: z.boolean(), +}); +export type SourceExportJobPayload = z.infer< + typeof sourceExportJobPayloadSchema +>; + +export const sourceRestoreJobPayloadSchema = z.object({ + mode: z.enum(["json", "zip", "lancedb"]), + inputPath: z.string().min(1), +}); +export type SourceRestoreJobPayload = z.infer< + typeof sourceRestoreJobPayloadSchema +>; diff --git a/packages/core/src/domain/repositories/job-repository.ts b/packages/core/src/domain/repositories/job-repository.ts index 99f08b67..d2c9a47f 100644 --- a/packages/core/src/domain/repositories/job-repository.ts +++ b/packages/core/src/domain/repositories/job-repository.ts @@ -72,6 +72,19 @@ export type IJobRepository = { markAsInProgress(id: string): Promise; markAsCompleted(id: string, result?: unknown): Promise; markAsFailed(id: string, error: string): Promise; + requestCancellation(id: string): Promise; + markAsCancelled(id: string, reason?: string): Promise; + isCancellationRequested(id: string): Promise; + setArtifact( + id: string, + artifact: { + path: string; + fileName: string; + contentType: string; + size: number; + expiresAt: Date; + }, + ): Promise; update(id: string, data: Partial): Promise; incrementProgress( id: string, diff --git a/packages/core/src/domain/sources/events.ts b/packages/core/src/domain/sources/events.ts index 75db80cd..4a41d26d 100644 --- a/packages/core/src/domain/sources/events.ts +++ b/packages/core/src/domain/sources/events.ts @@ -97,6 +97,12 @@ export const jobFailedEventSchema = z.object({ }); export type JobFailedEvent = z.infer; +export const jobCancelledEventSchema = z.object({ + jobId: z.string().optional(), + message: z.string().optional(), +}); +export type JobCancelledEvent = z.infer; + export const downloadErrorEventSchema = z.object({ url: z.string(), error: z.string(), @@ -183,6 +189,10 @@ export const jobEventSchema = z.discriminatedUnion("event", [ data: jobCompletedEventSchema, }), z.object({ event: z.literal("job-failed"), data: jobFailedEventSchema }), + z.object({ + event: z.literal("job-cancelled"), + data: jobCancelledEventSchema, + }), ]); export type JobEvent = z.infer; export type JobEventName = JobEvent["event"]; diff --git a/packages/db/src/repositories/job-repository.test.ts b/packages/db/src/repositories/job-repository.test.ts index 768376b4..8b22d628 100644 --- a/packages/db/src/repositories/job-repository.test.ts +++ b/packages/db/src/repositories/job-repository.test.ts @@ -98,12 +98,13 @@ describe("JobRepository", () => { }); it("returns the number of stale jobs requeued", async () => { + mockExecutor.returning.mockResolvedValueOnce([]); const count = await repository.requeueStaleInProgress( new Date("2026-06-23T00:00:00.000Z"), ); expect(count).toBe(1); - expect(mockExecutor.update).toHaveBeenCalledOnce(); + expect(mockExecutor.update).toHaveBeenCalledTimes(2); }); it("does not requeue thumbnail batch parent progress records", async () => { @@ -133,15 +134,17 @@ describe("JobRepository", () => { }); it("clears stale result and error markers when requeueing", async () => { + mockExecutor.returning.mockResolvedValueOnce([]); await repository.requeueStaleInProgress( new Date("2026-06-23T00:00:00.000Z"), ); - expect(mockExecutor.set).toHaveBeenCalledOnce(); - expect(mockExecutor.set).toHaveBeenCalledWith({ + expect(mockExecutor.set).toHaveBeenCalledTimes(2); + expect(mockExecutor.set).toHaveBeenLastCalledWith({ status: "pending", result: null, error: null, + startedAt: null, updatedAt: expect.any(Date), }); }); diff --git a/packages/db/src/repositories/job-repository.ts b/packages/db/src/repositories/job-repository.ts index 97bf7c79..951371a9 100644 --- a/packages/db/src/repositories/job-repository.ts +++ b/packages/db/src/repositories/job-repository.ts @@ -1,3 +1,4 @@ +import { batchParentJobTypes } from "@solid-imager/core/domain/jobs/schemas"; import type { BatchProgress, IJobRepository, @@ -13,6 +14,7 @@ import { eq, inArray, isNotNull, + isNull, lt, ne, not, @@ -246,32 +248,138 @@ export function createJobRepository( }, async markAsInProgress(id: string): Promise { + const now = new Date(); await db() .update(jobs) .set({ status: "in_progress", - updatedAt: new Date(), + attemptCount: sql`${jobs.attemptCount} + 1`, + startedAt: now, + finishedAt: null, + cancelRequestedAt: null, + cancelledAt: null, + updatedAt: now, }) .where(eq(jobs.id, id)); }, async markAsCompleted(id: string, result?: unknown): Promise { + const now = new Date(); await db() .update(jobs) .set({ status: "completed", result: result ?? null, - updatedAt: new Date(), + finishedAt: now, + updatedAt: now, }) - .where(eq(jobs.id, id)); + .where( + and( + eq(jobs.id, id), + eq(jobs.status, "in_progress"), + isNull(jobs.cancelRequestedAt), + ), + ); }, async markAsFailed(id: string, error: string): Promise { + const now = new Date(); await db() .update(jobs) .set({ status: "failed", error, + finishedAt: now, + updatedAt: now, + }) + .where( + and( + eq(jobs.id, id), + eq(jobs.status, "in_progress"), + isNull(jobs.cancelRequestedAt), + ), + ); + }, + + async requestCancellation(id: string): Promise { + const now = new Date(); + const cancelledPending = await db() + .update(jobs) + .set({ + status: "cancelled", + cancelRequestedAt: now, + cancelledAt: now, + finishedAt: now, + updatedAt: now, + }) + .where(and(eq(jobs.id, id), eq(jobs.status, "pending"))) + .returning(); + + if (cancelledPending.length > 0) { + return; + } + + await db() + .update(jobs) + .set({ + cancelRequestedAt: now, + updatedAt: now, + }) + .where( + and( + eq(jobs.id, id), + eq(jobs.status, "in_progress"), + isNull(jobs.cancelRequestedAt), + ), + ); + }, + + async markAsCancelled(id: string, reason?: string): Promise { + const now = new Date(); + const updates: Partial = { + status: "cancelled", + cancelRequestedAt: now, + cancelledAt: now, + finishedAt: now, + updatedAt: now, + }; + if (reason) { + updates.error = reason; + } + + await db() + .update(jobs) + .set(updates) + .where( + and( + eq(jobs.id, id), + inArray(jobs.status, ["pending", "in_progress"]), + ), + ); + }, + + async isCancellationRequested(id: string): Promise { + const [job] = await db() + .select({ + status: jobs.status, + cancelRequestedAt: jobs.cancelRequestedAt, + }) + .from(jobs) + .where(eq(jobs.id, id)); + return Boolean( + job && (job.status === "cancelled" || job.cancelRequestedAt !== null), + ); + }, + + async setArtifact(id, artifact): Promise { + await db() + .update(jobs) + .set({ + artifactPath: artifact.path, + artifactFileName: artifact.fileName, + artifactContentType: artifact.contentType, + artifactSize: artifact.size, + artifactExpiresAt: artifact.expiresAt, updatedAt: new Date(), }) .where(eq(jobs.id, id)); @@ -287,6 +395,24 @@ export function createJobRepository( if (data.result !== undefined) updates.result = data.result; if (data.error !== undefined) updates.error = data.error; if (data.parentId !== undefined) updates.parentId = data.parentId; + if (data.cancelRequestedAt !== undefined) + updates.cancelRequestedAt = data.cancelRequestedAt; + if (data.cancelledAt !== undefined) + updates.cancelledAt = data.cancelledAt; + if (data.attemptCount !== undefined) + updates.attemptCount = data.attemptCount; + if (data.startedAt !== undefined) updates.startedAt = data.startedAt; + if (data.finishedAt !== undefined) updates.finishedAt = data.finishedAt; + if (data.artifactPath !== undefined) + updates.artifactPath = data.artifactPath; + if (data.artifactFileName !== undefined) + updates.artifactFileName = data.artifactFileName; + if (data.artifactContentType !== undefined) + updates.artifactContentType = data.artifactContentType; + if (data.artifactSize !== undefined) + updates.artifactSize = data.artifactSize; + if (data.artifactExpiresAt !== undefined) + updates.artifactExpiresAt = data.artifactExpiresAt; updates.updatedAt = new Date(); await db().update(jobs).set(updates).where(eq(jobs.id, id)); @@ -360,28 +486,35 @@ export function createJobRepository( }, async requeueStaleInProgress(olderThan: Date): Promise { + const staleBaseCondition = and( + eq(jobs.status, "in_progress"), + lt(jobs.updatedAt, olderThan), + notInArray(jobs.type, [...batchParentJobTypes]), + ); + const cancelledRows = await db() + .update(jobs) + .set({ + status: "cancelled", + cancelledAt: new Date(), + finishedAt: new Date(), + updatedAt: new Date(), + }) + .where(and(staleBaseCondition, isNotNull(jobs.cancelRequestedAt))) + .returning(); + const rows = await db() .update(jobs) .set({ status: "pending", result: null, error: null, + startedAt: null, updatedAt: new Date(), }) - .where( - and( - eq(jobs.status, "in_progress"), - lt(jobs.updatedAt, olderThan), - notInArray(jobs.type, [ - "batch_ccip_parent", - "bulk_tagging_parent", - "thumbnail_generation_parent", - ]), - ), - ) + .where(and(staleBaseCondition, isNull(jobs.cancelRequestedAt))) .returning(); - return rows.length; + return rows.length + cancelledRows.length; }, }; } @@ -469,19 +602,25 @@ function buildSerializedClaimQuery( function buildClaimUpdate(now: Date) { return sql` UPDATE jobs - SET status = 'in_progress', updated_at = ${now} + SET status = 'in_progress', + attempt_count = attempt_count + 1, + started_at = ${now}, + finished_at = NULL, + cancel_requested_at = NULL, + cancelled_at = NULL, + updated_at = ${now} WHERE id IN (SELECT id FROM next_jobs) RETURNING id, type, source_id AS "mediaSourceId", - status, - payload, + status, + payload, result, error, created_at AS "createdAt", updated_at AS "updatedAt", - parent_id AS "parentId", + parent_id AS "parentId", cancel_requested_at AS "cancelRequestedAt", cancelled_at AS "cancelledAt", attempt_count AS "attemptCount", @@ -534,7 +673,7 @@ function mapClaimedJob(row: unknown): Job { parentId: nullableString(raw.parentId, "parentId"), cancelRequestedAt: nullableDate(raw.cancelRequestedAt, "cancelRequestedAt"), cancelledAt: nullableDate(raw.cancelledAt, "cancelledAt"), - attemptCount: requireNumber(raw.attemptCount, "attemptCount"), + attemptCount: nullableNumber(raw.attemptCount, "attemptCount") ?? 0, startedAt: nullableDate(raw.startedAt, "startedAt"), finishedAt: nullableDate(raw.finishedAt, "finishedAt"), artifactPath: nullableString(raw.artifactPath, "artifactPath"), diff --git a/packages/ui/src/hooks/use-batch-job-events.ts b/packages/ui/src/hooks/use-batch-job-events.ts index c21d1b0c..8af7164c 100644 --- a/packages/ui/src/hooks/use-batch-job-events.ts +++ b/packages/ui/src/hooks/use-batch-job-events.ts @@ -51,6 +51,8 @@ export function dispatchJobEvent( case "job-failed": handlers.handleJobFailed(event.data); break; + case "job-cancelled": + break; default: assertNever(event); }