From 73d66b75e3fb07fda15fcd4e1f517613cb43f155 Mon Sep 17 00:00:00 2001 From: neon <18559658+neonsolstice@users.noreply.github.com> Date: Mon, 19 Jan 2026 16:48:11 -0600 Subject: [PATCH] feat(scanner): add live scan progress via WebSocket and auto-scan --- client/package.json | 13 +- .../scanner/composables/useScanProgress.ts | 64 +++++ .../features/settings/LibrariesSettings.vue | 93 +++++++- client/vite.config.ts | 5 + packages/types/src/index.ts | 1 + packages/types/src/scanner.ts | 11 + server/package.json | 12 +- server/src/db/migrations/meta/_journal.json | 4 +- server/src/main.ts | 3 + server/src/modules/library/library.module.ts | 2 + server/src/modules/library/library.service.ts | 4 + .../modules/scanner/scan-job-store.service.ts | 74 ++++++ server/src/modules/scanner/scan.gateway.ts | 70 ++++++ .../src/modules/scanner/scanner.controller.ts | 5 +- server/src/modules/scanner/scanner.module.ts | 19 +- .../src/modules/scanner/scanner.repository.ts | 9 +- server/src/modules/scanner/scanner.service.ts | 225 ++++++++++++++---- 17 files changed, 541 insertions(+), 73 deletions(-) create mode 100644 client/src/features/scanner/composables/useScanProgress.ts create mode 100644 packages/types/src/scanner.ts create mode 100644 server/src/modules/scanner/scan-job-store.service.ts create mode 100644 server/src/modules/scanner/scan.gateway.ts diff --git a/client/package.json b/client/package.json index 6388be8ac..5264fc226 100644 --- a/client/package.json +++ b/client/package.json @@ -27,17 +27,18 @@ "pdfjs-dist": "^5.4.530", "pinia": "^3.0.4", "reka-ui": "^2.7.0", + "socket.io-client": "^4.8.3", "tailwind-merge": "^3.4.0", "tailwindcss": "^4.1.18", - "vue": "^3.5.26", + "vue": "^3.5.27", "vue-router": "^4.6.4", "vue-sonner": "^2.0.9", "vue-virtual-scroller": "^2.0.0-beta.8" }, "devDependencies": { - "@tsconfig/node24": "^24.0.3", + "@tsconfig/node24": "^24.0.4", "@types/jsdom": "^27.0.0", - "@types/node": "^22.19.6", + "@types/node": "^22.19.7", "@vitejs/plugin-vue": "^6.0.3", "@vitest/eslint-plugin": "^1.6.6", "@vue/eslint-config-typescript": "^14.6.0", @@ -45,12 +46,12 @@ "@vue/tsconfig": "^0.8.1", "eslint": "^9.39.2", "eslint-config-prettier": "^10.1.8", - "eslint-plugin-oxlint": "~1.39.0", - "eslint-plugin-vue": "~10.6.2", + "eslint-plugin-oxlint": "~1.41.0", + "eslint-plugin-vue": "~10.7.0", "jiti": "^2.6.1", "jsdom": "^27.4.0", "npm-run-all2": "^8.0.4", - "oxlint": "~1.39.0", + "oxlint": "~1.41.0", "prettier": "3.8.0", "tw-animate-css": "^1.4.0", "typescript": "~5.9.3", diff --git a/client/src/features/scanner/composables/useScanProgress.ts b/client/src/features/scanner/composables/useScanProgress.ts new file mode 100644 index 000000000..6b26ae3cf --- /dev/null +++ b/client/src/features/scanner/composables/useScanProgress.ts @@ -0,0 +1,64 @@ +import { ref } from 'vue' +import { io, Socket } from 'socket.io-client' +import type { ScanProgressEvent } from '@projectx/types' +import { getAccessToken } from '@/lib/api' + +let socket: Socket | null = null +const progressMap = ref>(new Map()) +const subscribedLibraries = new Set() + +function getSocket(): Socket { + if (!socket) { + // Use a callback so getAccessToken() is called at (re)connect time, not just once at creation. + socket = io('/scan', { + auth: (cb: (data: object) => void) => cb({ token: getAccessToken() }), + transports: ['websocket'], + autoConnect: true, + }) + + socket.on('scan:progress', (event: ScanProgressEvent) => { + progressMap.value.set(event.libraryId, event) + progressMap.value = new Map(progressMap.value) + if (event.status !== 'running') { + // Keep the final event for a short time so UI can show completion, then clear. + setTimeout(() => { + progressMap.value.delete(event.libraryId) + progressMap.value = new Map(progressMap.value) + }, 3000) + } + }) + + socket.on('disconnect', () => { + // Clear stale progress bars but keep subscribedLibraries so we can re-subscribe on reconnect. + progressMap.value = new Map() + }) + + socket.on('connect', () => { + // Re-subscribe to all libraries after reconnect (server-side rooms are per-socket). + for (const id of subscribedLibraries) { + socket!.emit('subscribe:library', id) + } + }) + } + return socket +} + +export { getSocket } + +export function useScanProgress() { + function subscribeLibrary(libraryId: number): void { + if (subscribedLibraries.has(libraryId)) return + subscribedLibraries.add(libraryId) + getSocket().emit('subscribe:library', libraryId) + } + + function getProgress(libraryId: number): ScanProgressEvent | undefined { + return progressMap.value.get(libraryId) + } + + function isScanning(libraryId: number): boolean { + return progressMap.value.get(libraryId)?.status === 'running' + } + + return { subscribeLibrary, getProgress, isScanning, progressMap } +} diff --git a/client/src/features/settings/LibrariesSettings.vue b/client/src/features/settings/LibrariesSettings.vue index 81ad247ad..0800a8c3f 100644 --- a/client/src/features/settings/LibrariesSettings.vue +++ b/client/src/features/settings/LibrariesSettings.vue @@ -1,5 +1,5 @@ @@ -150,7 +192,7 @@ function iconComponent(iconName: string | null | undefined) { @click="scanAll" > - {{ scanningAll ? 'Scanning…' : 'Scan All' }} + {{ scanningAll ? 'Scanning...' : 'Scan All' }} diff --git a/client/vite.config.ts b/client/vite.config.ts index e63975181..160b9f85e 100644 --- a/client/vite.config.ts +++ b/client/vite.config.ts @@ -16,6 +16,11 @@ export default defineConfig({ server: { proxy: { '/api': 'http://localhost:3000', + '/socket.io': { + target: 'http://localhost:3000', + ws: true, + changeOrigin: true, + }, }, }, }) diff --git a/packages/types/src/index.ts b/packages/types/src/index.ts index 8c14f6717..e8e3f9120 100644 --- a/packages/types/src/index.ts +++ b/packages/types/src/index.ts @@ -1,3 +1,4 @@ export * from "./auth"; export * from "./library"; export * from "./reader-settings"; +export * from "./scanner"; diff --git a/packages/types/src/scanner.ts b/packages/types/src/scanner.ts new file mode 100644 index 000000000..03fb0f036 --- /dev/null +++ b/packages/types/src/scanner.ts @@ -0,0 +1,11 @@ +export interface ScanProgressEvent { + jobId: number; + libraryId: number; + status: 'running' | 'completed' | 'failed'; + processed: number; + total: number; + added: number; + updated: number; + missing: number; + errorMessage?: string; +} diff --git a/server/package.json b/server/package.json index 3f774b122..53385aca8 100644 --- a/server/package.json +++ b/server/package.json @@ -24,6 +24,7 @@ "test:e2e": "jest --config ./test/jest-e2e.json" }, "dependencies": { + "@projectx/types": "workspace:*", "7z-wasm": "^1.2.0", "@fastify/cookie": "^11.0.2", "@fastify/formbody": "^8.0.2", @@ -34,8 +35,10 @@ "@nestjs/core": "^11.0.1", "@nestjs/jwt": "^11.0.2", "@nestjs/passport": "^11.0.5", - "@nestjs/platform-fastify": "^11.1.11", + "@nestjs/platform-fastify": "^11.1.12", + "@nestjs/platform-socket.io": "^11.1.12", "@nestjs/schedule": "^6.1.0", + "@nestjs/websockets": "^11.1.12", "bcryptjs": "^3.0.3", "chokidar": "^4.0.3", "class-transformer": "^0.5.1", @@ -48,10 +51,11 @@ "passport": "^0.7.0", "passport-jwt": "^4.0.1", "pdf-lib": "^1.17.1", - "pg": "^8.16.3", + "pg": "^8.17.1", "reflect-metadata": "^0.2.2", "rxjs": "^7.8.1", "sharp": "^0.34.5", + "socket.io": "^4.8.3", "unzipper": "^0.12.3" }, "devDependencies": { @@ -60,8 +64,8 @@ "@nestjs/cli": "^11.0.0", "@nestjs/schematics": "^11.0.0", "@nestjs/testing": "^11.0.1", - "@swc/cli": "^0.7.9", - "@swc/core": "^1.15.8", + "@swc/cli": "^0.7.10", + "@swc/core": "^1.15.10", "@types/bcrypt": "^6.0.0", "@types/bcryptjs": "^3.0.0", "@types/chokidar": "^1.7.5", diff --git a/server/src/db/migrations/meta/_journal.json b/server/src/db/migrations/meta/_journal.json index bdf814fc3..a0f24076f 100644 --- a/server/src/db/migrations/meta/_journal.json +++ b/server/src/db/migrations/meta/_journal.json @@ -5,14 +5,14 @@ { "idx": 0, "version": "7", - "when": 1768599599999, + "when": 1768909427779, "tag": "0000_baseline", "breakpoints": true }, { "idx": 1, "version": "7", - "when": 1768599599999, + "when": 1768909427779, "tag": "0001_add_scan_jobs_cascade", "breakpoints": true } diff --git a/server/src/main.ts b/server/src/main.ts index 3650ec676..0900594f0 100644 --- a/server/src/main.ts +++ b/server/src/main.ts @@ -1,6 +1,7 @@ import { NestFactory } from '@nestjs/core'; import { FastifyAdapter, NestFastifyApplication } from '@nestjs/platform-fastify'; import { ValidationPipe } from '@nestjs/common'; +import { IoAdapter } from '@nestjs/platform-socket.io'; import { AppModule } from './app.module'; import { GlobalExceptionFilter } from './common/filters/http-exception.filter'; import { join } from 'path'; @@ -14,6 +15,8 @@ async function bootstrap() { await pool.end(); const app = await NestFactory.create(AppModule, new FastifyAdapter({ logger: true })); + app.useWebSocketAdapter(new IoAdapter(app)); + app.setGlobalPrefix('api'); app.useGlobalPipes( diff --git a/server/src/modules/library/library.module.ts b/server/src/modules/library/library.module.ts index 47868c10d..8284cbc97 100644 --- a/server/src/modules/library/library.module.ts +++ b/server/src/modules/library/library.module.ts @@ -1,10 +1,12 @@ import { Module } from '@nestjs/common'; +import { ScannerModule } from '../scanner/scanner.module'; import { LibraryController } from './library.controller'; import { LibraryRepository } from './library.repository'; import { LibraryService } from './library.service'; @Module({ + imports: [ScannerModule], controllers: [LibraryController], providers: [LibraryService, LibraryRepository], exports: [LibraryService], diff --git a/server/src/modules/library/library.service.ts b/server/src/modules/library/library.service.ts index 69fcb618d..f40265563 100644 --- a/server/src/modules/library/library.service.ts +++ b/server/src/modules/library/library.service.ts @@ -5,6 +5,7 @@ import { join } from 'path'; import type { RequestUser } from '../../common/types/request-user'; import { isPrimaryFormat } from '../scanner/lib/classify'; +import { ScannerService } from '../scanner/scanner.service'; import { CreateLibraryDto } from './dto/create-library.dto'; import { GrantLibraryAccessDto } from './dto/grant-library-access.dto'; import { PrescanLibraryDto } from './dto/prescan-library.dto'; @@ -20,6 +21,7 @@ export class LibraryService { constructor( private readonly libraryRepo: LibraryRepository, private readonly config: ConfigService, + private readonly scannerService: ScannerService, ) { this.booksPath = this.config.get('storage.booksPath')!; } @@ -62,6 +64,8 @@ export class LibraryService { const folders = await Promise.all(dto.folders.map((path) => this.libraryRepo.insertFolder({ libraryId: library.id, path }))); + this.scannerService.startScanAsync(library.id); + return { ...library, folders: folders.map(([f]) => f) }; } diff --git a/server/src/modules/scanner/scan-job-store.service.ts b/server/src/modules/scanner/scan-job-store.service.ts new file mode 100644 index 000000000..0c825bfff --- /dev/null +++ b/server/src/modules/scanner/scan-job-store.service.ts @@ -0,0 +1,74 @@ +import { Injectable } from '@nestjs/common'; + +export interface ScanEntry { + jobId: number; + libraryId: number; + processed: number; + total: number; + added: number; + updated: number; + missing: number; + lastEmitMs: number; + lastEmitPct: number; +} + +@Injectable() +export class ScanJobStore { + private readonly byLibrary = new Map(); + + isRunning(libraryId: number): boolean { + return this.byLibrary.has(libraryId); + } + + create(jobId: number, libraryId: number, total: number): ScanEntry { + const entry: ScanEntry = { + jobId, + libraryId, + processed: 0, + total, + added: 0, + updated: 0, + missing: 0, + lastEmitMs: 0, + lastEmitPct: -1, + }; + this.byLibrary.set(libraryId, entry); + return entry; + } + + setTotal(libraryId: number, total: number): void { + const entry = this.byLibrary.get(libraryId); + if (entry) entry.total = total; + } + + increment(libraryId: number, delta: { processed?: number; added?: number; updated?: number; missing?: number }): ScanEntry | undefined { + const entry = this.byLibrary.get(libraryId); + if (!entry) return undefined; + if (delta.processed !== undefined) entry.processed += delta.processed; + if (delta.added !== undefined) entry.added += delta.added; + if (delta.updated !== undefined) entry.updated += delta.updated; + if (delta.missing !== undefined) entry.missing += delta.missing; + return entry; + } + + get(libraryId: number): ScanEntry | undefined { + return this.byLibrary.get(libraryId); + } + + shouldEmit(entry: ScanEntry): boolean { + if (entry.total > 0 && entry.processed >= entry.total) return true; + const pct = entry.total > 0 ? Math.floor((entry.processed / entry.total) * 100) : 0; + if (pct >= entry.lastEmitPct + 1) return true; + if (Date.now() - entry.lastEmitMs >= 1000) return true; + return false; + } + + markEmitted(entry: ScanEntry): void { + entry.lastEmitMs = Date.now(); + entry.lastEmitPct = entry.total > 0 ? Math.floor((entry.processed / entry.total) * 100) : 0; + } + + delete(libraryId: number): void { + this.byLibrary.delete(libraryId); + } +} diff --git a/server/src/modules/scanner/scan.gateway.ts b/server/src/modules/scanner/scan.gateway.ts new file mode 100644 index 000000000..20f8f80a0 --- /dev/null +++ b/server/src/modules/scanner/scan.gateway.ts @@ -0,0 +1,70 @@ +import { Logger } from '@nestjs/common'; +import { JwtService } from '@nestjs/jwt'; +import { + OnGatewayConnection, + OnGatewayDisconnect, + SubscribeMessage, + WebSocketGateway, + WebSocketServer, +} from '@nestjs/websockets'; +import { Server, Socket } from 'socket.io'; + +import type { ScanProgressEvent } from '@projectx/types'; +import { AuthService } from '../auth/auth.service'; +import { ScanJobStore } from './scan-job-store.service'; + +@WebSocketGateway({ namespace: '/scan', cors: { origin: '*', credentials: true } }) +export class ScanGateway implements OnGatewayConnection, OnGatewayDisconnect { + @WebSocketServer() server: Server; + private readonly logger = new Logger(ScanGateway.name); + + constructor( + private readonly jwtService: JwtService, + private readonly authService: AuthService, + private readonly scanJobStore: ScanJobStore, + ) {} + + async handleConnection(client: Socket): Promise { + try { + const token = client.handshake.auth?.token as string | undefined; + if (!token) throw new Error('No token provided'); + const payload = this.jwtService.verify<{ sub: number; ver: number }>(token); + const user = await this.authService.validateUser(payload.sub, payload.ver); + if (!user) throw new Error('User not found or token revoked'); + client.data.user = user; + this.logger.debug(`WS connected: user=${user.id} socket=${client.id}`); + } catch (err) { + this.logger.warn(`WS rejected: ${(err as Error).message} socket=${client.id}`); + client.disconnect(); + } + } + + handleDisconnect(client: Socket): void { + this.logger.debug(`WS disconnected: socket=${client.id}`); + } + + @SubscribeMessage('subscribe:library') + handleSubscribeLibrary(client: Socket, libraryId: number): void { + client.join(`library:${libraryId}`); + // Send current snapshot immediately so reconnecting clients catch up. + const entry = this.scanJobStore.get(libraryId); + if (entry) { + const event: ScanProgressEvent = { + jobId: entry.jobId, + libraryId, + status: 'running', + processed: entry.processed, + total: entry.total, + added: entry.added, + updated: entry.updated, + missing: entry.missing, + }; + client.emit('scan:progress', event); + } + } + + // Called by ScannerService — not a WS message handler. + emitProgress(event: ScanProgressEvent): void { + this.server.to(`library:${event.libraryId}`).emit('scan:progress', event); + } +} diff --git a/server/src/modules/scanner/scanner.controller.ts b/server/src/modules/scanner/scanner.controller.ts index 5f86651ff..baf25f163 100644 --- a/server/src/modules/scanner/scanner.controller.ts +++ b/server/src/modules/scanner/scanner.controller.ts @@ -1,4 +1,4 @@ -import { Controller, Param, ParseIntPipe, Post } from '@nestjs/common'; +import { Controller, HttpCode, HttpStatus, Param, ParseIntPipe, Post } from '@nestjs/common'; import { RequirePermission } from '../../common/decorators/require-permission.decorator'; import { ScannerService } from './scanner.service'; @@ -9,7 +9,8 @@ export class ScannerController { @Post('libraries/:id/scan') @RequirePermission('manage_libraries') + @HttpCode(HttpStatus.ACCEPTED) scan(@Param('id', ParseIntPipe) libraryId: number) { - return this.scannerService.scan(libraryId, 'manual'); + return this.scannerService.startScan(libraryId, 'manual'); } } diff --git a/server/src/modules/scanner/scanner.module.ts b/server/src/modules/scanner/scanner.module.ts index 9b537cb6e..4cc309371 100644 --- a/server/src/modules/scanner/scanner.module.ts +++ b/server/src/modules/scanner/scanner.module.ts @@ -1,14 +1,29 @@ import { Module } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { JwtModule } from '@nestjs/jwt'; +import { AuthModule } from '../auth/auth.module'; import { MetadataModule } from '../metadata/metadata.module'; +import { ScanGateway } from './scan.gateway'; +import { ScanJobStore } from './scan-job-store.service'; import { ScannerController } from './scanner.controller'; import { ScannerRepository } from './scanner.repository'; import { ScannerService } from './scanner.service'; @Module({ - imports: [MetadataModule], + imports: [ + MetadataModule, + AuthModule, + JwtModule.registerAsync({ + inject: [ConfigService], + useFactory: (config: ConfigService) => ({ + secret: config.get('auth.jwtSecret'), + signOptions: { expiresIn: config.get('auth.jwtExpiresIn') as any }, + }), + }), + ], controllers: [ScannerController], - providers: [ScannerService, ScannerRepository], + providers: [ScannerService, ScannerRepository, ScanGateway, ScanJobStore], exports: [ScannerService], }) export class ScannerModule {} diff --git a/server/src/modules/scanner/scanner.repository.ts b/server/src/modules/scanner/scanner.repository.ts index 7bc7cade6..d0c468287 100644 --- a/server/src/modules/scanner/scanner.repository.ts +++ b/server/src/modules/scanner/scanner.repository.ts @@ -1,5 +1,5 @@ import { Inject, Injectable } from '@nestjs/common'; -import { and, eq, inArray } from 'drizzle-orm'; +import { and, eq, inArray, ne } from 'drizzle-orm'; import { NodePgDatabase } from 'drizzle-orm/node-postgres'; import { DB } from '../../db'; @@ -30,6 +30,13 @@ export class ScannerRepository { await this.db.update(scanJobs).set({ status: 'failed', errorMessage, completedAt: new Date() }).where(eq(scanJobs.id, id)); } + async failAllRunningJobs(errorMessage: string): Promise { + await this.db + .update(scanJobs) + .set({ status: 'failed', errorMessage, completedAt: new Date() }) + .where(eq(scanJobs.status, 'running')); + } + // ── Library Folders ──────────────────────────────────────────────────────── async findLibraryFolders(libraryId: number) { diff --git a/server/src/modules/scanner/scanner.service.ts b/server/src/modules/scanner/scanner.service.ts index 1f5a84ed6..ab7244fa9 100644 --- a/server/src/modules/scanner/scanner.service.ts +++ b/server/src/modules/scanner/scanner.service.ts @@ -1,14 +1,18 @@ -import { Injectable, Logger, NotFoundException } from '@nestjs/common'; - -const METADATA_FORMATS = new Set(['epub', 'mobi', 'azw3', 'azw', 'cbz', 'cbr', 'cb7', 'pdf']); +import { ConflictException, Injectable, Logger, NotFoundException, OnApplicationBootstrap } from '@nestjs/common'; +import type { ScanProgressEvent } from '@projectx/types'; import { MetadataService } from '../metadata/metadata.service'; +import { ScanGateway } from './scan.gateway'; +import { ScanJobStore } from './scan-job-store.service'; import { classifyFile } from './lib/classify'; import { sha256File } from './lib/hash'; import { waitForStability } from './lib/stability'; import { BookCandidate, FileStat, findBookCandidates } from './lib/walk'; import { ScannerRepository } from './scanner.repository'; +const METADATA_FORMATS = new Set(['epub', 'mobi', 'azw3', 'azw', 'cbz', 'cbr', 'cb7', 'pdf']); +const BATCH_SIZE = 5; + interface ScanCounts { addedCount: number; updatedCount: number; @@ -16,86 +20,207 @@ interface ScanCounts { } @Injectable() -export class ScannerService { +export class ScannerService implements OnApplicationBootstrap { private readonly logger = new Logger(ScannerService.name); constructor( private readonly scannerRepo: ScannerRepository, private readonly metadataService: MetadataService, + private readonly scanJobStore: ScanJobStore, + private readonly scanGateway: ScanGateway, ) {} - async scan(libraryId: number, triggeredBy: 'manual' | 'watcher' | 'schedule' = 'manual') { + async onApplicationBootstrap(): Promise { + await this.scannerRepo.failAllRunningJobs('Server restarted during scan'); + } + + async startScan(libraryId: number, triggeredBy: 'manual' | 'watcher' | 'schedule'): Promise<{ jobId: number }> { + if (this.scanJobStore.isRunning(libraryId)) { + throw new ConflictException(`A scan is already running for library ${libraryId}`); + } + const folders = await this.scannerRepo.findLibraryFolders(libraryId); if (folders.length === 0) throw new NotFoundException(`Library ${libraryId} has no folders`); const job = await this.scannerRepo.createScanJob(libraryId, triggeredBy); - this.logger.log(`Scan job ${job.id} started for library ${libraryId}`); + + this.scanJobStore.create(job.id, libraryId, 0); + this.emitFromStore(libraryId, job.id, 'running'); + + this.runScan(libraryId, job.id, folders).catch((err) => + this.logger.error(`Scan job ${job.id} crashed unexpectedly: ${(err as Error).message}`), + ); + + return { jobId: job.id }; + } + + startScanAsync(libraryId: number): void { + if (this.scanJobStore.isRunning(libraryId)) return; + this.startScan(libraryId, 'manual').catch((err) => + this.logger.error(`Auto-scan failed to start for library ${libraryId}: ${(err as Error).message}`), + ); + } + + private async runScan( + libraryId: number, + jobId: number, + folders: Awaited>, + ): Promise { + this.logger.log(`Scan job ${jobId} started for library ${libraryId}`); + + type FolderWork = { + id: number; + libraryId: number; + path: string; + candidates: BookCandidate[]; + knownBooks: Awaited>; + knownFiles: Awaited>; + }; + + const folderWork: FolderWork[] = []; + let totalCandidates = 0; + + for (const folder of folders) { + let candidates: BookCandidate[] = []; + try { + candidates = await findBookCandidates(folder.path); + } catch (err) { + this.logger.warn(`Cannot walk ${folder.path}: ${err}`); + } + + const [knownBooks, knownFiles] = await Promise.all([ + this.scannerRepo.findBooksByLibraryFolder(folder.id), + this.scannerRepo.findBookFilesByLibraryFolder(folder.id), + ]); + + folderWork.push({ ...folder, candidates, knownBooks, knownFiles }); + totalCandidates += candidates.length; + } + + this.scanJobStore.setTotal(libraryId, totalCandidates); + this.emitFromStore(libraryId, jobId, 'running'); const totals: ScanCounts = { addedCount: 0, updatedCount: 0, missingCount: 0 }; try { - for (const folder of folders) { - this.logger.log(`Scanning folder: ${folder.path}`); - const counts = await this.scanFolder(folder.id, folder.libraryId, folder.path); + for (const { id: folderId, path: folderPath, candidates, knownBooks, knownFiles } of folderWork) { + const counts = await this.scanFolderCandidates( + folderId, + libraryId, + folderPath, + candidates, + knownBooks, + knownFiles, + jobId, + ); totals.addedCount += counts.addedCount; totals.updatedCount += counts.updatedCount; totals.missingCount += counts.missingCount; } - await this.scannerRepo.completeScanJob(job.id, totals); - this.logger.log(`Scan job ${job.id} completed — ${JSON.stringify(totals)}`); + await this.scannerRepo.completeScanJob(jobId, totals); + this.logger.log(`Scan job ${jobId} completed — ${JSON.stringify(totals)}`); + this.emitFromStore(libraryId, jobId, 'completed'); } catch (err) { const message = err instanceof Error ? err.message : String(err); - await this.scannerRepo.failScanJob(job.id, message); - this.logger.error(`Scan job ${job.id} failed: ${message}`); - throw err; + await this.scannerRepo.failScanJob(jobId, message).catch(() => { + // Job row may have been cascade-deleted if library was deleted. + }); + this.logger.error(`Scan job ${jobId} failed: ${message}`); + this.emitFromStore(libraryId, jobId, 'failed', message); + } finally { + this.scanJobStore.delete(libraryId); } - - return { jobId: job.id, ...totals }; } - private async scanFolder(libraryFolderId: number, libraryId: number, folderPath: string): Promise { + private async scanFolderCandidates( + libraryFolderId: number, + libraryId: number, + folderPath: string, + candidates: BookCandidate[], + knownBooks: Awaited>, + knownFiles: Awaited>, + jobId: number, + ): Promise { const counts: ScanCounts = { addedCount: 0, updatedCount: 0, missingCount: 0 }; - const [knownBooks, knownFiles] = await Promise.all([ - this.scannerRepo.findBooksByLibraryFolder(libraryFolderId), - this.scannerRepo.findBookFilesByLibraryFolder(libraryFolderId), - ]); - const bookByFolderPath = new Map(knownBooks.map((b) => [b.folderPath, b])); const fileByPath = new Map(knownFiles.map((f) => [f.absolutePath, f])); const fileByIno = new Map(knownFiles.map((f) => [f.ino, f])); const seenBookIds = new Set(); - let candidates: BookCandidate[]; - try { - candidates = await findBookCandidates(folderPath); - } catch (err) { - this.logger.warn(`Cannot walk ${folderPath}: ${err}`); - return counts; - } + for (let i = 0; i < candidates.length; i += BATCH_SIZE) { + const batch = candidates.slice(i, i + BATCH_SIZE); - for (const candidate of candidates) { - const book = await this.upsertBook(candidate, libraryId, libraryFolderId, bookByFolderPath, counts); - seenBookIds.add(book.id); + const results = await Promise.all( + batch.map((c) => this.processCandidate(c, libraryId, libraryFolderId, bookByFolderPath, fileByPath, fileByIno)), + ); - for (const fileStat of candidate.files) { - const { format } = classifyFile(fileStat.absolutePath); - const isNew = await this.processFile(fileStat, book.id, libraryFolderId, fileByPath, fileByIno, counts); - if (isNew && format && METADATA_FORMATS.has(format)) { - await this.metadataService.extractAndSave(book.id, fileStat.absolutePath, format); - } + for (const r of results) { + seenBookIds.add(r.bookId); + counts.addedCount += r.added; + counts.updatedCount += r.updated; + } + + const entry = this.scanJobStore.increment(libraryId, { processed: batch.length }); + if (entry && this.scanJobStore.shouldEmit(entry)) { + this.emitFromStore(libraryId, jobId, 'running'); + this.scanJobStore.markEmitted(entry); } } const missingIds = knownBooks.filter((b) => !seenBookIds.has(b.id)).map((b) => b.id); - - await this.scannerRepo.markBooksAsMissing(missingIds); - counts.missingCount += missingIds.length; + if (missingIds.length > 0) { + await this.scannerRepo.markBooksAsMissing(missingIds); + counts.missingCount += missingIds.length; + this.scanJobStore.increment(libraryId, { missing: missingIds.length }); + } return counts; } + private async processCandidate( + candidate: BookCandidate, + libraryId: number, + libraryFolderId: number, + bookByFolderPath: Map, + fileByPath: Map, + fileByIno: Map, + ): Promise<{ bookId: number; added: number; updated: number }> { + const counts = { added: 0, updated: 0 }; + const fileCounts: ScanCounts = { addedCount: 0, updatedCount: 0, missingCount: 0 }; + + const book = await this.upsertBook(candidate, libraryId, libraryFolderId, bookByFolderPath, fileCounts); + counts.added += fileCounts.addedCount; + counts.updated += fileCounts.updatedCount; + + for (const fileStat of candidate.files) { + const { format } = classifyFile(fileStat.absolutePath); + const fileCount: ScanCounts = { addedCount: 0, updatedCount: 0, missingCount: 0 }; + let isNew: boolean; + + try { + isNew = await this.processFile(fileStat, book.id, libraryFolderId, fileByPath, fileByIno, fileCount); + } catch (err) { + this.logger.warn(`Failed to process file ${fileStat.absolutePath}: ${err}`); + continue; + } + + counts.added += fileCount.addedCount; + counts.updated += fileCount.updatedCount; + + if (isNew && format && METADATA_FORMATS.has(format)) { + try { + await this.metadataService.extractAndSave(book.id, fileStat.absolutePath, format); + } catch (err) { + this.logger.warn(`Metadata extraction failed for ${fileStat.absolutePath}: ${err}`); + } + } + } + + return { bookId: book.id, ...counts }; + } + private async upsertBook( candidate: BookCandidate, libraryId: number, @@ -124,7 +249,6 @@ export class ScannerService { return existing; } - // Returns true if a new file record was created (triggers metadata extraction). private async processFile( fileStat: FileStat, bookId: number, @@ -141,7 +265,6 @@ export class ScannerService { const byPath = fileByPath.get(fileStat.absolutePath); if (byPath) { const changed = fileStat.sizeBytes !== byPath.sizeBytes || fileStat.mtime.getTime() !== byPath.mtime?.getTime(); - if (changed) { await this.scannerRepo.updateBookFile(byPath.id, { ino: fileStat.ino, @@ -203,4 +326,20 @@ export class ScannerService { counts.addedCount++; return true; } + + private emitFromStore(libraryId: number, jobId: number, status: 'running' | 'completed' | 'failed', errorMessage?: string): void { + const entry = this.scanJobStore.get(libraryId); + const event: ScanProgressEvent = { + jobId, + libraryId, + status, + processed: entry?.processed ?? 0, + total: entry?.total ?? 0, + added: entry?.added ?? 0, + updated: entry?.updated ?? 0, + missing: entry?.missing ?? 0, + errorMessage, + }; + this.scanGateway.emitProgress(event); + } }