nontara/nontara
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(scanner): implement concurrent processing with queue management

Browse Source 
Add p-queue dependency and implement controlled concurrent processing for library scanning operations. Introduce separate queues for directory processing, file processing, metadata API calls, and database operations to prevent system overload and improve scanning performance.

Key improvements:
- Directory queue with 15 concurrent directory processing limit
- File processing queue with 5 concurrent files
- Rate-limited metadata queue (40 requests per 10 seconds for TMDB)
- Database queue matching connection pool size
- Batch database operations to reduce query count
- Progress tracking and queue statistics
- Optimized credit insertion with batch operations
This commit is contained in:
Kenzuya
2025-10-22 17:19:17 +07:00
parent a3144c1796
commit 366b4a728a
6 changed files with 963 additions and 637 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
apps/server/package.json
View File

@@ -29,6 +29,7 @@
"drizzle-orm": "^0.38.0",
"fluent-ffmpeg": "^2.1.3",
"hono": "^4.8.2",
"p-queue": "^9.0.0",
"zod": "^4.0.2"
},
"devDependencies": {

221
apps/server/src/services/libraryScanner/LibraryScanner.ts
View File

@@ -1,6 +1,7 @@
import { readdir } from "node:fs/promises";
import { join } from "node:path";
import { eq } from "drizzle-orm";
import { eq, inArray } from "drizzle-orm";
import PQueue from "p-queue";
import { database } from "@/db";
import { libraryFolders, mediaItems } from "@/db/schema";
import type { LibraryType } from "@/db/schema/enums";
@@ -19,8 +20,32 @@ export class LibraryScannerService extends BaseService {
private scanners: Map<LibraryType, BaseLibraryScanner<unknown, unknown>> =
new Map();
// Video file extensions as Set for O(1) lookup
private readonly VIDEO_EXTENSIONS = new Set([
".mp4",
".mkv",
".avi",
".mov",
".wmv",
".flv",
".webm",
".m4v",
".mpg",
".mpeg",
]);
// Queue for controlled concurrent directory processing
private directoryQueue: PQueue;
constructor() {
super("LibraryScannerService");
// Initialize directory processing queue with concurrency limit
// This prevents overwhelming the system when scanning directories with many subfolders
this.directoryQueue = new PQueue({
concurrency: 15, // Process up to 15 directories concurrently
autoStart: true,
});
}
private get metadataService(): MetadataService {
@@ -40,6 +65,10 @@ export class LibraryScannerService extends BaseService {
public async dispose(): Promise<void> {
this.logger.info("Disposing library scanner service.");
// Clear directory queue
this.directoryQueue.clear();
this.scanners.clear();
this.isRunning = false;
}
@@ -71,8 +100,7 @@ export class LibraryScannerService extends BaseService {
this.logger.info(`Found ${folders.length} library folder(s) to scan`);
for (let i = 0; i < folders.length; i++) {
const folder = folders[i];
for (const folder of folders) {
await this.scanLibraryFolder(folder);
}
@@ -131,20 +159,8 @@ export class LibraryScannerService extends BaseService {
}
private isVideoFile(filename: string): boolean {
const videoExtensions = [
".mp4",
".mkv",
".avi",
".mov",
".wmv",
".flv",
".webm",
".m4v",
".mpg",
".mpeg",
];
const lowerFilename = filename.toLowerCase();
return videoExtensions.some((ext) => lowerFilename.endsWith(ext));
const ext = filename.substring(filename.lastIndexOf(".")).toLowerCase();
return this.VIDEO_EXTENSIONS.has(ext);
}
private async scanDirectory(
@@ -159,70 +175,151 @@ export class LibraryScannerService extends BaseService {
directories: 0,
};
let processedFiles = 0;
let skippedFiles = 0;
try {
this.logger.debug(
`[${libraryName}] Scanning directory: ${directoryPath}`,
);
const entries = await readdir(directoryPath, { withFileTypes: true });
const scanner = this.getScannerForType(libraryType);
if (!scanner) {
this.logger.warn(`Unsupported library type "${libraryType}"`);
return stats;
}
// Separate subdirectories and video files for concurrent processing
const subdirectories: string[] = [];
const videoFiles: Array<{
filename: string;
absolutePath: string;
}> = [];
for (const entry of entries) {
const fullPath = join(directoryPath, entry.name);
if (entry.isDirectory()) {
subdirectories.push(fullPath);
stats.directories++;
this.logger.debug(
`[${libraryName}] Found subdirectory: ${entry.name}, crawling...`,
);
const subStats = await this.scanDirectory(
fullPath,
libraryType,
libraryName,
metadataLanguage,
libraryFolderId,
);
stats.files += subStats.files;
stats.directories += subStats.directories;
} else if (entry.isFile() && this.isVideoFile(entry.name)) {
stats.files++;
const existingMedia = await database
.select()
.from(mediaItems)
.where(eq(mediaItems.path, fullPath))
.limit(1);
if (existingMedia.length > 0) {
this.logger.debug(
`[${libraryName}] ✓ Already in database, skipping: ${entry.name}`,
);
skippedFiles++;
continue;
}
const scanner = this.getScannerForType(libraryType);
if (!scanner) {
this.logger.warn(
`Unsupported library type "${libraryType}" while scanning ${fullPath}.`,
);
continue;
}
const processed = await scanner.scanFile({
videoFiles.push({
filename: entry.name,
absolutePath: fullPath,
});
}
}
// Process all subdirectories with controlled concurrency using queue
if (subdirectories.length > 0) {
this.logger.debug(
`[${libraryName}] Found ${subdirectories.length} subdirectories, processing with controlled concurrency...`,
);
// Use p-queue to limit concurrent directory processing
// This prevents memory/CPU issues with thousands of subdirectories
const subDirPromises = subdirectories.map((subDirPath) =>
this.directoryQueue.add(async () =>
this.scanDirectory(
subDirPath,
libraryType,
libraryName,
metadataLanguage,
libraryFolderId,
),
),
);
const subDirResults = await Promise.all(subDirPromises);
// Aggregate stats from all subdirectories
for (const subStats of subDirResults) {
stats.files += subStats.files;
stats.directories += subStats.directories;
}
}
// Batch check which files already exist in database (1 query instead of N)
let filesToScan = videoFiles;
let skippedFiles = 0;
if (videoFiles.length > 0) {
const allPaths = videoFiles.map((f) => f.absolutePath);
const existingMedia = await database
.select({ path: mediaItems.path })
.from(mediaItems)
.where(inArray(mediaItems.path, allPaths));
const existingPathsSet = new Set(existingMedia.map((m) => m.path));
// Filter out files that already exist
filesToScan = videoFiles.filter(
(f) => !existingPathsSet.has(f.absolutePath),
);
skippedFiles = videoFiles.length - filesToScan.length;
if (skippedFiles > 0) {
this.logger.debug(
`[${libraryName}] Skipping ${skippedFiles} files already in database`,
);
}
}
// Process new files concurrently using scanner's internal queue
if (filesToScan.length > 0) {
this.logger.info(
`[${libraryName}] Processing ${filesToScan.length} new files concurrently...`,
);
// Reset stats and set total count for progress tracking
scanner.resetStats();
for (const _file of filesToScan) {
scanner.incrementTotal();
}
// Queue all files for concurrent processing
const scanPromises = filesToScan.map((file) =>
scanner.scanFile({
filename: file.filename,
absolutePath: file.absolutePath,
libraryName,
metadataLanguage,
libraryFolderId,
});
}),
);
if (processed) processedFiles++;
else skippedFiles++;
// Wait for all scans to complete
const results = await Promise.allSettled(scanPromises);
// Wait for all queues to finish
await scanner.waitForCompletion();
// Track which files failed for detailed reporting
const failedFilesList: string[] = [];
results.forEach((result, index) => {
if (result.status === "rejected" || !result.value) {
failedFilesList.push(filesToScan[index].filename);
}
});
// Count successes and failures
const processedFiles = results.filter(
(r) => r.status === "fulfilled" && r.value,
).length;
const failedFiles = failedFilesList.length;
this.logger.info(
`[${libraryName}] Batch complete: ${processedFiles} successful, ${failedFiles} failed`,
);
if (failedFilesList.length > 0 && failedFilesList.length <= 10) {
this.logger.warn(
`[${libraryName}] Failed files: ${failedFilesList.join(", ")}`,
);
} else if (failedFilesList.length > 10) {
this.logger.warn(
`[${libraryName}] ${failedFilesList.length} files failed (too many to list)`,
);
}
}
@@ -230,7 +327,7 @@ export class LibraryScannerService extends BaseService {
`[${libraryName}] Directory scan complete: ${directoryPath}`,
);
this.logger.info(
` → Found ${stats.files} files | Processed: ${processedFiles} new, ${skippedFiles} skipped`,
` → Found ${stats.files} files | ${filesToScan.length} new, ${skippedFiles} skipped`,
);
return stats;

251
apps/server/src/services/libraryScanner/scanners/BaseLibraryScanner.ts
View File

@@ -2,6 +2,7 @@ import { findBy } from "@nontara/language-codes";
import type { ConsolaInstance } from "consola";
import { inArray } from "drizzle-orm";
import type ffmpeg from "fluent-ffmpeg";
import PQueue from "p-queue";
import type { database } from "@/db";
import { genres, mediaStreams, titleGenres, titleProviders } from "@/db/schema";
import type {
@@ -36,12 +37,62 @@ export abstract class BaseLibraryScanner<
protected readonly logger: ConsolaInstance;
protected metadataService: MetadataService;
// Concurrent processing queues (accessible by child classes)
protected readonly fileProcessingQueue: PQueue;
protected readonly metadataQueue: PQueue;
protected readonly databaseQueue: PQueue;
// Progress tracking
private scanStats = {
total: 0,
processed: 0,
failed: 0,
};
constructor(
protected readonly scannerName: string,
metadataService: MetadataService,
) {
this.logger = createLogger(scannerName);
this.metadataService = metadataService;
// File processing queue: Process 5 files simultaneously
this.fileProcessingQueue = new PQueue({
concurrency: 5,
timeout: 120000,
autoStart: true,
});
// Metadata API queue: Rate-limited for TMDB (40 requests per 10 seconds)
this.metadataQueue = new PQueue({
concurrency: 10,
interval: 10000,
intervalCap: 40,
timeout: 30000,
autoStart: true,
});
// Database queue: Match connection pool size
this.databaseQueue = new PQueue({
concurrency: 10,
timeout: 60000,
autoStart: true,
});
// Queue event handlers
this.fileProcessingQueue.on("active", () => {
this.logger.debug(
`Processing: ${this.fileProcessingQueue.pending}/${this.scanStats.total} files`,
);
});
this.fileProcessingQueue.on("idle", () => {
this.logger.success("File processing queue completed");
});
this.fileProcessingQueue.on("error", (error) => {
this.logger.error(`File processing queue error: ${error}`);
});
}
public abstract getLibraryType(): LibraryType;
@@ -66,57 +117,162 @@ export abstract class BaseLibraryScanner<
protected abstract formatParsedInfo(parsedInfo: TParsedInfo): string;
/**
* Reset scan statistics
*/
public resetStats(): void {
this.scanStats = {
total: 0,
processed: 0,
failed: 0,
};
}
/**
* Increment total files counter
*/
public incrementTotal(): void {
this.scanStats.total++;
}
/**
* Get queue statistics and progress
*/
public getQueueStats() {
return {
files: {
total: this.scanStats.total,
processed: this.scanStats.processed,
failed: this.scanStats.failed,
remaining:
this.fileProcessingQueue.size + this.fileProcessingQueue.pending,
percentage:
this.scanStats.total > 0
? Math.round(
(this.scanStats.processed / this.scanStats.total) * 100,
)
: 0,
},
queues: {
fileProcessing: {
size: this.fileProcessingQueue.size,
pending: this.fileProcessingQueue.pending,
isPaused: this.fileProcessingQueue.isPaused,
},
metadata: {
size: this.metadataQueue.size,
pending: this.metadataQueue.pending,
isPaused: this.metadataQueue.isPaused,
},
database: {
size: this.databaseQueue.size,
pending: this.databaseQueue.pending,
isPaused: this.databaseQueue.isPaused,
},
},
};
}
/**
* Pause all scanning queues
*/
public pause(): void {
this.fileProcessingQueue.pause();
this.metadataQueue.pause();
this.databaseQueue.pause();
this.logger.info("All scanning queues paused");
}
/**
* Resume all scanning queues
*/
public resume(): void {
this.fileProcessingQueue.start();
this.metadataQueue.start();
this.databaseQueue.start();
this.logger.info("All scanning queues resumed");
}
/**
* Wait for all queues to complete their current work
*/
public async waitForCompletion(): Promise<void> {
await Promise.all([
this.fileProcessingQueue.onIdle(),
this.metadataQueue.onIdle(),
this.databaseQueue.onIdle(),
]);
}
public async scanFile(options: ScanFileOptions): Promise<boolean> {
const { filename, absolutePath, libraryName } = options;
this.logger.debug(`[${libraryName}] Processing file: ${filename}`);
const parsedInfo: TParsedInfo | null = await this.parseFilename(filename);
if (!parsedInfo) {
this.logger.warn(`[${libraryName}] Cannot parse filename: ${filename}`);
return false;
}
try {
const parsedInfo: TParsedInfo | null = await this.parseFilename(filename);
if (!parsedInfo) {
this.logger.warn(`[${libraryName}] Cannot parse filename: ${filename}`);
this.scanStats.failed++;
return false;
}
this.logger.info(
`[${libraryName}] Parsed: ${this.formatParsedInfo(parsedInfo)}`,
);
this.logger.debug("Inspecting media file with ffprobe");
const mediaInfo = await this.inspectMediaFile(absolutePath);
if (!mediaInfo) {
this.logger.error(
`[${libraryName}] Failed to inspect media file: ${filename}`,
this.logger.info(
`[${libraryName}] Parsed: ${this.formatParsedInfo(parsedInfo)}`,
);
this.logger.debug("Inspecting media file with ffprobe");
const mediaInfo = await this.inspectMediaFile(absolutePath);
if (!mediaInfo) {
this.logger.error(
`[${libraryName}] Failed to inspect media file: ${filename}`,
);
this.scanStats.failed++;
return false;
}
const detectedLanguage = this.detectLanguageFromStreams(
mediaInfo.streams,
);
const durationSeconds = Math.round(mediaInfo.format.duration || 0);
this.logger.debug(
`Duration: ${Math.round(durationSeconds / 60)} minutes, Language: ${detectedLanguage || "unknown"}`,
);
const mediaInspection: MediaInspectionResult = {
durationSeconds,
detectedLanguage,
mediaInfo,
};
this.logger.debug("Fetching metadata from providers");
const metadata: TResolvedMetadata | null = await this.resolveMetadata(
parsedInfo,
mediaInspection,
options.metadataLanguage,
filename,
);
const result = await this.insertToDatabase(
parsedInfo,
metadata,
mediaInspection,
options,
);
if (result) {
this.scanStats.processed++;
} else {
this.scanStats.failed++;
}
return result;
} catch (error) {
this.scanStats.failed++;
this.logger.error(`[${libraryName}] Scan failed: ${error}`);
return false;
}
const detectedLanguage = this.detectLanguageFromStreams(mediaInfo.streams);
const durationSeconds = Math.round(mediaInfo.format.duration || 0);
this.logger.debug(
`Duration: ${Math.round(durationSeconds / 60)} minutes, Language: ${detectedLanguage || "unknown"}`,
);
const mediaInspection: MediaInspectionResult = {
durationSeconds,
detectedLanguage,
mediaInfo,
};
this.logger.debug("Fetching metadata from providers");
const metadata: TResolvedMetadata | null = await this.resolveMetadata(
parsedInfo,
mediaInspection,
options.metadataLanguage,
filename,
);
return await this.insertToDatabase(
parsedInfo,
metadata,
mediaInspection,
options,
);
}
protected async inspectMediaFile(
@@ -193,10 +349,13 @@ export abstract class BaseLibraryScanner<
}
// Batch insert all title-genre relationships
const titleGenreValues = genreNames.map((genreName) => ({
titleId,
genreId: existingGenreMap.get(genreName)!,
}));
const titleGenreValues = genreNames
.map((genreName) => {
const genreId = existingGenreMap.get(genreName);
if (!genreId) return null;
return { titleId, genreId };
})
.filter((value): value is NonNullable<typeof value> => value !== null);
await tx.insert(titleGenres).values(titleGenreValues).onConflictDoNothing();

290
apps/server/src/services/libraryScanner/scanners/MovieLibraryScanner.ts
View File

@@ -1,4 +1,4 @@
import { eq } from "drizzle-orm";
import { inArray } from "drizzle-orm";
import { database } from "@/db";
import { credits, mediaItems, movies, people, titles } from "@/db/schema";
import type { LibraryType } from "@/db/schema/enums";
@@ -61,7 +61,10 @@ export class MovieLibraryScanner extends BaseLibraryScanner<
};
try {
return await this.metadataService.resolveMovie(info);
// Use metadata queue for rate-limited API calls
return await this.metadataQueue.add(async () => {
return await this.metadataService.resolveMovie(info);
});
} catch (error) {
this.logger.debug(
`Metadata resolution failed for "${parsedInfo.title}" (${parsedInfo.year}) [${filename}]: ${error}`,
@@ -104,76 +107,90 @@ export class MovieLibraryScanner extends BaseLibraryScanner<
try {
let createdMovieId: string | null = null;
await database.transaction(async (tx) => {
const insertedTitle = await tx
.insert(titles)
.values({
libraryFolderId,
type: "MOVIE" as const,
title: finalMetadata.title,
originalTitle: finalMetadata.originalTitle,
tagline: finalMetadata.tagline || null,
overview: finalMetadata.description,
year: finalMetadata.year,
releaseDate: finalMetadata.releaseDate,
poster: finalMetadata.posterUrl,
backdrop: finalMetadata.backdropUrl || null,
rating: finalMetadata.rating || null,
originalLanguage: finalMetadata.originalLanguage || null,
homepage: finalMetadata.homepage || null,
})
.returning({ id: titles.id });
// Use database queue to control concurrent database operations
await this.databaseQueue.add(async () => {
await database.transaction(async (tx) => {
const insertedTitle = await tx
.insert(titles)
.values({
libraryFolderId,
type: "MOVIE" as const,
title: finalMetadata.title,
originalTitle: finalMetadata.originalTitle,
tagline: finalMetadata.tagline || null,
overview: finalMetadata.description,
year: finalMetadata.year,
releaseDate: finalMetadata.releaseDate,
poster: finalMetadata.posterUrl,
backdrop: finalMetadata.backdropUrl || null,
rating: finalMetadata.rating || null,
originalLanguage: finalMetadata.originalLanguage || null,
homepage: finalMetadata.homepage || null,
})
.returning({ id: titles.id });
const titleId = insertedTitle[0].id;
const titleId = insertedTitle[0].id;
const insertedMovie = await tx
.insert(movies)
.values({
titleId,
runtimeSeconds: durationSeconds,
})
.returning({ id: movies.id });
const insertedMovie = await tx
.insert(movies)
.values({
titleId,
runtimeSeconds: durationSeconds,
})
.returning({ id: movies.id });
const movieId = insertedMovie[0].id;
const movieId = insertedMovie[0].id;
const insertedMediaItem = await tx
.insert(mediaItems)
.values({
parentType: "MOVIE" as const,
movieId,
episodeId: null,
path: absolutePath,
etag: null,
sizeBytes: mediaInfo.format.size,
container: mediaInfo.format.format_name || null,
})
.returning({ id: mediaItems.id });
const insertedMediaItem = await tx
.insert(mediaItems)
.values({
parentType: "MOVIE" as const,
movieId,
episodeId: null,
path: absolutePath,
etag: null,
sizeBytes: mediaInfo.format.size,
container: mediaInfo.format.format_name || null,
})
.returning({ id: mediaItems.id });
const mediaItemId = insertedMediaItem[0].id;
const mediaItemId = insertedMediaItem[0].id;
if (finalMetadata.genres && finalMetadata.genres.length > 0) {
await this.insertGenres(titleId, finalMetadata.genres, tx);
}
if (finalMetadata.genres && finalMetadata.genres.length > 0) {
await this.insertGenres(titleId, finalMetadata.genres, tx);
}
if (finalMetadata.providerIds && finalMetadata.providerIds.length > 0) {
await this.insertProviderIds(titleId, finalMetadata.providerIds, tx);
}
if (
finalMetadata.providerIds &&
finalMetadata.providerIds.length > 0
) {
await this.insertProviderIds(
titleId,
finalMetadata.providerIds,
tx,
);
}
if (mediaInfo.streams && mediaInfo.streams.length > 0) {
await this.insertMediaStreams(mediaItemId, mediaInfo.streams, tx);
}
if (mediaInfo.streams && mediaInfo.streams.length > 0) {
await this.insertMediaStreams(mediaItemId, mediaInfo.streams, tx);
}
createdMovieId = movieId;
this.logger.debug(`Successfully inserted movie ID ${movieId}`);
createdMovieId = movieId;
this.logger.debug(`Successfully inserted movie ID ${movieId}`);
});
});
// Fetch and insert credits using metadata queue
if (createdMovieId && resolvedMetadata) {
try {
this.logger.debug("Fetching credits from metadata provider");
const creditsData = await this.resolveCredits(
createdMovieId,
resolvedMetadata,
);
const movieId = createdMovieId;
// Use metadata queue for rate-limited API calls
const creditsData = await this.metadataQueue.add(async () => {
return await this.resolveCredits(movieId, resolvedMetadata);
});
if (creditsData) {
await this.insertCredits(createdMovieId, creditsData);
@@ -219,81 +236,102 @@ export class MovieLibraryScanner extends BaseLibraryScanner<
movieId: string,
creditsData: ResolvedTitleCredits,
): Promise<void> {
let castInserted = 0;
let crewInserted = 0;
await database.transaction(async (tx) => {
for (const castMember of creditsData.cast) {
let personId: string;
const existingPerson = await tx
.select({ id: people.id })
.from(people)
.where(eq(people.name, castMember.name))
.limit(1);
// Collect all unique person names
const allNames = [
...creditsData.cast.map((c) => c.name),
...creditsData.crew.map((c) => c.name),
];
if (existingPerson.length > 0) {
personId = existingPerson[0].id;
} else {
const insertedPerson = await tx
.insert(people)
.values({
name: castMember.name,
profilePicture: castMember.profilePath,
})
.returning({ id: people.id });
personId = insertedPerson[0].id;
}
await tx
.insert(credits)
.values({
personId,
movieId,
role: "CAST" as const,
character: castMember.character,
department: castMember.knownForDepartment,
order: castMember.order,
})
.onConflictDoNothing();
castInserted++;
if (allNames.length === 0) {
return;
}
for (const crewMember of creditsData.crew) {
let personId: string;
const existingPerson = await tx
.select({ id: people.id })
.from(people)
.where(eq(people.name, crewMember.name))
.limit(1);
// Batch lookup existing people (1 query)
const existingPeople = await tx
.select({ id: people.id, name: people.name })
.from(people)
.where(inArray(people.name, allNames));
if (existingPerson.length > 0) {
personId = existingPerson[0].id;
} else {
const insertedPerson = await tx
.insert(people)
.values({
name: crewMember.name,
profilePicture: crewMember.profilePath,
})
.returning({ id: people.id });
personId = insertedPerson[0].id;
const peopleMap = new Map(existingPeople.map((p) => [p.name, p.id]));
// Determine which people need to be created
const missingPeople = [
...creditsData.cast
.filter((c) => !peopleMap.has(c.name))
.map((c) => ({ name: c.name, profilePicture: c.profilePath })),
...creditsData.crew
.filter((c) => !peopleMap.has(c.name))
.map((c) => ({ name: c.name, profilePicture: c.profilePath })),
];
// Remove duplicates from missing people
const uniqueMissingPeople = Array.from(
new Map(missingPeople.map((p) => [p.name, p])).values(),
);
// Batch insert missing people (1 query)
if (uniqueMissingPeople.length > 0) {
const newPeople = await tx
.insert(people)
.values(uniqueMissingPeople)
.returning({ id: people.id, name: people.name });
for (const person of newPeople) {
peopleMap.set(person.name, person.id);
}
await tx
.insert(credits)
.values({
personId,
movieId,
role: "CREW" as const,
job: crewMember.job,
department: crewMember.department,
order: null,
})
.onConflictDoNothing();
crewInserted++;
}
// Build all credits entries
const allCredits = [
...creditsData.cast
.map((castMember) => {
const personId = peopleMap.get(castMember.name);
if (!personId) return null;
return {
personId,
movieId,
episodeId: null,
seriesId: null,
role: "CAST" as const,
character: castMember.character,
department: castMember.knownForDepartment,
order: castMember.order,
job: null,
};
})
.filter(
(credit): credit is NonNullable<typeof credit> => credit !== null,
),
...creditsData.crew
.map((crewMember) => {
const personId = peopleMap.get(crewMember.name);
if (!personId) return null;
return {
personId,
movieId,
episodeId: null,
seriesId: null,
role: "CREW" as const,
character: null,
department: crewMember.department,
order: null,
job: crewMember.job,
};
})
.filter(
(credit): credit is NonNullable<typeof credit> => credit !== null,
),
];
// Batch insert all credits (1 query)
if (allCredits.length > 0) {
await tx.insert(credits).values(allCredits).onConflictDoNothing();
}
this.logger.debug(
`Inserted ${creditsData.cast.length} cast and ${creditsData.crew.length} crew`,
);
});
this.logger.debug(`Inserted ${castInserted} cast and ${crewInserted} crew`);
}
}

826
apps/server/src/services/libraryScanner/scanners/SeriesLibraryScanner.ts
View File

@@ -1,4 +1,4 @@
import { and, eq } from "drizzle-orm";
import { and, eq, inArray } from "drizzle-orm";
import { database } from "@/db";
import {
credits,
@@ -79,7 +79,7 @@ export class SeriesLibraryScanner extends BaseLibraryScanner<
filename: string,
): Promise<ResolvedSeriesMetadata | null> {
const { durationSeconds, detectedLanguage } = mediaInspection;
const info: SeriesInfo = {
fileName: filename,
runtimeSeconds: durationSeconds,
@@ -92,7 +92,10 @@ export class SeriesLibraryScanner extends BaseLibraryScanner<
};
try {
return await this.metadataService.resolveSeries(info);
// Use metadata queue for rate-limited API calls
return await this.metadataQueue.add(async () => {
return await this.metadataService.resolveSeries(info);
});
} catch (error) {
this.logger.debug(`Metadata resolution failed: ${error}`);
return null;
@@ -110,36 +113,140 @@ export class SeriesLibraryScanner extends BaseLibraryScanner<
const { absolutePath, libraryName, libraryFolderId } = options;
try {
await database.transaction(async (tx) => {
const seriesTitle = resolvedMetadata?.title || parsedInfo.title;
const seasonNumber =
resolvedMetadata?.season.number || parsedInfo.season;
const seasonName = resolvedMetadata?.season.name;
// Use database queue to control concurrent database operations
await this.databaseQueue.add(async () => {
await database.transaction(async (tx) => {
const seriesTitle = resolvedMetadata?.title || parsedInfo.title;
const seasonNumber =
resolvedMetadata?.season.number || parsedInfo.season;
const seasonName = resolvedMetadata?.season.name;
// Create cache key for this series
const cacheKey = `${libraryFolderId}:${seriesTitle}`;
// Create cache key for this series
const cacheKey = `${libraryFolderId}:${seriesTitle}`;
// Try to get from cache first
let cacheEntry = this.seriesCache.get(cacheKey);
let seriesId: string;
let titleId: string;
let seasonId: string;
let isNewlyCreatedSeries = false;
// Try to get from cache first
let cacheEntry = this.seriesCache.get(cacheKey);
let seriesId: string;
let titleId: string;
let seasonId: string;
let isNewlyCreatedSeries = false;
if (cacheEntry) {
// Use cached IDs
seriesId = cacheEntry.seriesId;
titleId = cacheEntry.titleId;
if (cacheEntry) {
// Use cached IDs
seriesId = cacheEntry.seriesId;
titleId = cacheEntry.titleId;
// Check if season is in cache
const cachedSeasonId = cacheEntry.seasons.get(seasonNumber);
if (cachedSeasonId) {
seasonId = cachedSeasonId;
this.logger.debug(
`Using cached series/season IDs for "${seriesTitle}" S${seasonNumber}`,
);
// Check if season is in cache
const cachedSeasonId = cacheEntry.seasons.get(seasonNumber);
if (cachedSeasonId) {
seasonId = cachedSeasonId;
this.logger.debug(
`Using cached series/season IDs for "${seriesTitle}" S${seasonNumber}`,
);
} else {
// Season not in cache, check database
const existingSeason = await tx
.select({ id: seasons.id })
.from(seasons)
.where(
and(
eq(seasons.seriesId, seriesId),
eq(seasons.seasonNumber, seasonNumber),
),
)
.limit(1);
if (existingSeason.length > 0) {
seasonId = existingSeason[0].id;
} else {
const insertedSeason = await tx
.insert(seasons)
.values({
seriesId,
seasonNumber,
title: seasonName,
})
.returning({ id: seasons.id });
seasonId = insertedSeason[0].id;
}
// Add season to cache
cacheEntry.seasons.set(seasonNumber, seasonId);
}
} else {
// Season not in cache, check database
// Not in cache, do optimized query with JOIN
this.logger.debug(
`Cache miss for "${seriesTitle}", querying database`,
);
// Query title and series in one go
const existingData = await tx
.select({
titleId: titles.id,
seriesId: series.id,
})
.from(titles)
.leftJoin(series, eq(series.titleId, titles.id))
.where(
and(
eq(titles.title, seriesTitle),
eq(titles.libraryFolderId, libraryFolderId),
),
)
.limit(1);
if (existingData.length > 0 && existingData[0].titleId) {
titleId = existingData[0].titleId;
if (existingData[0].seriesId) {
seriesId = existingData[0].seriesId;
} else {
// Title exists but series doesn't
const insertedSeries = await tx
.insert(series)
.values({
titleId,
status: (resolvedMetadata?.status as SeriesStatus) || null,
})
.returning({ id: series.id });
seriesId = insertedSeries[0].id;
isNewlyCreatedSeries = true;
}
} else {
// Neither title nor series exist, create both
const insertedTitle = await tx
.insert(titles)
.values({
libraryFolderId,
type: "SERIES" as const,
title: seriesTitle,
originalTitle:
resolvedMetadata?.originalTitle || parsedInfo.title,
tagline: resolvedMetadata?.tagline || null,
overview: resolvedMetadata?.description || null,
poster: resolvedMetadata?.posterUrl || null,
backdrop: resolvedMetadata?.backdropUrl || null,
rating: resolvedMetadata?.rating || null,
originalLanguage: resolvedMetadata?.originalLanguage || null,
homepage: resolvedMetadata?.homepage || null,
year: resolvedMetadata?.year || null,
releaseDate: resolvedMetadata?.firstAirDate || null,
})
.returning({ id: titles.id });
titleId = insertedTitle[0].id;
const insertedSeries = await tx
.insert(series)
.values({
titleId,
status: (resolvedMetadata?.status as SeriesStatus) || null,
})
.returning({ id: series.id });
seriesId = insertedSeries[0].id;
isNewlyCreatedSeries = true;
}
// Now handle season
const existingSeason = await tx
.select({ id: seasons.id })
.from(seasons)
@@ -165,245 +272,136 @@ export class SeriesLibraryScanner extends BaseLibraryScanner<
seasonId = insertedSeason[0].id;
}
// Add season to cache
cacheEntry.seasons.set(seasonNumber, seasonId);
}
} else {
// Not in cache, do optimized query with JOIN
this.logger.debug(
`Cache miss for "${seriesTitle}", querying database`,
);
// Query title and series in one go
const existingData = await tx
.select({
titleId: titles.id,
seriesId: series.id,
})
.from(titles)
.leftJoin(series, eq(series.titleId, titles.id))
.where(
and(
eq(titles.title, seriesTitle),
eq(titles.libraryFolderId, libraryFolderId),
),
)
.limit(1);
if (existingData.length > 0 && existingData[0].titleId) {
titleId = existingData[0].titleId;
if (existingData[0].seriesId) {
seriesId = existingData[0].seriesId;
} else {
// Title exists but series doesn't
const insertedSeries = await tx
.insert(series)
.values({
titleId,
status: (resolvedMetadata?.status as SeriesStatus) || null,
})
.returning({ id: series.id });
seriesId = insertedSeries[0].id;
isNewlyCreatedSeries = true;
}
} else {
// Neither title nor series exist, create both
const insertedTitle = await tx
.insert(titles)
.values({
libraryFolderId,
type: "SERIES" as const,
title: seriesTitle,
originalTitle:
resolvedMetadata?.originalTitle || parsedInfo.title,
tagline: resolvedMetadata?.tagline || null,
overview: resolvedMetadata?.description || null,
poster: resolvedMetadata?.posterUrl || null,
backdrop: resolvedMetadata?.backdropUrl || null,
rating: resolvedMetadata?.rating || null,
originalLanguage: resolvedMetadata?.originalLanguage || null,
homepage: resolvedMetadata?.homepage || null,
year: resolvedMetadata?.year || null,
releaseDate: resolvedMetadata?.firstAirDate || null,
})
.returning({ id: titles.id });
titleId = insertedTitle[0].id;
const insertedSeries = await tx
.insert(series)
.values({
titleId,
status: (resolvedMetadata?.status as SeriesStatus) || null,
})
.returning({ id: series.id });
seriesId = insertedSeries[0].id;
isNewlyCreatedSeries = true;
}
// Now handle season
const existingSeason = await tx
.select({ id: seasons.id })
.from(seasons)
.where(
and(
eq(seasons.seriesId, seriesId),
eq(seasons.seasonNumber, seasonNumber),
),
)
.limit(1);
if (existingSeason.length > 0) {
seasonId = existingSeason[0].id;
} else {
const insertedSeason = await tx
.insert(seasons)
.values({
seriesId,
seasonNumber,
title: seasonName,
})
.returning({ id: seasons.id });
seasonId = insertedSeason[0].id;
}
// Add to cache
cacheEntry = {
seriesId,
titleId,
seasons: new Map([[seasonNumber, seasonId]]),
};
this.seriesCache.set(cacheKey, cacheEntry);
this.logger.debug(`Cached series/season IDs for "${seriesTitle}"`);
}
const episodeNumber =
resolvedMetadata?.episode.number || parsedInfo.episode;
const episodeTitle =
resolvedMetadata?.episode.title || `Episode ${episodeNumber}`;
const episodeDescription =
resolvedMetadata?.episode.description || null;
const episodeRuntime =
resolvedMetadata?.runtimeSeconds || durationSeconds;
// Check if episode already exists to avoid duplicate inserts
const existingEpisode = await tx
.select({ id: episodes.id })
.from(episodes)
.where(
and(
eq(episodes.seriesId, seriesId),
eq(episodes.seasonId, seasonId),
eq(episodes.episodeNumber, episodeNumber),
),
)
.limit(1);
if (existingEpisode.length > 0) {
this.logger.warn(
`[${libraryName}] Episode already exists: "${parsedInfo.title}" S${String(parsedInfo.season).padStart(2, "0")}E${String(parsedInfo.episode).padStart(2, "0")}`,
);
return; // Skip this episode
}
const insertedEpisode = await tx
.insert(episodes)
.values({
seriesId,
seasonId,
seasonNumber,
episodeNumber,
title: episodeTitle,
overview: episodeDescription,
airDate: resolvedMetadata?.airDate || null,
runtimeSeconds: episodeRuntime,
backgroundPath: resolvedMetadata?.episode.backgroundUrl || null,
})
.returning({ id: episodes.id });
const episodeId = insertedEpisode[0].id;
const insertedMediaItem = await tx
.insert(mediaItems)
.values({
parentType: "EPISODE" as const,
movieId: null,
episodeId,
path: absolutePath,
etag: null,
sizeBytes: mediaInfo.format.size,
container: mediaInfo.format.format_name || null,
})
.returning({ id: mediaItems.id });
const mediaItemId = insertedMediaItem[0].id;
if (mediaInfo.streams && mediaInfo.streams.length > 0) {
await this.insertMediaStreams(mediaItemId, mediaInfo.streams, tx);
}
const genresToInsert =
resolvedMetadata?.genres && resolvedMetadata.genres.length > 0
? resolvedMetadata.genres
: ["TV Series"];
await this.insertGenres(titleId, genresToInsert, tx);
if (
resolvedMetadata?.providerIds &&
resolvedMetadata.providerIds.length > 0
) {
await this.insertProviderIds(
titleId,
resolvedMetadata.providerIds,
tx,
);
}
// Resolve and insert series-level credits (main cast/crew)
// Only fetch once when series is newly created
if (isNewlyCreatedSeries && resolvedMetadata) {
try {
this.logger.debug("Fetching series credits from metadata provider");
const seriesCreditsData = await this.resolveSeriesCredits(
// Add to cache
cacheEntry = {
seriesId,
resolvedMetadata,
);
if (seriesCreditsData) {
await this.insertSeriesCredits(seriesId, seriesCreditsData, tx);
} else {
this.logger.warn("Series credits not found");
}
} catch (error) {
this.logger.warn(`Failed to resolve series credits: ${error}`);
titleId,
seasons: new Map([[seasonNumber, seasonId]]),
};
this.seriesCache.set(cacheKey, cacheEntry);
this.logger.debug(`Cached series/season IDs for "${seriesTitle}"`);
}
}
// Resolve and insert episode credits
if (resolvedMetadata) {
try {
this.logger.debug(
"Fetching episode credits from metadata provider",
const episodeNumber =
resolvedMetadata?.episode.number || parsedInfo.episode;
const episodeTitle =
resolvedMetadata?.episode.title || `Episode ${episodeNumber}`;
const episodeDescription =
resolvedMetadata?.episode.description || null;
const episodeRuntime =
resolvedMetadata?.runtimeSeconds || durationSeconds;
// Check if episode already exists to avoid duplicate inserts
const existingEpisode = await tx
.select({ id: episodes.id })
.from(episodes)
.where(
and(
eq(episodes.seriesId, seriesId),
eq(episodes.seasonId, seasonId),
eq(episodes.episodeNumber, episodeNumber),
),
)
.limit(1);
if (existingEpisode.length > 0) {
this.logger.warn(
`[${libraryName}] Episode already exists: "${parsedInfo.title}" S${String(parsedInfo.season).padStart(2, "0")}E${String(parsedInfo.episode).padStart(2, "0")}`,
);
const creditsData = await this.resolveEpisodeCredits(
return; // Skip this episode
}
const insertedEpisode = await tx
.insert(episodes)
.values({
seriesId,
seasonId,
seasonNumber,
episodeNumber,
title: episodeTitle,
overview: episodeDescription,
airDate: resolvedMetadata?.airDate || null,
runtimeSeconds: episodeRuntime,
backgroundPath: resolvedMetadata?.episode.backgroundUrl || null,
})
.returning({ id: episodes.id });
const episodeId = insertedEpisode[0].id;
const insertedMediaItem = await tx
.insert(mediaItems)
.values({
parentType: "EPISODE" as const,
movieId: null,
episodeId,
resolvedMetadata,
);
path: absolutePath,
etag: null,
sizeBytes: mediaInfo.format.size,
container: mediaInfo.format.format_name || null,
})
.returning({ id: mediaItems.id });
if (creditsData) {
await this.insertEpisodeCredits(episodeId, creditsData, tx);
} else {
this.logger.warn("Episode credits not found");
}
} catch (error) {
this.logger.warn(`Failed to resolve episode credits: ${error}`);
const mediaItemId = insertedMediaItem[0].id;
if (mediaInfo.streams && mediaInfo.streams.length > 0) {
await this.insertMediaStreams(mediaItemId, mediaInfo.streams, tx);
}
}
this.logger.debug(
`Successfully inserted series "${parsedInfo.title}" S${String(parsedInfo.season).padStart(2, "0")}E${String(parsedInfo.episode).padStart(2, "0")}`,
);
const genresToInsert =
resolvedMetadata?.genres && resolvedMetadata.genres.length > 0
? resolvedMetadata.genres
: ["TV Series"];
await this.insertGenres(titleId, genresToInsert, tx);
if (
resolvedMetadata?.providerIds &&
resolvedMetadata.providerIds.length > 0
) {
await this.insertProviderIds(
titleId,
resolvedMetadata.providerIds,
tx,
);
}
// Parallelize metadata fetching for series and episode credits
if (resolvedMetadata) {
try {
this.logger.debug("Fetching credits from metadata provider");
// Fetch series and episode credits in parallel
const [seriesCreditsData, episodeCreditsData] = await Promise.all(
[
isNewlyCreatedSeries
? this.metadataQueue.add(async () =>
this.resolveSeriesCredits(seriesId, resolvedMetadata),
)
: Promise.resolve(null),
this.metadataQueue.add(async () =>
this.resolveEpisodeCredits(episodeId, resolvedMetadata),
),
],
);
// Insert credits in parallel
await Promise.all([
seriesCreditsData
? this.insertSeriesCredits(seriesId, seriesCreditsData, tx)
: Promise.resolve(),
episodeCreditsData
? this.insertEpisodeCredits(episodeId, episodeCreditsData, tx)
: Promise.resolve(),
]);
} catch (error) {
this.logger.warn(`Failed to resolve/insert credits: ${error}`);
}
}
this.logger.debug(
`Successfully inserted series "${parsedInfo.title}" S${String(parsedInfo.season).padStart(2, "0")}E${String(parsedInfo.episode).padStart(2, "0")}`,
);
});
});
this.logger.success(
@@ -427,7 +425,7 @@ export class SeriesLibraryScanner extends BaseLibraryScanner<
providerId: metadata.id,
});
} catch (error) {
this.logger.debug(`Failed to resolve series credits: ${error}`);
this.logger.warn(`Failed to resolve series credits: ${error}`);
return null;
}
}
@@ -445,7 +443,7 @@ export class SeriesLibraryScanner extends BaseLibraryScanner<
episodeNumber: metadata.episode.number,
});
} catch (error) {
this.logger.debug(`Failed to resolve episode credits: ${error}`);
this.logger.warn(`Failed to resolve episode credits: ${error}`);
return null;
}
}
@@ -455,85 +453,100 @@ export class SeriesLibraryScanner extends BaseLibraryScanner<
creditsData: ResolvedTitleCredits,
tx: DbTransaction,
): Promise<void> {
let castInserted = 0;
let crewInserted = 0;
// Collect all unique person names
const allNames = [
...creditsData.cast.map((c) => c.name),
...creditsData.crew.map((c) => c.name),
];
for (const castMember of creditsData.cast) {
let personId: string;
const existingPerson = await tx
.select({ id: people.id })
.from(people)
.where(eq(people.name, castMember.name))
.limit(1);
if (existingPerson.length > 0) {
personId = existingPerson[0].id;
} else {
const insertedPerson = await tx
.insert(people)
.values({
name: castMember.name,
profilePicture: castMember.profilePath,
})
.returning({ id: people.id });
personId = insertedPerson[0].id;
}
await tx
.insert(credits)
.values({
personId,
episodeId,
movieId: null,
seriesId: null,
role: "CAST" as const,
character: castMember.character,
department: castMember.knownForDepartment,
order: castMember.order,
})
.onConflictDoNothing();
castInserted++;
if (allNames.length === 0) {
return;
}
for (const crewMember of creditsData.crew) {
let personId: string;
const existingPerson = await tx
.select({ id: people.id })
.from(people)
.where(eq(people.name, crewMember.name))
.limit(1);
// Batch lookup existing people (1 query)
const existingPeople = await tx
.select({ id: people.id, name: people.name })
.from(people)
.where(inArray(people.name, allNames));
if (existingPerson.length > 0) {
personId = existingPerson[0].id;
} else {
const insertedPerson = await tx
.insert(people)
.values({
name: crewMember.name,
profilePicture: crewMember.profilePath,
})
.returning({ id: people.id });
personId = insertedPerson[0].id;
const peopleMap = new Map(existingPeople.map((p) => [p.name, p.id]));
// Determine which people need to be created
const missingPeople = [
...creditsData.cast
.filter((c) => !peopleMap.has(c.name))
.map((c) => ({ name: c.name, profilePicture: c.profilePath })),
...creditsData.crew
.filter((c) => !peopleMap.has(c.name))
.map((c) => ({ name: c.name, profilePicture: c.profilePath })),
];
// Remove duplicates from missing people
const uniqueMissingPeople = Array.from(
new Map(missingPeople.map((p) => [p.name, p])).values(),
);
// Batch insert missing people (1 query)
if (uniqueMissingPeople.length > 0) {
const newPeople = await tx
.insert(people)
.values(uniqueMissingPeople)
.returning({ id: people.id, name: people.name });
for (const person of newPeople) {
peopleMap.set(person.name, person.id);
}
}
await tx
.insert(credits)
.values({
personId,
episodeId,
movieId: null,
seriesId: null,
role: "CREW" as const,
job: crewMember.job,
department: crewMember.department,
order: null,
// Build all credits entries
const allCredits = [
...creditsData.cast
.map((castMember) => {
const personId = peopleMap.get(castMember.name);
if (!personId) return null;
return {
personId,
episodeId,
movieId: null,
seriesId: null,
role: "CAST" as const,
character: castMember.character,
department: castMember.knownForDepartment,
order: castMember.order,
job: null,
};
})
.onConflictDoNothing();
crewInserted++;
.filter(
(credit): credit is NonNullable<typeof credit> => credit !== null,
),
...creditsData.crew
.map((crewMember) => {
const personId = peopleMap.get(crewMember.name);
if (!personId) return null;
return {
personId,
episodeId,
movieId: null,
seriesId: null,
role: "CREW" as const,
character: null,
department: crewMember.department,
order: null,
job: crewMember.job,
};
})
.filter(
(credit): credit is NonNullable<typeof credit> => credit !== null,
),
];
// Batch insert all credits (1 query)
if (allCredits.length > 0) {
await tx.insert(credits).values(allCredits).onConflictDoNothing();
}
this.logger.debug(
`Inserted ${castInserted} cast and ${crewInserted} crew for episode`,
`Inserted ${creditsData.cast.length} cast and ${creditsData.crew.length} crew for episode`,
);
}
@@ -542,85 +555,100 @@ export class SeriesLibraryScanner extends BaseLibraryScanner<
creditsData: ResolvedTitleCredits,
tx: DbTransaction,
): Promise<void> {
let castInserted = 0;
let crewInserted = 0;
// Collect all unique person names
const allNames = [
...creditsData.cast.map((c) => c.name),
...creditsData.crew.map((c) => c.name),
];
for (const castMember of creditsData.cast) {
let personId: string;
const existingPerson = await tx
.select({ id: people.id })
.from(people)
.where(eq(people.name, castMember.name))
.limit(1);
if (existingPerson.length > 0) {
personId = existingPerson[0].id;
} else {
const insertedPerson = await tx
.insert(people)
.values({
name: castMember.name,
profilePicture: castMember.profilePath,
})
.returning({ id: people.id });
personId = insertedPerson[0].id;
}
await tx
.insert(credits)
.values({
personId,
seriesId,
movieId: null,
episodeId: null,
role: "CAST" as const,
character: castMember.character,
department: castMember.knownForDepartment,
order: castMember.order,
})
.onConflictDoNothing();
castInserted++;
if (allNames.length === 0) {
return;
}
for (const crewMember of creditsData.crew) {
let personId: string;
const existingPerson = await tx
.select({ id: people.id })
.from(people)
.where(eq(people.name, crewMember.name))
.limit(1);
// Batch lookup existing people (1 query)
const existingPeople = await tx
.select({ id: people.id, name: people.name })
.from(people)
.where(inArray(people.name, allNames));
if (existingPerson.length > 0) {
personId = existingPerson[0].id;
} else {
const insertedPerson = await tx
.insert(people)
.values({
name: crewMember.name,
profilePicture: crewMember.profilePath,
})
.returning({ id: people.id });
personId = insertedPerson[0].id;
const peopleMap = new Map(existingPeople.map((p) => [p.name, p.id]));
// Determine which people need to be created
const missingPeople = [
...creditsData.cast
.filter((c) => !peopleMap.has(c.name))
.map((c) => ({ name: c.name, profilePicture: c.profilePath })),
...creditsData.crew
.filter((c) => !peopleMap.has(c.name))
.map((c) => ({ name: c.name, profilePicture: c.profilePath })),
];
// Remove duplicates from missing people
const uniqueMissingPeople = Array.from(
new Map(missingPeople.map((p) => [p.name, p])).values(),
);
// Batch insert missing people (1 query)
if (uniqueMissingPeople.length > 0) {
const newPeople = await tx
.insert(people)
.values(uniqueMissingPeople)
.returning({ id: people.id, name: people.name });
for (const person of newPeople) {
peopleMap.set(person.name, person.id);
}
}
await tx
.insert(credits)
.values({
personId,
seriesId,
movieId: null,
episodeId: null,
role: "CREW" as const,
job: crewMember.job,
department: crewMember.department,
order: null,
// Build all credits entries
const allCredits = [
...creditsData.cast
.map((castMember) => {
const personId = peopleMap.get(castMember.name);
if (!personId) return null;
return {
personId,
seriesId,
movieId: null,
episodeId: null,
role: "CAST" as const,
character: castMember.character,
department: castMember.knownForDepartment,
order: castMember.order,
job: null,
};
})
.onConflictDoNothing();
crewInserted++;
.filter(
(credit): credit is NonNullable<typeof credit> => credit !== null,
),
...creditsData.crew
.map((crewMember) => {
const personId = peopleMap.get(crewMember.name);
if (!personId) return null;
return {
personId,
seriesId,
movieId: null,
episodeId: null,
role: "CREW" as const,
character: null,
department: crewMember.department,
order: null,
job: crewMember.job,
};
})
.filter(
(credit): credit is NonNullable<typeof credit> => credit !== null,
),
];
// Batch insert all credits (1 query)
if (allCredits.length > 0) {
await tx.insert(credits).values(allCredits).onConflictDoNothing();
}
this.logger.debug(
`Inserted ${castInserted} cast and ${crewInserted} crew for series`,
`Inserted ${creditsData.cast.length} cast and ${creditsData.crew.length} crew for series`,
);
}
}

11
bun.lock
View File

@@ -24,6 +24,7 @@
"drizzle-orm": "^0.38.0",
"fluent-ffmpeg": "^2.1.3",
"hono": "^4.8.2",
"p-queue": "^9.0.0",
"zod": "^4.0.2",
},
"devDependencies": {
@@ -1042,6 +1043,8 @@
"esutils": ["esutils@2.0.3", "", {}, "sha512-kVscqXk4OCp68SZ0dkgEKVi6/8ij300KBWTJq32P/dYeWTSwK41WyTxalN1eRmA5Z9UU/LX9D7FWSmV9SAYx6g=="],
"eventemitter3": ["eventemitter3@5.0.1", "", {}, "sha512-GWkBvjiSZK87ELrYOSESUYeVIc9mvLLf/nXalMOS5dYrgZq9o5OVkbZAVM06CVxYsCwH9BDZFPlQTlPA1j4ahA=="],
"expand-template": ["expand-template@2.0.3", "", {}, "sha512-XYfuKMvj4O35f/pOXLObndIRvyQ+/+6AhODh+OKWj9S9498pHHn/IMszH+gt0fBCRWMNfk1ZSp5x3AifmnI2vg=="],
"exsolve": ["exsolve@1.0.7", "", {}, "sha512-VO5fQUzZtI6C+vx4w/4BWJpg3s/5l+6pRQEHzFRM8WFi4XffSP1Z+4qi7GbjWbvRQEbdIco5mIMq+zX4rPuLrw=="],
@@ -1284,6 +1287,10 @@
"p-locate": ["p-locate@5.0.0", "", { "dependencies": { "p-limit": "^3.0.2" } }, "sha512-LaNjtRWUBY++zB5nE/NwcaoMylSPk+S+ZHNB1TzdbMJMny6dynpAGt7X/tl/QYq3TIeE6nxHppbo2LGymrG5Pw=="],
"p-queue": ["p-queue@9.0.0", "", { "dependencies": { "eventemitter3": "^5.0.1", "p-timeout": "^7.0.0" } }, "sha512-KO1RyxstL9g1mK76530TExamZC/S2Glm080Nx8PE5sTd7nlduDQsAfEl4uXX+qZjLiwvDauvzXavufy3+rJ9zQ=="],
"p-timeout": ["p-timeout@7.0.1", "", {}, "sha512-AxTM2wDGORHGEkPCt8yqxOTMgpfbEHqF51f/5fJCmwFC3C/zNcGT63SymH2ttOAaiIws2zVg4+izQCjrakcwHg=="],
"parent-module": ["parent-module@1.0.1", "", { "dependencies": { "callsites": "^3.0.0" } }, "sha512-GQ2EWRpQV8/o+Aw8YqtfZZPfNRWZYkbidE9k5rpl/hC3vtHHBfGm2Ifi6qWV+coDGkrUKZAxE3Lot5kcsRlh+g=="],
"parse5": ["parse5@7.3.0", "", { "dependencies": { "entities": "^6.0.0" } }, "sha512-IInvU7fabl34qmi9gY8XOVxhYyMyuH2xUNpb2q8/Y+7552KlejkRvqvD19nMoUW/uQGGbqNpA6Tufu5FL5BZgw=="],
@@ -1606,8 +1613,6 @@
"@libsql/hrana-client/node-fetch": ["node-fetch@3.3.2", "", { "dependencies": { "data-uri-to-buffer": "^4.0.0", "fetch-blob": "^3.1.4", "formdata-polyfill": "^4.0.10" } }, "sha512-dRB78srN/l6gqWulah9SrxeYnxeddIG30+GOqK/9OlLVyLg3HPnr6SqOWTWOXKRwC2eGYCkZ59NNuSgvSrpgOA=="],
"@nontara/server/@types/bun": ["@types/bun@1.2.23", "", { "dependencies": { "bun-types": "1.2.23" } }, "sha512-le8ueOY5b6VKYf19xT3McVbXqLqmxzPXHsQT/q9JHgikJ2X22wyTW3g3ohz2ZMnp7dod6aduIiq8A14Xyimm0A=="],
"@nontora/tmdb-metadata/@types/bun": ["@types/bun@1.2.23", "", { "dependencies": { "bun-types": "1.2.23" } }, "sha512-le8ueOY5b6VKYf19xT3McVbXqLqmxzPXHsQT/q9JHgikJ2X22wyTW3g3ohz2ZMnp7dod6aduIiq8A14Xyimm0A=="],
"@tailwindcss/oxide/detect-libc": ["detect-libc@2.1.2", "", {}, "sha512-Btj2BOOO83o3WyH59e8MgXsxEQVcarkUOpEYrubB0urwnN10yQ364rsiByU11nZlqWYZm05i/of7io4mzihBtQ=="],
@@ -1728,8 +1733,6 @@
"@esbuild-kit/core-utils/esbuild/@esbuild/win32-x64": ["@esbuild/win32-x64@0.18.20", "", { "os": "win32", "cpu": "x64" }, "sha512-kTdfRcSiDfQca/y9QIkng02avJ+NCaQvrMejlsB3RRv5sE9rRoeBPISaZpKxHELzRxZyLvNts1P27W3wV+8geQ=="],
"@nontara/server/@types/bun/bun-types": ["bun-types@1.2.23", "", { "dependencies": { "@types/node": "*" }, "peerDependencies": { "@types/react": "^19" } }, "sha512-R9f0hKAZXgFU3mlrA0YpE/fiDvwV0FT9rORApt2aQVWSuJDzZOyB5QLc0N/4HF57CS8IXJ6+L5E4W1bW6NS2Aw=="],
"@nontora/tmdb-metadata/@types/bun/bun-types": ["bun-types@1.2.23", "", { "dependencies": { "@types/node": "*" }, "peerDependencies": { "@types/react": "^19" } }, "sha512-R9f0hKAZXgFU3mlrA0YpE/fiDvwV0FT9rORApt2aQVWSuJDzZOyB5QLc0N/4HF57CS8IXJ6+L5E4W1bW6NS2Aw=="],
"@tanstack/router-plugin/chokidar/glob-parent": ["glob-parent@5.1.2", "", { "dependencies": { "is-glob": "^4.0.1" } }, "sha512-AOIgSQCepiJYwP3ARnGx+5VnTu2HBYdzbGP45eLw1vr3zB3vZLeyed1sC9hnbcOc9/SrMyM5RPQrkGz4aS9Zow=="],