fix: make download queue recovery durable

sake/drizzle/0023_queue_task_recovery.sql

@@ -0,0 +1,3 @@
+ALTER TABLE `QueueJobs` ADD `task_type` text;
+--> statement-breakpoint
+ALTER TABLE `QueueJobs` ADD `task_payload` text;

sake/drizzle/meta/_journal.json

@@ -162,6 +162,13 @@
       "when": 1782627929457,
       "tag": "0022_hardcover_progress_sync",
       "breakpoints": true
+    },
+    {
+      "idx": 23,
+      "version": "7",
+      "when": 1782627929458,
+      "tag": "0023_queue_task_recovery",
+      "breakpoints": true
     }
   ]
 }

sake/src/lib/server/application/composition/downloads.ts

@@ -1,5 +1,6 @@
 import { DavUploadServiceFactory } from '$lib/server/infrastructure/factories/DavUploadServiceFactory';
-import { downloadQueue } from '$lib/server/infrastructure/queue/downloadQueue';
+import { DownloadQueue } from '$lib/server/infrastructure/queue/downloadQueue';
+import { QueueJobRepository } from '$lib/server/infrastructure/repositories/QueueJobRepository';
 import { DownloadBookUseCase } from '$lib/server/application/use-cases/DownloadBookUseCase';
 import { QueueDownloadUseCase } from '$lib/server/application/use-cases/QueueDownloadUseCase';
 import { QueueSearchBookUseCase } from '$lib/server/application/use-cases/QueueSearchBookUseCase';
@@ -31,6 +32,7 @@ import {
 	zlibraryClient
 } from './foundation';
 import { externalBookMetadataService } from './providers';
