mirror of
https://github.com/bookorbit/bookorbit.git
synced 2026-11-03 08:21:39 -06:00
feat(scanner): add live scan progress via WebSocket and auto-scan
This commit is contained in:
+8
-4
@@ -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",
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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<NestFastifyApplication>(AppModule, new FastifyAdapter({ logger: true }));
|
||||
|
||||
app.useWebSocketAdapter(new IoAdapter(app));
|
||||
|
||||
app.setGlobalPrefix('api');
|
||||
|
||||
app.useGlobalPipes(
|
||||
|
||||
@@ -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],
|
||||
|
||||
@@ -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<string>('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) };
|
||||
}
|
||||
|
||||
|
||||
@@ -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<number, ScanEntry>();
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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<void> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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');
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<string>('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 {}
|
||||
|
||||
@@ -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<void> {
|
||||
await this.db
|
||||
.update(scanJobs)
|
||||
.set({ status: 'failed', errorMessage, completedAt: new Date() })
|
||||
.where(eq(scanJobs.status, 'running'));
|
||||
}
|
||||
|
||||
// ── Library Folders ────────────────────────────────────────────────────────
|
||||
|
||||
async findLibraryFolders(libraryId: number) {
|
||||
|
||||
@@ -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<void> {
|
||||
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<ReturnType<ScannerRepository['findLibraryFolders']>>,
|
||||
): Promise<void> {
|
||||
this.logger.log(`Scan job ${jobId} started for library ${libraryId}`);
|
||||
|
||||
type FolderWork = {
|
||||
id: number;
|
||||
libraryId: number;
|
||||
path: string;
|
||||
candidates: BookCandidate[];
|
||||
knownBooks: Awaited<ReturnType<ScannerRepository['findBooksByLibraryFolder']>>;
|
||||
knownFiles: Awaited<ReturnType<ScannerRepository['findBookFilesByLibraryFolder']>>;
|
||||
};
|
||||
|
||||
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<ScanCounts> {
|
||||
private async scanFolderCandidates(
|
||||
libraryFolderId: number,
|
||||
libraryId: number,
|
||||
folderPath: string,
|
||||
candidates: BookCandidate[],
|
||||
knownBooks: Awaited<ReturnType<ScannerRepository['findBooksByLibraryFolder']>>,
|
||||
knownFiles: Awaited<ReturnType<ScannerRepository['findBookFilesByLibraryFolder']>>,
|
||||
jobId: number,
|
||||
): Promise<ScanCounts> {
|
||||
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<number>();
|
||||
|
||||
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<string, { id: number; status: string; folderPath: string }>,
|
||||
fileByPath: Map<string, { id: number; ino: number; sizeBytes: number | null; mtime: Date | null; hash: string | null }>,
|
||||
fileByIno: Map<number, { id: number; absolutePath: string }>,
|
||||
): 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);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user