+import { downloadSearchBookUseCase } from './search';
 
 export const downloadBookUseCase = new DownloadBookUseCase(
 	zlibraryClient,
@@ -41,12 +43,6 @@ export const downloadBookUseCase = new DownloadBookUseCase(
 	undefined,
 	externalBookMetadataService
 );
-export const queueDownloadUseCase = new QueueDownloadUseCase(downloadQueue);
-export const queueSearchBookUseCase = new QueueSearchBookUseCase(downloadQueue);
-export const getQueueStatusUseCase = new GetQueueStatusUseCase(
-	downloadQueue,
-	hardcoverProgressSyncJobRepository
-);
 export const getNewBooksForDeviceUseCase = new GetNewBooksForDeviceUseCase(bookRepository);
 export const confirmDownloadUseCase = new ConfirmDownloadUseCase(deviceDownloadRepository);
 export const removeDeviceDownloadUseCase = new RemoveDeviceDownloadUseCase(deviceDownloadRepository);
@@ -80,6 +76,18 @@ export const putLibraryFileUseCase = new PutLibraryFileUseCase(
 	undefined,
 	externalBookMetadataService
 );
+export const downloadQueue = new DownloadQueue(
+	new QueueJobRepository(),
+	downloadBookUseCase,
+	downloadSearchBookUseCase,
+	putLibraryFileUseCase
+);
+export const queueDownloadUseCase = new QueueDownloadUseCase(downloadQueue);
+export const queueSearchBookUseCase = new QueueSearchBookUseCase(downloadQueue);
+export const getQueueStatusUseCase = new GetQueueStatusUseCase(
+	downloadQueue,
+	hardcoverProgressSyncJobRepository
+);
 export const exportDeviceLibraryBookUseCase = new ExportDeviceLibraryBookUseCase(
 	bookRepository,
 	deviceDownloadRepository,

sake/src/lib/server/infrastructure/db/schema.ts

@@ -239,6 +239,8 @@ export const queueJobs = sqliteTable(
 		year: integer('year'),
 		userId: text('user_id').notNull(),
 		userKey: text('user_key').notNull(),
+		taskType: text('task_type', { enum: ['zlibrary', 'provider-import'] }),
+		taskPayload: text('task_payload'),
 		status: text('status', {
 			enum: ['queued', 'processing', 'completed', 'failed']
 		}).notNull(),

sake/src/lib/server/infrastructure/queue/downloadQueue.ts

@@ -1,25 +1,19 @@
 import { DownloadBookUseCase } from '$lib/server/application/use-cases/DownloadBookUseCase';
 import { DownloadSearchBookUseCase } from '$lib/server/application/use-cases/DownloadSearchBookUseCase';
 import { PutLibraryFileUseCase } from '$lib/server/application/use-cases/PutLibraryFileUseCase';
-import { ManagedBookCoverService } from '$lib/server/application/services/ManagedBookCoverService';
-import { ZLibraryClient } from '$lib/server/infrastructure/clients/ZLibraryClient';
 import { createChildLogger, toLogError } from '$lib/server/infrastructure/logging/logger';
 import { QueueJobRepository } from '$lib/server/infrastructure/repositories/QueueJobRepository';
 import type { QueueJobRecord } from '$lib/server/infrastructure/repositories/QueueJobRepository';
-import { BookRepository } from '$lib/server/infrastructure/repositories/BookRepository';
-import { S3Storage } from '$lib/server/infrastructure/storage/S3Storage';
-import { DavUploadServiceFactory } from '$lib/server/infrastructure/factories/DavUploadServiceFactory';
 import {
 	RECOVERY_REQUEUE_REQUIRED_ERROR,
-	PERSISTED_QUEUE_USER_KEY
+	parseQueueTask,
+	serializeQueueTask
 } from '$lib/server/infrastructure/queue/persistence';
-import { createLazySingleton } from '$lib/server/utils/createLazySingleton';
 import type {
+	DownloadQueueTaskInput,
 	SearchImportQueueTaskInput,
 	ZLibraryQueueTaskInput
 } from '$lib/server/application/ports/DownloadQueuePort';
-import { createSearchProviders } from '$lib/server/infrastructure/search-providers/searchProviderFactory';
-import { SEARCH_PROVIDER_IDS } from '$lib/types/Search/Provider';
 import { randomUUID } from 'node:crypto';
 
 interface BaseQueuedDownload {
@@ -78,48 +72,24 @@ export interface QueuedDownloadSnapshot {
 	finishedAt?: string;
 }
 
-class DownloadQueue {
-	private queue: QueuedDownload[] = [];
+export class DownloadQueue {
 	private isProcessing = false;
 	private readonly defaultMaxAttempts = 3;
 	private readonly terminalRetentionMs = 90 * 24 * 60 * 60 * 1000;
 	private readonly queueLogger = createChildLogger({ component: 'downloadQueue' });
-	private readonly queueJobRepository = new QueueJobRepository();
 	private isInitialized = false;
 	private initializePromise: Promise<void> | null = null;
-	private readonly storage = new S3Storage();
-	private readonly bookRepository = new BookRepository();
-	private readonly managedBookCoverService = new ManagedBookCoverService(this.storage);
-	private readonly zlibraryClient = new ZLibraryClient('https://1lib.sk');
-
-	private readonly downloadBookUseCase = new DownloadBookUseCase(
-		this.zlibraryClient,
-		this.bookRepository,
-		this.storage,
-		() => DavUploadServiceFactory.createS3(),
-		this.managedBookCoverService
-	);
-	private readonly downloadSearchBookUseCase = new DownloadSearchBookUseCase(
-		createSearchProviders([...SEARCH_PROVIDER_IDS], {
-			zlibrary: this.zlibraryClient
-		})
-	);
-	private readonly putLibraryFileUseCase = new PutLibraryFileUseCase(
-		this.storage,
-		this.bookRepository,
-		this.managedBookCoverService
-	);
+	private readonly volatileZLibraryCredentials = new Map<string, { userId: string; userKey: string }>();
+
+	constructor(
+		private readonly queueJobRepository: QueueJobRepository,
+		private readonly downloadBookUseCase: DownloadBookUseCase,
+		private readonly downloadSearchBookUseCase: DownloadSearchBookUseCase,
+		private readonly putLibraryFileUseCase: PutLibraryFileUseCase
+	) {}
 
 	async enqueue(
-		task:
-			| Omit<
-					QueuedZLibraryDownload,
-					'id' | 'status' | 'attempts' | 'maxAttempts' | 'createdAt' | 'updatedAt' | 'finishedAt'
-			  >
-			| Omit<
-					QueuedSearchImportDownload,
-					'id' | 'status' | 'attempts' | 'maxAttempts' | 'createdAt' | 'updatedAt' | 'finishedAt'
-			  >
+		task: DownloadQueueTaskInput
 	): Promise<string> {
 		await this.ensureInitialized();
 
@@ -136,7 +106,9 @@ class DownloadQueue {
 			updatedAt: now
 		};
 
-		this.queue.push(queuedTask);
+		if (task.source === 'zlibrary') {
+			this.volatileZLibraryCredentials.set(id, { userId: task.userId, userKey: task.userKey });
+		}
 		await this.queueJobRepository.create(this.toQueueJobRecord(queuedTask));
 		this.queueLogger.info(
 			{ event: 'queue.task.enqueued', taskId: id, bookId: task.bookId, title: task.title },
@@ -185,21 +157,17 @@ class DownloadQueue {
 	private async initialize(): Promise<void> {
 		try {
 			const nowIso = new Date().toISOString();
-			const failedRecoveredJobs = await this.queueJobRepository.failActiveJobsAfterRestart(
-				RECOVERY_REQUEUE_REQUIRED_ERROR,
-				nowIso,
-				nowIso
-			);
+			const failedRecoveredJobs = await this.queueJobRepository.recoverActiveJobsAfterRestart(nowIso, nowIso);
 			await this.cleanupOldTerminalJobs();
 			this.isInitialized = true;
 
 			if (failedRecoveredJobs > 0) {
 				this.queueLogger.warn(
 					{
-						event: 'queue.recovery.credentials_missing',
+						event: 'queue.recovery.non_resumable',
 						failedRecoveredJobs
 					},
-					'Marked queued jobs as failed because queue state cannot resume after restart'
+					'Marked queue jobs without resumable payloads as failed after restart'
 				);
 			}
 		} catch (error: unknown) {
@@ -216,13 +184,17 @@ class DownloadQueue {
 		this.isProcessing = true;
 		try {
 			while (true) {
-					const task = this.queue.find((candidate) => candidate.status === 'queued');
-					if (!task) {
-						break;
-					}
+				const record = await this.queueJobRepository.claimNextQueued();
+				if (!record) break;
+				const task = this.toQueuedDownload(record);
+				if (!task) {
+					const now = new Date().toISOString();
+					await this.queueJobRepository.updateFailed(record.id, record.attempts, RECOVERY_REQUEUE_REQUIRED_ERROR, now, now);
+					continue;
+				}
 
-					task.status = 'processing';
-					task.updatedAt = new Date();
+				task.status = 'processing';
+				task.updatedAt = new Date();
 					this.queueLogger.info(
 						{ event: 'queue.task.processing', taskId: task.id, bookId: task.bookId, title: task.title },
 						'Processing queue task'
@@ -270,9 +242,7 @@ class DownloadQueue {
 						},
 						'Queue task failed'
 					);
-					} finally {
-						this.queue = this.queue.filter((candidate) => candidate.id !== task.id);
-					}
+				}
 				}
 				await this.cleanupOldTerminalJobs();
 			} finally {
@@ -356,7 +326,7 @@ class DownloadQueue {
 				year: task.year ?? undefined,
 				downloadToDevice: false
 			},
-			credentials: {
+			credentials: this.volatileZLibraryCredentials.get(task.id) ?? {
 				userId: task.userId,
 				userKey: task.userKey
 			}
@@ -458,7 +428,9 @@ class DownloadQueue {
 			language: task.language,
 			year: task.year,
 			userId: task.userId,
-			userKey: PERSISTED_QUEUE_USER_KEY,
+			userKey: '',
+			taskType: task.source,
+			taskPayload: serializeQueueTask(task),
 			status: task.status,
 			attempts: task.attempts,
 			maxAttempts: task.maxAttempts,
@@ -469,11 +441,40 @@ class DownloadQueue {
 		};
 	}
 
+	private toQueuedDownload(record: QueueJobRecord): QueuedDownload | null {
+		const task = parseQueueTask<DownloadQueueTaskInput>(record.taskPayload);
+		if (!task || task.source !== record.taskType || task.bookId !== record.bookId) {
+			return null;
+		}
+
+		if (task.source === 'zlibrary') {
+			const credentials = this.volatileZLibraryCredentials.get(record.id);
+			if (!credentials) return null;
+			return {
+				...task,
+				...record,
+				source: 'zlibrary',
+				...credentials,
+				createdAt: new Date(record.createdAt),
+				updatedAt: new Date(record.updatedAt),
+				finishedAt: record.finishedAt ? new Date(record.finishedAt) : undefined
+			};
+		}
+
+		return {
+			...task,
+			...record,
+			source: 'provider-import',
+			userKey: '',
+			createdAt: new Date(record.createdAt),
+			updatedAt: new Date(record.updatedAt),
+			finishedAt: record.finishedAt ? new Date(record.finishedAt) : undefined
+		};
+	}
+
 	private toArrayBuffer(data: Uint8Array): ArrayBuffer {
 		const copy = new Uint8Array(data.byteLength);
 		copy.set(data);
 		return copy.buffer;
 	}
 }
-
-export const downloadQueue = createLazySingleton(() => new DownloadQueue());

sake/src/lib/server/infrastructure/queue/persistence.ts

@@ -1,11 +1,54 @@
 export const PERSISTED_QUEUE_USER_KEY = '';
 
+export const QUEUE_TASK_PAYLOAD_VERSION = 1;
+
 export const RECOVERY_REQUEUE_REQUIRED_ERROR =
-	'Queued job could not resume after restart because transient queue data is not persisted. Requeue the download.';
+	'Queued job could not resume after restart because its task payload is missing or invalid. Requeue the download.';
+
+export const RECOVERY_ZLIBRARY_CREDENTIALS_MISSING_ERROR =
+	'Z-Library queue job could not resume after restart because its credentials are intentionally not persisted. Requeue the download.';
 
 export function sanitizePersistedQueueJob<T extends { userKey: string }>(job: T): T {
 	return {
 		...job,
 		userKey: PERSISTED_QUEUE_USER_KEY
 	};
 }
+
+export interface PersistedQueueTaskPayload<T> {
+	version: typeof QUEUE_TASK_PAYLOAD_VERSION;
+	task: T;
+}
+
+export function serializeQueueTask<T extends { userKey: string }>(task: T): string {
+	const { userKey: _userKey, ...safeTask } = task;
+	const payload: PersistedQueueTaskPayload<Omit<T, 'userKey'>> = {
+		version: QUEUE_TASK_PAYLOAD_VERSION,
+		task: safeTask
+	};
+	return JSON.stringify(payload);
+}
+
+export function parseQueueTask<T>(payload: string | null | undefined): T | null {
+	if (!payload) {
+		return null;
+	}
+
+	try {
+		const parsed: unknown = JSON.parse(payload);
+		if (
+			typeof parsed !== 'object' ||
+			parsed === null ||
+			!('version' in parsed) ||
+			parsed.version !== QUEUE_TASK_PAYLOAD_VERSION ||
+			!('task' in parsed) ||
+			typeof parsed.task !== 'object' ||
+			parsed.task === null
+		) {
+			return null;
+		}
+		return parsed.task as T;
+	} catch {
+		return null;
+	}
+}