barbercollie/ReadMeABook
1
0
Fork 1
mirror of https://github.com/kikootwo/ReadMeABook.git synced 2026-06-03 04:40:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement centralized logging with RMABLogger

Browse Source 
Replaces scattered console statements with a unified RMABLogger across backend API routes and services. Adds LOG_LEVEL-based filtering, job-aware database persistence, and context-based logging. Updates documentation to describe the new logging system and usage patterns. Also documents qBittorrent CSRF header fix
This commit is contained in:
kikootwo
2026-01-12 12:45:48 -05:00
parent ba5f5cf7d6
commit 682836237b
118 changed files with 1623 additions and 1079 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/processors/audible-refresh.processor.ts
+11 -11
View File
@@ -6,7 +6,7 @@
*/
import { prisma } from '../db';
import { createJobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
export interface AudibleRefreshPayload {
jobId?: string;
@@ -15,9 +15,9 @@ export interface AudibleRefreshPayload {
export async function processAudibleRefresh(payload: AudibleRefreshPayload): Promise<any> {
const { jobId, scheduledJobId } = payload;
const logger = jobId ? createJobLogger(jobId, 'AudibleRefresh') : null;
const logger = RMABLogger.forJob(jobId, 'AudibleRefresh');
await logger?.info('Starting Audible data refresh...');
logger.info('Starting Audible data refresh...');
const { getAudibleService } = await import('../integrations/audible.service');
const { getThumbnailCacheService } = await import('../services/thumbnail-cache.service');
@@ -40,13 +40,13 @@ export async function processAudibleRefresh(payload: AudibleRefreshPayload): Pro
newReleaseRank: null,
},
});
await logger?.info('Cleared previous popular/new-release flags in audible_cache');
logger.info('Cleared previous popular/new-release flags in audible_cache');
// Fetch popular and new releases - 200 items each
const popular = await audibleService.getPopularAudiobooks(200);
const newReleases = await audibleService.getNewReleases(200);
await logger?.info(`Fetched ${popular.length} popular, ${newReleases.length} new releases from Audible`);
logger.info(`Fetched ${popular.length} popular, ${newReleases.length} new releases from Audible`);
// Persist to audible_cache
let popularSaved = 0;
@@ -99,7 +99,7 @@ export async function processAudibleRefresh(payload: AudibleRefreshPayload): Pro
popularSaved++;
} catch (error) {
await logger?.error(`Failed to save popular audiobook ${audiobook.title}: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Failed to save popular audiobook ${audiobook.title}: ${error instanceof Error ? error.message : 'Unknown error'}`);
}
}
@@ -149,20 +149,20 @@ export async function processAudibleRefresh(payload: AudibleRefreshPayload): Pro
newReleasesSaved++;
} catch (error) {
await logger?.error(`Failed to save new release ${audiobook.title}: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Failed to save new release ${audiobook.title}: ${error instanceof Error ? error.message : 'Unknown error'}`);
}
}
await logger?.info(`Saved ${popularSaved} popular and ${newReleasesSaved} new releases to audible_cache`);
logger.info(`Saved ${popularSaved} popular and ${newReleasesSaved} new releases to audible_cache`);
// Cleanup unused thumbnails
await logger?.info('Cleaning up unused thumbnails...');
logger.info('Cleaning up unused thumbnails...');
const allActiveAsins = await prisma.audibleCache.findMany({
select: { asin: true },
});
const activeAsinSet = new Set(allActiveAsins.map(item => item.asin));
const deletedCount = await thumbnailCache.cleanupUnusedThumbnails(activeAsinSet);
await logger?.info(`Cleanup complete: ${deletedCount} unused thumbnails removed`);
logger.info(`Cleanup complete: ${deletedCount} unused thumbnails removed`);
return {
success: true,
@@ -172,7 +172,7 @@ export async function processAudibleRefresh(payload: AudibleRefreshPayload): Pro
thumbnailsDeleted: deletedCount,
};
} catch (error) {
await logger?.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
throw error;
}
}
src/lib/processors/cleanup-seeded-torrents.processor.ts
+16 -16
View File
@@ -6,7 +6,7 @@
*/
import { prisma } from '../db';
import { createJobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
export interface CleanupSeededTorrentsPayload {
jobId?: string;
@@ -15,9 +15,9 @@ export interface CleanupSeededTorrentsPayload {
export async function processCleanupSeededTorrents(payload: CleanupSeededTorrentsPayload): Promise<any> {
const { jobId, scheduledJobId } = payload;
const logger = jobId ? createJobLogger(jobId, 'CleanupSeededTorrents') : null;
const logger = RMABLogger.forJob(jobId, 'CleanupSeededTorrents');
await logger?.info('Starting cleanup job for seeded torrents...');
logger.info('Starting cleanup job for seeded torrents...');
try {
// Get indexer configuration with per-indexer seeding times
@@ -26,7 +26,7 @@ export async function processCleanupSeededTorrents(payload: CleanupSeededTorrent
const indexersConfigStr = await configService.get('prowlarr_indexers');
if (!indexersConfigStr) {
await logger?.warn('No indexer configuration found, skipping');
logger.warn('No indexer configuration found, skipping');
return {
success: false,
message: 'No indexer configuration',
@@ -42,7 +42,7 @@ export async function processCleanupSeededTorrents(payload: CleanupSeededTorrent
indexerConfigMap.set(indexer.name, indexer);
}
await logger?.info(`Loaded configuration for ${indexerConfigMap.size} indexers`);
logger.info(`Loaded configuration for ${indexerConfigMap.size} indexers`);
// Find all completed requests + soft-deleted requests (orphaned downloads)
// IMPORTANT: Only cleanup requests that are truly complete and not being actively processed
@@ -76,7 +76,7 @@ export async function processCleanupSeededTorrents(payload: CleanupSeededTorrent
take: 100, // Limit to 100 requests per run
});
await logger?.info(`Found ${completedRequests.length} requests to check (status: 'available' or soft-deleted)`);
logger.info(`Found ${completedRequests.length} requests to check (status: 'available' or soft-deleted)`);
let cleaned = 0;
let skipped = 0;
@@ -95,7 +95,7 @@ export async function processCleanupSeededTorrents(payload: CleanupSeededTorrent
// For soft-deleted SABnzbd requests, hard delete immediately (no seeding needed)
if (request.deletedAt) {
await prisma.request.delete({ where: { id: request.id } });
await logger?.info(`Hard-deleted orphaned SABnzbd request ${request.id}`);
logger.info(`Hard-deleted orphaned SABnzbd request ${request.id}`);
}
continue;
}
@@ -116,7 +116,7 @@ export async function processCleanupSeededTorrents(payload: CleanupSeededTorrent
// For soft-deleted requests with unlimited seeding, hard delete immediately
if (request.deletedAt) {
await prisma.request.delete({ where: { id: request.id } });
await logger?.info(`Hard-deleted orphaned request ${request.id} with unlimited seeding`);
logger.info(`Hard-deleted orphaned request ${request.id} with unlimited seeding`);
}
noConfig++;
continue;
@@ -146,7 +146,7 @@ export async function processCleanupSeededTorrents(payload: CleanupSeededTorrent
continue;
}
await logger?.info(`Torrent ${torrent.name} (${indexerName}) has met seeding requirement (${Math.floor(actualSeedingTime / 60)}/${seedingConfig.seedingTimeMinutes} minutes)`);
logger.info(`Torrent ${torrent.name} (${indexerName}) has met seeding requirement (${Math.floor(actualSeedingTime / 60)}/${seedingConfig.seedingTimeMinutes} minutes)`);
// CRITICAL: Check if any other active (non-deleted) request is using this same torrent hash
// This prevents deleting shared torrents when user re-requests the same audiobook
@@ -165,12 +165,12 @@ export async function processCleanupSeededTorrents(payload: CleanupSeededTorrent
});
if (otherActiveRequests.length > 0) {
await logger?.info(`Skipping torrent deletion - ${otherActiveRequests.length} other active request(s) still using this torrent (IDs: ${otherActiveRequests.map(r => r.id).join(', ')})`);
logger.info(`Skipping torrent deletion - ${otherActiveRequests.length} other active request(s) still using this torrent (IDs: ${otherActiveRequests.map(r => r.id).join(', ')})`);
// If this is a soft-deleted request, hard delete it but DON'T delete the torrent
if (request.deletedAt) {
await prisma.request.delete({ where: { id: request.id } });
await logger?.info(`Hard-deleted orphaned request ${request.id} (kept shared torrent for active requests)`);
logger.info(`Hard-deleted orphaned request ${request.id} (kept shared torrent for active requests)`);
}
skipped++;
@@ -183,18 +183,18 @@ export async function processCleanupSeededTorrents(payload: CleanupSeededTorrent
// If this is a soft-deleted request (orphaned download), hard delete it now
if (request.deletedAt) {
await prisma.request.delete({ where: { id: request.id } });
await logger?.info(`Hard-deleted orphaned request ${request.id} after torrent cleanup`);
logger.info(`Hard-deleted orphaned request ${request.id} after torrent cleanup`);
} else {
await logger?.info(`Deleted torrent and files for active request ${request.id}`);
logger.info(`Deleted torrent and files for active request ${request.id}`);
}
cleaned++;
} catch (error) {
await logger?.error(`Failed to cleanup request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Failed to cleanup request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
}
}
await logger?.info(`Cleanup complete: ${cleaned} torrents cleaned, ${skipped} still seeding, ${noConfig} unlimited`);
logger.info(`Cleanup complete: ${cleaned} torrents cleaned, ${skipped} still seeding, ${noConfig} unlimited`);
return {
success: true,
@@ -205,7 +205,7 @@ export async function processCleanupSeededTorrents(payload: CleanupSeededTorrent
unlimited: noConfig,
};
} catch (error) {
await logger?.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
throw error;
}
}
src/lib/processors/download-torrent.processor.ts
+13 -13
View File
@@ -8,7 +8,7 @@ import { prisma } from '../db';
import { getQBittorrentService } from '../integrations/qbittorrent.service';
import { getSABnzbdService } from '../integrations/sabnzbd.service';
import { getConfigService } from '../services/config.service';
import { createJobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
/**
* Process download job
@@ -18,10 +18,10 @@ import { createJobLogger } from '../utils/job-logger';
export async function processDownloadTorrent(payload: DownloadTorrentPayload): Promise<any> {
const { requestId, audiobook, torrent, jobId } = payload;
const logger = jobId ? createJobLogger(jobId, 'DownloadTorrent') : null;
const logger = RMABLogger.forJob(jobId, 'DownloadTorrent');
await logger?.info(`Processing request ${requestId} for "${audiobook.title}"`);
await logger?.info(`Selected result: ${torrent.title}`, {
logger.info(`Processing request ${requestId} for "${audiobook.title}"`);
logger.info(`Selected result: ${torrent.title}`, {
size: torrent.size,
seeders: torrent.seeders,
format: torrent.format,
@@ -48,7 +48,7 @@ export async function processDownloadTorrent(payload: DownloadTorrentPayload): P
if (clientType === 'sabnzbd') {
// Route to SABnzbd
await logger?.info(`Routing to SABnzbd`);
logger.info(`Routing to SABnzbd`);
const sabnzbd = await getSABnzbdService();
downloadClientId = await sabnzbd.addNZB(torrent.downloadUrl, {
@@ -57,7 +57,7 @@ export async function processDownloadTorrent(payload: DownloadTorrentPayload): P
});
downloadClient = 'sabnzbd';
await logger?.info(`NZB added with ID: ${downloadClientId}`);
logger.info(`NZB added with ID: ${downloadClientId}`);
// Create DownloadHistory record
const downloadHistory = await prisma.downloadHistory.create({
@@ -79,7 +79,7 @@ export async function processDownloadTorrent(payload: DownloadTorrentPayload): P
},
});
await logger?.info(`Created download history record: ${downloadHistory.id}`);
logger.info(`Created download history record: ${downloadHistory.id}`);
// Trigger monitor download job with initial delay
const jobQueue = getJobQueueService();
@@ -91,7 +91,7 @@ export async function processDownloadTorrent(payload: DownloadTorrentPayload): P
3 // Wait 3 seconds before first check
);
await logger?.info(`Started monitoring job for request ${requestId} (SABnzbd, 3s initial delay)`);
logger.info(`Started monitoring job for request ${requestId} (SABnzbd, 3s initial delay)`);
return {
success: true,
@@ -107,7 +107,7 @@ export async function processDownloadTorrent(payload: DownloadTorrentPayload): P
};
} else {
// Route to qBittorrent (default)
await logger?.info(`Routing to qBittorrent`);
logger.info(`Routing to qBittorrent`);
const qbt = await getQBittorrentService();
downloadClientId = await qbt.addTorrent(torrent.downloadUrl, {
@@ -118,7 +118,7 @@ export async function processDownloadTorrent(payload: DownloadTorrentPayload): P
});
downloadClient = 'qbittorrent';
await logger?.info(`Torrent added with hash: ${downloadClientId}`);
logger.info(`Torrent added with hash: ${downloadClientId}`);
// Create DownloadHistory record
const downloadHistory = await prisma.downloadHistory.create({
@@ -140,7 +140,7 @@ export async function processDownloadTorrent(payload: DownloadTorrentPayload): P
},
});
await logger?.info(`Created download history record: ${downloadHistory.id}`);
logger.info(`Created download history record: ${downloadHistory.id}`);
// Trigger monitor download job with initial delay
const jobQueue = getJobQueueService();
@@ -152,7 +152,7 @@ export async function processDownloadTorrent(payload: DownloadTorrentPayload): P
3 // Wait 3 seconds before first check to avoid race condition
);
await logger?.info(`Started monitoring job for request ${requestId} (qBittorrent, 3s initial delay)`);
logger.info(`Started monitoring job for request ${requestId} (qBittorrent, 3s initial delay)`);
return {
success: true,
@@ -169,7 +169,7 @@ export async function processDownloadTorrent(payload: DownloadTorrentPayload): P
};
}
} catch (error) {
await logger?.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
// Update request status to failed
await prisma.request.update({
src/lib/processors/match-plex.processor.ts
+11 -11
View File
@@ -11,7 +11,7 @@ import { prisma } from '../db';
import { getLibraryService } from '../services/library';
import { compareTwoStrings } from 'string-similarity';
import { getConfigService } from '../services/config.service';
import { createJobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
/**
* Process match library job (DEPRECATED - use scan_library instead)
@@ -20,10 +20,10 @@ import { createJobLogger } from '../utils/job-logger';
export async function processMatchPlex(payload: MatchPlexPayload): Promise<any> {
const { requestId, audiobookId, title, author, jobId } = payload;
const logger = jobId ? createJobLogger(jobId, 'MatchLibrary') : null;
const logger = RMABLogger.forJob(jobId, 'MatchLibrary');
await logger?.warn('DEPRECATED: match_plex job is deprecated. Use scan_plex instead.');
await logger?.info(`Matching "${title}" by ${author} in library`);
logger.warn('DEPRECATED: match_plex job is deprecated. Use scan_plex instead.');
logger.info(`Matching "${title}" by ${author} in library`);
try {
// Get library service and configuration
@@ -31,7 +31,7 @@ export async function processMatchPlex(payload: MatchPlexPayload): Promise<any>
const libraryService = await getLibraryService();
const backendMode = await configService.getBackendMode();
await logger?.info(`Backend mode: ${backendMode}`);
logger.info(`Backend mode: ${backendMode}`);
// Get configured library ID
const libraryId = backendMode === 'audiobookshelf'
@@ -45,10 +45,10 @@ export async function processMatchPlex(payload: MatchPlexPayload): Promise<any>
// Search library using abstraction layer
const searchResults = await libraryService.searchItems(libraryId, title);
await logger?.info(`Found ${searchResults.length} results in library`);
logger.info(`Found ${searchResults.length} results in library`);
if (searchResults.length === 0) {
await logger?.warn(`No matches found in library for "${title}"`);
logger.warn(`No matches found in library for "${title}"`);
// Mark as completed anyway - the file is there, library just needs time to scan
await prisma.request.update({
@@ -92,7 +92,7 @@ export async function processMatchPlex(payload: MatchPlexPayload): Promise<any>
const bestMatch = matches[0];
await logger?.info(`Best match: "${bestMatch.item.title}" by ${bestMatch.item.author || 'Unknown'}`, {
logger.info(`Best match: "${bestMatch.item.title}" by ${bestMatch.item.author || 'Unknown'}`, {
score: Math.round(bestMatch.score * 100),
titleScore: Math.round(bestMatch.titleScore * 100),
authorScore: Math.round(bestMatch.authorScore * 100),
@@ -100,7 +100,7 @@ export async function processMatchPlex(payload: MatchPlexPayload): Promise<any>
// Accept match if score >= 70%
if (bestMatch.score >= 0.7) {
await logger?.info(`Match accepted!`);
logger.info(`Match accepted!`);
// Update audiobook with library item ID
const updateData: any = {
@@ -144,7 +144,7 @@ export async function processMatchPlex(payload: MatchPlexPayload): Promise<any>
},
};
} else {
await logger?.warn(`Match score too low (${Math.round(bestMatch.score * 100)}%), but marking as completed anyway`);
logger.warn(`Match score too low (${Math.round(bestMatch.score * 100)}%), but marking as completed anyway`);
// Mark as completed even if match is poor
await prisma.request.update({
@@ -166,7 +166,7 @@ export async function processMatchPlex(payload: MatchPlexPayload): Promise<any>
};
}
} catch (error) {
await logger?.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
// Don't fail the request - the files are organized correctly
// Just log the error and mark as completed
src/lib/processors/monitor-download.processor.ts
+12 -12
View File
@@ -7,7 +7,7 @@ import path from 'path';
import { MonitorDownloadPayload, getJobQueueService } from '../services/job-queue.service';
import { prisma } from '../db';
import { getQBittorrentService } from '../integrations/qbittorrent.service';
import { createJobLogger, JobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
import { PathMapper } from '../utils/path-mapper';
import { getConfigService } from '../services/config.service';
@@ -18,7 +18,7 @@ import { getConfigService } from '../services/config.service';
async function getTorrentWithRetry(
qbt: any,
hash: string,
logger: JobLogger | null,
logger: RMABLogger,
maxRetries: number = 3,
initialDelayMs: number = 500
): Promise<any> {
@@ -37,7 +37,7 @@ async function getTorrentWithRetry(
// Exponential backoff: 500ms, 1000ms, 2000ms
const delayMs = initialDelayMs * Math.pow(2, attempt);
await logger?.warn(`Torrent ${hash} not found, retrying in ${delayMs}ms (attempt ${attempt + 1}/${maxRetries})`);
logger.warn(`Torrent ${hash} not found, retrying in ${delayMs}ms (attempt ${attempt + 1}/${maxRetries})`);
await new Promise(resolve => setTimeout(resolve, delayMs));
}
@@ -55,7 +55,7 @@ async function getTorrentWithRetry(
export async function processMonitorDownload(payload: MonitorDownloadPayload): Promise<any> {
const { requestId, downloadHistoryId, downloadClientId, downloadClient, jobId } = payload;
const logger = jobId ? createJobLogger(jobId, 'MonitorDownload') : null;
const logger = RMABLogger.forJob(jobId, 'MonitorDownload');
try {
let progress: any;
@@ -96,7 +96,7 @@ export async function processMonitorDownload(payload: MonitorDownloadPayload): P
// Store download path if available (only set after completion)
downloadPath = nzbInfo.downloadPath;
await logger?.info(`SABnzbd status: ${nzbInfo.status}`, {
logger.info(`SABnzbd status: ${nzbInfo.status}`, {
progress: `${(nzbInfo.progress * 100).toFixed(1)}%`,
speed: `${(nzbInfo.downloadSpeed / 1024 / 1024).toFixed(2)} MB/s`,
});
@@ -123,7 +123,7 @@ export async function processMonitorDownload(payload: MonitorDownloadPayload): P
// Check download state
if (progress.state === 'completed') {
await logger?.info(`Download completed for request ${requestId}`);
logger.info(`Download completed for request ${requestId}`);
// Ensure we have a download path
if (!downloadPath) {
@@ -145,7 +145,7 @@ export async function processMonitorDownload(payload: MonitorDownloadPayload): P
localPath: pathMappingConfig.download_client_local_path || '',
});
await logger?.info(`Download completed`, {
logger.info(`Download completed`, {
downloadClient,
downloadPath,
organizePath: organizePath !== downloadPath ? `${organizePath} (mapped)` : organizePath,
@@ -183,7 +183,7 @@ export async function processMonitorDownload(payload: MonitorDownloadPayload): P
organizePath
);
await logger?.info(`Triggered organize_files job for request ${requestId}`);
logger.info(`Triggered organize_files job for request ${requestId}`);
return {
success: true,
@@ -194,7 +194,7 @@ export async function processMonitorDownload(payload: MonitorDownloadPayload): P
downloadPath: organizePath,
};
} else if (progress.state === 'failed') {
await logger?.error(`Download failed for request ${requestId}`);
logger.error(`Download failed for request ${requestId}`);
// Update request to failed
await prisma.request.update({
@@ -236,7 +236,7 @@ export async function processMonitorDownload(payload: MonitorDownloadPayload): P
// Only log every 5% progress to reduce log spam
const shouldLog = progress.percent % 5 === 0 || progress.percent < 5;
if (shouldLog) {
await logger?.info(`Request ${requestId}: ${progress.percent}% complete (${progress.state})`, {
logger.info(`Request ${requestId}: ${progress.percent}% complete (${progress.state})`, {
speed: progress.speed,
eta: progress.eta,
});
@@ -254,7 +254,7 @@ export async function processMonitorDownload(payload: MonitorDownloadPayload): P
};
}
} catch (error) {
await logger?.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
// Check if this is a transient "torrent not found" error
const errorMessage = error instanceof Error ? error.message : '';
@@ -263,7 +263,7 @@ export async function processMonitorDownload(payload: MonitorDownloadPayload): P
if (isTorrentNotFound) {
// Transient error - don't mark request as failed, let Bull retry
// The request stays in 'downloading' status until Bull exhausts all retries
await logger?.warn(`Transient error for request ${requestId}, allowing Bull to retry`);
logger.warn(`Transient error for request ${requestId}, allowing Bull to retry`);
} else {
// Permanent error - mark request as failed immediately
await prisma.request.update({
src/lib/processors/monitor-rss-feeds.processor.ts
+12 -12
View File
@@ -6,7 +6,7 @@
*/
import { prisma } from '../db';
import { createJobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
import { getJobQueueService } from '../services/job-queue.service';
export interface MonitorRssFeedsPayload {
@@ -16,9 +16,9 @@ export interface MonitorRssFeedsPayload {
export async function processMonitorRssFeeds(payload: MonitorRssFeedsPayload): Promise<any> {
const { jobId, scheduledJobId } = payload;
const logger = jobId ? createJobLogger(jobId, 'MonitorRssFeeds') : null;
const logger = RMABLogger.forJob(jobId, 'MonitorRssFeeds');
await logger?.info(`Starting RSS feed monitoring...`);
logger.info(`Starting RSS feed monitoring...`);
// Get indexer configuration
const { getConfigService } = await import('../services/config.service');
@@ -26,7 +26,7 @@ export async function processMonitorRssFeeds(payload: MonitorRssFeedsPayload): P
const indexersConfigStr = await configService.get('prowlarr_indexers');
if (!indexersConfigStr) {
await logger?.warn(`No indexers configured, skipping`);
logger.warn(`No indexers configured, skipping`);
return { success: false, message: 'No indexers configured', skipped: true };
}
@@ -38,11 +38,11 @@ export async function processMonitorRssFeeds(payload: MonitorRssFeedsPayload): P
);
if (rssEnabledIndexers.length === 0) {
await logger?.warn(`No indexers with RSS enabled, skipping`);
logger.warn(`No indexers with RSS enabled, skipping`);
return { success: false, message: 'No RSS-enabled indexers', skipped: true };
}
await logger?.info(`Monitoring ${rssEnabledIndexers.length} RSS-enabled indexers`);
logger.info(`Monitoring ${rssEnabledIndexers.length} RSS-enabled indexers`);
// Get RSS feeds from all enabled indexers
const { getProwlarrService } = await import('../integrations/prowlarr.service');
@@ -51,7 +51,7 @@ export async function processMonitorRssFeeds(payload: MonitorRssFeedsPayload): P
const indexerIds = rssEnabledIndexers.map((i: any) => i.id);
const rssResults = await prowlarrService.getAllRssFeeds(indexerIds);
await logger?.info(`Retrieved ${rssResults.length} items from RSS feeds`);
logger.info(`Retrieved ${rssResults.length} items from RSS feeds`);
if (rssResults.length === 0) {
return { success: true, message: 'No RSS results', matched: 0 };
@@ -67,7 +67,7 @@ export async function processMonitorRssFeeds(payload: MonitorRssFeedsPayload): P
take: 100,
});
await logger?.info(`Found ${missingRequests.length} requests awaiting search`);
logger.info(`Found ${missingRequests.length} requests awaiting search`);
if (missingRequests.length === 0) {
return { success: true, message: 'No missing requests', matched: 0 };
@@ -92,7 +92,7 @@ export async function processMonitorRssFeeds(payload: MonitorRssFeedsPayload): P
const titleMatchCount = titleWords.filter(word => word.length > 2 && torrentTitle.includes(word)).length;
if (hasAuthor && titleMatchCount >= 2) {
await logger?.info(`Match found! "${audiobook.title}" by ${audiobook.author} matches torrent: ${torrent.title}`);
logger.info(`Match found! "${audiobook.title}" by ${audiobook.author} matches torrent: ${torrent.title}`);
// Trigger search job to process this request
try {
@@ -102,9 +102,9 @@ export async function processMonitorRssFeeds(payload: MonitorRssFeedsPayload): P
author: audiobook.author,
});
matched++;
await logger?.info(`Triggered search job for request ${request.id}`);
logger.info(`Triggered search job for request ${request.id}`);
} catch (error) {
await logger?.error(`Failed to trigger search for request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Failed to trigger search for request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
}
// Only trigger once per request
@@ -113,7 +113,7 @@ export async function processMonitorRssFeeds(payload: MonitorRssFeedsPayload): P
}
}
await logger?.info(`RSS monitoring complete: ${matched} matches found and queued for processing`);
logger.info(`RSS monitoring complete: ${matched} matches found and queued for processing`);
return {
success: true,
src/lib/processors/organize-files.processor.ts
+13 -14
View File
@@ -6,7 +6,7 @@
import { OrganizeFilesPayload, getJobQueueService } from '../services/job-queue.service';
import { prisma } from '../db';
import { getFileOrganizer } from '../utils/file-organizer';
import { createJobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
import { getLibraryService } from '../services/library';
import { getConfigService } from '../services/config.service';
@@ -17,11 +17,10 @@ import { getConfigService } from '../services/config.service';
export async function processOrganizeFiles(payload: OrganizeFilesPayload): Promise<any> {
const { requestId, audiobookId, downloadPath, jobId } = payload;
// Create logger (fallback to console-only if jobId not provided)
const logger = jobId ? createJobLogger(jobId, 'OrganizeFiles') : null;
const logger = RMABLogger.forJob(jobId, 'OrganizeFiles');
await logger?.info(`Processing request ${requestId}`);
await logger?.info(`Download path: ${downloadPath}`);
logger.info(`Processing request ${requestId}`);
logger.info(`Download path: ${downloadPath}`);
try {
// Update request status to processing
@@ -43,7 +42,7 @@ export async function processOrganizeFiles(payload: OrganizeFilesPayload): Promi
throw new Error(`Audiobook ${audiobookId} not found`);
}
await logger?.info(`Organizing: ${audiobook.title} by ${audiobook.author}`);
logger.info(`Organizing: ${audiobook.title} by ${audiobook.author}`);
// Get file organizer (reads media_dir from database config)
const organizer = await getFileOrganizer();
@@ -65,7 +64,7 @@ export async function processOrganizeFiles(payload: OrganizeFilesPayload): Promi
throw new Error(`File organization failed: ${result.errors.join(', ')}`);
}
await logger?.info(`Successfully moved ${result.filesMovedCount} files to ${result.targetPath}`);
logger.info(`Successfully moved ${result.filesMovedCount} files to ${result.targetPath}`);
// Update audiobook record with file path and status
await prisma.audiobook.update({
@@ -89,7 +88,7 @@ export async function processOrganizeFiles(payload: OrganizeFilesPayload): Promi
},
});
await logger?.info(`Request ${requestId} completed successfully - status: downloaded`, {
logger.info(`Request ${requestId} completed successfully - status: downloaded`, {
success: true,
message: 'Files organized successfully',
requestId,
@@ -128,13 +127,13 @@ export async function processOrganizeFiles(payload: OrganizeFilesPayload): Promi
// Trigger scan (implementation is backend-specific)
await libraryService.triggerLibraryScan(libraryId);
await logger?.info(
logger.info(
`Triggered ${backendMode} filesystem scan for library ${libraryId}`
);
} catch (error) {
// Log error but don't fail the job
await logger?.error(
logger.error(
`Failed to trigger filesystem scan: ${error instanceof Error ? error.message : 'Unknown error'}`,
{
error: error instanceof Error ? error.stack : undefined,
@@ -144,7 +143,7 @@ export async function processOrganizeFiles(payload: OrganizeFilesPayload): Promi
// Continue - scheduled scans will eventually detect the book
}
} else {
await logger?.info(
logger.info(
`${backendMode} filesystem scan trigger disabled (relying on filesystem watcher)`
);
}
@@ -161,7 +160,7 @@ export async function processOrganizeFiles(payload: OrganizeFilesPayload): Promi
errors: result.errors,
};
} catch (error) {
await logger?.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
const errorMessage = error instanceof Error ? error.message : 'File organization failed';
@@ -191,7 +190,7 @@ export async function processOrganizeFiles(payload: OrganizeFilesPayload): Promi
if (newAttempts < currentRequest.maxImportRetries) {
// Still have retries left - queue for re-import
await logger?.warn(`Retryable error for request ${requestId}, queueing for retry (attempt ${newAttempts}/${currentRequest.maxImportRetries})`);
logger.warn(`Retryable error for request ${requestId}, queueing for retry (attempt ${newAttempts}/${currentRequest.maxImportRetries})`);
await prisma.request.update({
where: { id: requestId },
@@ -213,7 +212,7 @@ export async function processOrganizeFiles(payload: OrganizeFilesPayload): Promi
};
} else {
// Max retries exceeded - move to warn status
await logger?.warn(`Max retries (${currentRequest.maxImportRetries}) exceeded for request ${requestId}, moving to warn status`);
logger.warn(`Max retries (${currentRequest.maxImportRetries}) exceeded for request ${requestId}, moving to warn status`);
await prisma.request.update({
where: { id: requestId },
src/lib/processors/plex-recently-added.processor.ts
+14 -14
View File
@@ -6,7 +6,7 @@
*/
import { prisma } from '../db';
import { createJobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
import { getLibraryService } from '../services/library';
export interface PlexRecentlyAddedPayload {
@@ -16,14 +16,14 @@ export interface PlexRecentlyAddedPayload {
export async function processPlexRecentlyAddedCheck(payload: PlexRecentlyAddedPayload): Promise<any> {
const { jobId, scheduledJobId } = payload;
const logger = jobId ? createJobLogger(jobId, 'RecentlyAdded') : null;
const logger = RMABLogger.forJob(jobId, 'RecentlyAdded');
const { getConfigService } = await import('../services/config.service');
const configService = getConfigService();
// Get backend mode
const backendMode = await configService.getBackendMode();
await logger?.info(`Backend mode: ${backendMode}`);
logger.info(`Backend mode: ${backendMode}`);
// Validate configuration based on backend mode
if (backendMode === 'audiobookshelf') {
@@ -40,7 +40,7 @@ export async function processPlexRecentlyAddedCheck(payload: PlexRecentlyAddedPa
if (missingFields.length > 0) {
const errorMsg = `Audiobookshelf is not configured. Missing: ${missingFields.join(', ')}`;
await logger?.warn(errorMsg);
logger.warn(errorMsg);
return { success: false, message: errorMsg, skipped: true };
}
} else {
@@ -57,12 +57,12 @@ export async function processPlexRecentlyAddedCheck(payload: PlexRecentlyAddedPa
if (missingFields.length > 0) {
const errorMsg = `Plex is not configured. Missing: ${missingFields.join(', ')}`;
await logger?.warn(errorMsg);
logger.warn(errorMsg);
return { success: false, message: errorMsg, skipped: true };
}
}
await logger?.info(`Starting recently added check...`);
logger.info(`Starting recently added check...`);
// Get library service (automatically selects Plex or Audiobookshelf)
const libraryService = await getLibraryService();
@@ -76,7 +76,7 @@ export async function processPlexRecentlyAddedCheck(payload: PlexRecentlyAddedPa
// Fetch top 10 recently added items using abstraction layer
const recentItems = await libraryService.getRecentlyAdded(libraryId!, 10);
await logger?.info(`Found ${recentItems.length} recently added items`);
logger.info(`Found ${recentItems.length} recently added items`);
if (recentItems.length === 0) {
return { success: true, message: 'No recent items', newCount: 0, updatedCount: 0, matchedDownloads: 0 };
@@ -112,7 +112,7 @@ export async function processPlexRecentlyAddedCheck(payload: PlexRecentlyAddedPa
},
});
newCount++;
await logger?.info(`New item added: ${item.title} by ${item.author}`);
logger.info(`New item added: ${item.title} by ${item.author}`);
} else {
await prisma.plexLibrary.update({
where: { plexGuid: item.externalId },
@@ -144,7 +144,7 @@ export async function processPlexRecentlyAddedCheck(payload: PlexRecentlyAddedPa
});
if (downloadedRequests.length > 0) {
await logger?.info(`Checking ${downloadedRequests.length} downloaded requests for matches`);
logger.info(`Checking ${downloadedRequests.length} downloaded requests for matches`);
const { findPlexMatch } = await import('../utils/audiobook-matcher');
@@ -159,7 +159,7 @@ export async function processPlexRecentlyAddedCheck(payload: PlexRecentlyAddedPa
});
if (match) {
await logger?.info(`Match found: "${audiobook.title}" → "${match.title}"`);
logger.info(`Match found: "${audiobook.title}" → "${match.title}"`);
// Update audiobook with matched library item ID
const updateData: any = { updatedAt: new Date() };
@@ -187,18 +187,18 @@ export async function processPlexRecentlyAddedCheck(payload: PlexRecentlyAddedPa
const itemId = match.plexGuid; // plexGuid contains the Audiobookshelf item ID
const asin = audiobook.audibleAsin || undefined;
const matchInfo = asin ? ` with ASIN ${asin}` : '';
await logger?.info(`Triggering metadata match for matched item: ${itemId}${matchInfo}`);
logger.info(`Triggering metadata match for matched item: ${itemId}${matchInfo}`);
const { triggerABSItemMatch } = await import('../services/audiobookshelf/api');
await triggerABSItemMatch(itemId, asin);
}
}
} catch (error) {
await logger?.error(`Failed to match request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Failed to match request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
}
}
}
await logger?.info(`Complete: ${newCount} new, ${updatedCount} updated, ${matchedDownloads} matched downloads`);
logger.info(`Complete: ${newCount} new, ${updatedCount} updated, ${matchedDownloads} matched downloads`);
return {
success: true,
@@ -209,7 +209,7 @@ export async function processPlexRecentlyAddedCheck(payload: PlexRecentlyAddedPa
matchedDownloads,
};
} catch (error) {
await logger?.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
throw error;
}
}
src/lib/processors/retry-failed-imports.processor.ts
+23 -23
View File
@@ -6,7 +6,7 @@
*/
import { prisma } from '../db';
import { createJobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
import { getJobQueueService } from '../services/job-queue.service';
import { getConfigService } from '../services/config.service';
import { PathMapper } from '../utils/path-mapper';
@@ -18,9 +18,9 @@ export interface RetryFailedImportsPayload {
export async function processRetryFailedImports(payload: RetryFailedImportsPayload): Promise<any> {
const { jobId, scheduledJobId } = payload;
const logger = jobId ? createJobLogger(jobId, 'RetryFailedImports') : null;
const logger = RMABLogger.forJob(jobId, 'RetryFailedImports');
await logger?.info('Starting retry job for requests awaiting import...');
logger.info('Starting retry job for requests awaiting import...');
try {
// Load path mapping configuration once
@@ -54,7 +54,7 @@ export async function processRetryFailedImports(payload: RetryFailedImportsPaylo
take: 50, // Limit to 50 requests per run
});
await logger?.info(`Found ${requests.length} requests awaiting import`);
logger.info(`Found ${requests.length} requests awaiting import`);
if (requests.length === 0) {
return {
@@ -75,7 +75,7 @@ export async function processRetryFailedImports(payload: RetryFailedImportsPaylo
const downloadHistory = request.downloadHistory[0];
if (!downloadHistory) {
await logger?.warn(`No download history found for request ${request.id}, skipping`);
logger.warn(`No download history found for request ${request.id}, skipping`);
skipped++;
continue;
}
@@ -91,16 +91,16 @@ export async function processRetryFailedImports(payload: RetryFailedImportsPaylo
const torrent = await qbt.getTorrent(downloadHistory.torrentHash);
const qbPath = `${torrent.save_path}/${torrent.name}`;
downloadPath = PathMapper.transform(qbPath, mappingConfig);
await logger?.info(
logger.info(
`Got download path from qBittorrent for request ${request.id}: ${qbPath}` +
(downloadPath !== qbPath ? ` → ${downloadPath} (mapped)` : '')
);
} catch (qbtError) {
// Torrent not found in qBittorrent - try to construct path from config
await logger?.warn(`Torrent not found in qBittorrent for request ${request.id}, falling back to configured path`);
logger.warn(`Torrent not found in qBittorrent for request ${request.id}, falling back to configured path`);
if (!downloadHistory.torrentName) {
await logger?.warn(`No torrent name stored for request ${request.id}, cannot construct fallback path, skipping`);
logger.warn(`No torrent name stored for request ${request.id}, cannot construct fallback path, skipping`);
skipped++;
continue;
}
@@ -108,14 +108,14 @@ export async function processRetryFailedImports(payload: RetryFailedImportsPaylo
const downloadDir = await configService.get('download_dir');
if (!downloadDir) {
await logger?.error(`download_dir not configured, cannot retry request ${request.id}, skipping`);
logger.error(`download_dir not configured, cannot retry request ${request.id}, skipping`);
skipped++;
continue;
}
const fallbackPath = `${downloadDir}/${downloadHistory.torrentName}`;
downloadPath = PathMapper.transform(fallbackPath, mappingConfig);
await logger?.info(
logger.info(
`Using fallback download path for request ${request.id}: ${fallbackPath}` +
(downloadPath !== fallbackPath ? ` → ${downloadPath} (mapped)` : '')
);
@@ -128,15 +128,15 @@ export async function processRetryFailedImports(payload: RetryFailedImportsPaylo
const nzbInfo = await sabnzbd.getNZB(downloadHistory.nzbId);
if (nzbInfo && nzbInfo.downloadPath) {
downloadPath = PathMapper.transform(nzbInfo.downloadPath, mappingConfig);
await logger?.info(
logger.info(
`Got download path from SABnzbd for request ${request.id}: ${nzbInfo.downloadPath}` +
(downloadPath !== nzbInfo.downloadPath ? ` → ${downloadPath} (mapped)` : '')
);
} else {
await logger?.warn(`NZB ${downloadHistory.nzbId} not found or has no download path for request ${request.id}, falling back to configured path`);
logger.warn(`NZB ${downloadHistory.nzbId} not found or has no download path for request ${request.id}, falling back to configured path`);
if (!downloadHistory.torrentName) {
await logger?.warn(`No name stored for request ${request.id}, cannot construct fallback path, skipping`);
logger.warn(`No name stored for request ${request.id}, cannot construct fallback path, skipping`);
skipped++;
continue;
}
@@ -144,27 +144,27 @@ export async function processRetryFailedImports(payload: RetryFailedImportsPaylo
const downloadDir = await configService.get('download_dir');
if (!downloadDir) {
await logger?.error(`download_dir not configured, cannot retry request ${request.id}, skipping`);
logger.error(`download_dir not configured, cannot retry request ${request.id}, skipping`);
skipped++;
continue;
}
const fallbackPath = `${downloadDir}/${downloadHistory.torrentName}`;
downloadPath = PathMapper.transform(fallbackPath, mappingConfig);
await logger?.info(
logger.info(
`Using fallback download path for request ${request.id}: ${fallbackPath}` +
(downloadPath !== fallbackPath ? ` → ${downloadPath} (mapped)` : '')
);
}
} catch (sabnzbdError) {
await logger?.warn(`SABnzbd error for request ${request.id}: ${sabnzbdError instanceof Error ? sabnzbdError.message : 'Unknown error'}, skipping`);
logger.warn(`SABnzbd error for request ${request.id}: ${sabnzbdError instanceof Error ? sabnzbdError.message : 'Unknown error'}, skipping`);
skipped++;
continue;
}
} else {
// No download client ID - use fallback path
if (!downloadHistory.torrentName) {
await logger?.warn(`No download client ID or name for request ${request.id}, skipping`);
logger.warn(`No download client ID or name for request ${request.id}, skipping`);
skipped++;
continue;
}
@@ -172,14 +172,14 @@ export async function processRetryFailedImports(payload: RetryFailedImportsPaylo
const downloadDir = await configService.get('download_dir');
if (!downloadDir) {
await logger?.error(`download_dir not configured, cannot retry request ${request.id}, skipping`);
logger.error(`download_dir not configured, cannot retry request ${request.id}, skipping`);
skipped++;
continue;
}
const configuredPath = `${downloadDir}/${downloadHistory.torrentName}`;
downloadPath = PathMapper.transform(configuredPath, mappingConfig);
await logger?.info(
logger.info(
`Using configured download path for request ${request.id}: ${configuredPath}` +
(downloadPath !== configuredPath ? ` → ${downloadPath} (mapped)` : '')
);
@@ -191,14 +191,14 @@ export async function processRetryFailedImports(payload: RetryFailedImportsPaylo
downloadPath
);
triggered++;
await logger?.info(`Triggered organize job for request ${request.id}: ${request.audiobook.title}`);
logger.info(`Triggered organize job for request ${request.id}: ${request.audiobook.title}`);
} catch (error) {
await logger?.error(`Failed to trigger organize for request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Failed to trigger organize for request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
skipped++;
}
}
await logger?.info(`Triggered ${triggered}/${requests.length} organize jobs (${skipped} skipped)`);
logger.info(`Triggered ${triggered}/${requests.length} organize jobs (${skipped} skipped)`);
return {
success: true,
@@ -208,7 +208,7 @@ export async function processRetryFailedImports(payload: RetryFailedImportsPaylo
skipped,
};
} catch (error) {
await logger?.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
throw error;
}
}
src/lib/processors/retry-missing-torrents.processor.ts
+8 -8
View File
@@ -6,7 +6,7 @@
*/
import { prisma } from '../db';
import { createJobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
import { getJobQueueService } from '../services/job-queue.service';
export interface RetryMissingTorrentsPayload {
@@ -16,9 +16,9 @@ export interface RetryMissingTorrentsPayload {
export async function processRetryMissingTorrents(payload: RetryMissingTorrentsPayload): Promise<any> {
const { jobId, scheduledJobId } = payload;
const logger = jobId ? createJobLogger(jobId, 'RetryMissingTorrents') : null;
const logger = RMABLogger.forJob(jobId, 'RetryMissingTorrents');
await logger?.info('Starting retry job for requests awaiting search...');
logger.info('Starting retry job for requests awaiting search...');
try {
// Find all active requests in awaiting_search status
@@ -33,7 +33,7 @@ export async function processRetryMissingTorrents(payload: RetryMissingTorrentsP
take: 50, // Limit to 50 requests per run
});
await logger?.info(`Found ${requests.length} requests awaiting search`);
logger.info(`Found ${requests.length} requests awaiting search`);
if (requests.length === 0) {
return {
@@ -55,13 +55,13 @@ export async function processRetryMissingTorrents(payload: RetryMissingTorrentsP
author: request.audiobook.author,
});
triggered++;
await logger?.info(`Triggered search for request ${request.id}: ${request.audiobook.title}`);
logger.info(`Triggered search for request ${request.id}: ${request.audiobook.title}`);
} catch (error) {
await logger?.error(`Failed to trigger search for request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Failed to trigger search for request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
}
}
await logger?.info(`Triggered ${triggered}/${requests.length} search jobs`);
logger.info(`Triggered ${triggered}/${requests.length} search jobs`);
return {
success: true,
@@ -70,7 +70,7 @@ export async function processRetryMissingTorrents(payload: RetryMissingTorrentsP
triggered,
};
} catch (error) {
await logger?.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
throw error;
}
}
src/lib/processors/scan-plex.processor.ts
+28 -28
View File
@@ -10,7 +10,7 @@ import { ScanPlexPayload } from '../services/job-queue.service';
import { prisma } from '../db';
import { getLibraryService } from '../services/library';
import { getConfigService } from '../services/config.service';
import { createJobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
/**
* Process library scan job
@@ -19,9 +19,9 @@ import { createJobLogger } from '../utils/job-logger';
export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
const { libraryId, partial, path, jobId } = payload;
const logger = jobId ? createJobLogger(jobId, 'ScanLibrary') : null;
const logger = RMABLogger.forJob(jobId, 'ScanLibrary');
await logger?.info(`Scanning library ${libraryId || 'default'}${partial ? ' (partial)' : ''}`);
logger.info(`Scanning library ${libraryId || 'default'}${partial ? ' (partial)' : ''}`);
try {
// 1. Get library service (automatically selects Plex or Audiobookshelf based on config)
@@ -29,7 +29,7 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
const configService = getConfigService();
const backendMode = await configService.getBackendMode();
await logger?.info(`Backend mode: ${backendMode}`);
logger.info(`Backend mode: ${backendMode}`);
// 2. Get configured library ID
let targetLibraryId = libraryId;
@@ -50,12 +50,12 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
}
}
await logger?.info(`Fetching content from library ${targetLibraryId}`);
logger.info(`Fetching content from library ${targetLibraryId}`);
// 3. Get all audiobooks from library using abstraction layer
const libraryItems = await libraryService.getLibraryItems(targetLibraryId);
await logger?.info(`Found ${libraryItems.length} items in library`);
logger.info(`Found ${libraryItems.length} items in library`);
let newCount = 0;
let updatedCount = 0;
@@ -120,7 +120,7 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
});
newCount++;
await logger?.info(`Added new: "${item.title}" by ${item.author}`);
logger.info(`Added new: "${item.title}" by ${item.author}`);
results.push({
id: newLibraryItem.id,
@@ -130,16 +130,16 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
});
}
} catch (error) {
await logger?.error(`Failed to process "${item.title}": ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Failed to process "${item.title}": ${error instanceof Error ? error.message : 'Unknown error'}`);
skippedCount++;
}
}
await logger?.info(`Scan complete: ${libraryItems.length} items scanned, ${newCount} new, ${updatedCount} updated, ${skippedCount} skipped`);
logger.info(`Scan complete: ${libraryItems.length} items scanned, ${newCount} new, ${updatedCount} updated, ${skippedCount} skipped`);
// 5. Remove stale records from plex_library (items no longer in the actual library)
// This ensures the database is a fresh snapshot of the library state
await logger?.info(`Checking for stale library records...`);
logger.info(`Checking for stale library records...`);
const scannedPlexGuids = libraryItems
.filter(item => item.externalId)
@@ -163,7 +163,7 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
});
if (staleLibraryItems.length > 0) {
await logger?.info(`Found ${staleLibraryItems.length} stale library records to remove`);
logger.info(`Found ${staleLibraryItems.length} stale library records to remove`);
// For each stale library item, clean up references
for (const staleItem of staleLibraryItems) {
@@ -214,7 +214,7 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
}
}
await logger?.info(`Reset audiobook "${staleItem.title}" (no longer in library)`);
logger.info(`Reset audiobook "${staleItem.title}" (no longer in library)`);
}
// Delete the stale library record
@@ -224,21 +224,21 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
staleRemovedCount++;
} catch (error) {
await logger?.error(`Failed to remove stale library item "${staleItem.title}": ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Failed to remove stale library item "${staleItem.title}": ${error instanceof Error ? error.message : 'Unknown error'}`);
}
}
await logger?.info(`Removed ${staleRemovedCount} stale records, reset ${audiobooksReset} audiobooks and ${requestsReset} requests`);
logger.info(`Removed ${staleRemovedCount} stale records, reset ${audiobooksReset} audiobooks and ${requestsReset} requests`);
} else {
await logger?.info(`No stale library records found`);
logger.info(`No stale library records found`);
}
} else {
await logger?.warn(`Scan returned no items - skipping stale record cleanup to prevent data loss`);
logger.warn(`Scan returned no items - skipping stale record cleanup to prevent data loss`);
}
// 5b. Clean up orphaned audiobooks (audiobooks with plexGuid/absItemId that don't exist in plex_library)
// This handles cases where the library record was already deleted but audiobook record wasn't updated
await logger?.info(`Checking for orphaned audiobooks...`);
logger.info(`Checking for orphaned audiobooks...`);
const allPlexGuidsInLibrary = await prisma.plexLibrary.findMany({
select: { plexGuid: true },
@@ -277,7 +277,7 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
// This audiobook is orphaned - its library link points to nothing
try {
await logger?.info(`Found orphaned audiobook: "${audiobook.title}" (linked to non-existent library item)`);
logger.info(`Found orphaned audiobook: "${audiobook.title}" (linked to non-existent library item)`);
// Clear library linkage
await prisma.audiobook.update({
@@ -306,18 +306,18 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
}
}
} catch (error) {
await logger?.error(`Failed to reset orphaned audiobook "${audiobook.title}": ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Failed to reset orphaned audiobook "${audiobook.title}": ${error instanceof Error ? error.message : 'Unknown error'}`);
}
}
if (orphanedAudiobooksReset > 0) {
await logger?.info(`Reset ${orphanedAudiobooksReset} orphaned audiobooks and ${orphanedRequestsReset} requests`);
logger.info(`Reset ${orphanedAudiobooksReset} orphaned audiobooks and ${orphanedRequestsReset} requests`);
} else {
await logger?.info(`No orphaned audiobooks found`);
logger.info(`No orphaned audiobooks found`);
}
// 6. Match downloaded requests against library
await logger?.info(`Checking for downloaded requests to match...`);
logger.info(`Checking for downloaded requests to match...`);
const downloadedRequests = await prisma.request.findMany({
where: {
status: 'downloaded',
@@ -327,7 +327,7 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
take: 50, // Limit to prevent overwhelming
});
await logger?.info(`Found ${downloadedRequests.length} downloaded requests to match`);
logger.info(`Found ${downloadedRequests.length} downloaded requests to match`);
let matchedCount = 0;
const { findPlexMatch } = await import('../utils/audiobook-matcher');
@@ -346,7 +346,7 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
});
if (match) {
await logger?.info(`Match found! "${audiobook.title}" -> "${match.title}"`);
logger.info(`Match found! "${audiobook.title}" -> "${match.title}"`);
// Update audiobook with matched library item ID (plexGuid or abs_item_id)
const updateData: any = { updatedAt: new Date() };
@@ -379,17 +379,17 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
const itemId = match.plexGuid; // plexGuid contains the Audiobookshelf item ID
const asin = audiobook.audibleAsin || undefined;
const matchInfo = asin ? ` with ASIN ${asin}` : '';
await logger?.info(`Triggering metadata match for matched item: ${itemId}${matchInfo}`);
logger.info(`Triggering metadata match for matched item: ${itemId}${matchInfo}`);
const { triggerABSItemMatch } = await import('../services/audiobookshelf/api');
await triggerABSItemMatch(itemId, asin);
}
}
} catch (error) {
await logger?.error(`Failed to match request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Failed to match request ${request.id}: ${error instanceof Error ? error.message : 'Unknown error'}`);
}
}
await logger?.info(`Matched ${matchedCount}/${downloadedRequests.length} downloaded requests`, {
logger.info(`Matched ${matchedCount}/${downloadedRequests.length} downloaded requests`, {
totalScanned: libraryItems.length,
newCount,
updatedCount,
@@ -420,7 +420,7 @@ export async function processScanPlex(payload: ScanPlexPayload): Promise<any> {
matchedDownloads: matchedCount,
};
} catch (error) {
await logger?.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
throw error;
}
}
src/lib/processors/search-indexers.processor.ts
+32 -32
View File
@@ -7,7 +7,7 @@ import { SearchIndexersPayload, getJobQueueService } from '../services/job-queue
import { prisma } from '../db';
import { getProwlarrService } from '../integrations/prowlarr.service';
import { getRankingAlgorithm } from '../utils/ranking-algorithm';
import { createJobLogger } from '../utils/job-logger';
import { RMABLogger } from '../utils/logger';
/**
* Process search indexers job
@@ -16,9 +16,9 @@ import { createJobLogger } from '../utils/job-logger';
export async function processSearchIndexers(payload: SearchIndexersPayload): Promise<any> {
const { requestId, audiobook, jobId } = payload;
const logger = jobId ? createJobLogger(jobId, 'SearchIndexers') : null;
const logger = RMABLogger.forJob(jobId, 'SearchIndexers');
await logger?.info(`Processing request ${requestId} for "${audiobook.title}"`);
logger.info(`Processing request ${requestId} for "${audiobook.title}"`);
try {
// Update request status to searching
@@ -56,7 +56,7 @@ export async function processSearchIndexers(payload: SearchIndexersPayload): Pro
const flagConfigStr = await configService.get('indexer_flag_config');
const flagConfigs = flagConfigStr ? JSON.parse(flagConfigStr) : [];
await logger?.info(`Searching ${enabledIndexerIds.length} enabled indexers`);
logger.info(`Searching ${enabledIndexerIds.length} enabled indexers`);
// Get Prowlarr service
const prowlarr = await getProwlarrService();
@@ -64,7 +64,7 @@ export async function processSearchIndexers(payload: SearchIndexersPayload): Pro
// Build search query (title only - cast wide net, let ranking filter)
const searchQuery = audiobook.title;
await logger?.info(`Searching for: "${searchQuery}"`);
logger.info(`Searching for: "${searchQuery}"`);
// Search indexers - ONLY enabled ones
const searchResults = await prowlarr.search(searchQuery, {
@@ -74,11 +74,11 @@ export async function processSearchIndexers(payload: SearchIndexersPayload): Pro
indexerIds: enabledIndexerIds, // Filter by enabled indexers
});
await logger?.info(`Found ${searchResults.length} raw results`);
logger.info(`Found ${searchResults.length} raw results`);
if (searchResults.length === 0) {
// No results found - queue for re-search instead of failing
await logger?.warn(`No torrents found for request ${requestId}, marking as awaiting_search`);
logger.warn(`No torrents found for request ${requestId}, marking as awaiting_search`);
await prisma.request.update({
where: { id: requestId },
@@ -117,14 +117,14 @@ export async function processSearchIndexers(payload: SearchIndexersPayload): Pro
result.score >= 50 && result.finalScore < 50
).length;
await logger?.info(`Ranked ${rankedResults.length} results, ${filteredResults.length} above threshold (50/100 base + final)`);
logger.info(`Ranked ${rankedResults.length} results, ${filteredResults.length} above threshold (50/100 base + final)`);
if (disqualifiedByNegativeBonus > 0) {
await logger?.info(`${disqualifiedByNegativeBonus} torrents disqualified by negative flag bonuses`);
logger.info(`${disqualifiedByNegativeBonus} torrents disqualified by negative flag bonuses`);
}
if (filteredResults.length === 0) {
// No quality results found - queue for re-search instead of failing
await logger?.warn(`No quality matches found for request ${requestId} (all below 50/100), marking as awaiting_search`);
logger.warn(`No quality matches found for request ${requestId} (all below 50/100), marking as awaiting_search`);
await prisma.request.update({
where: { id: requestId },
@@ -148,38 +148,38 @@ export async function processSearchIndexers(payload: SearchIndexersPayload): Pro
// Log top 3 results with detailed breakdown
const top3 = filteredResults.slice(0, 3);
await logger?.info(`==================== RANKING DEBUG ====================`);
await logger?.info(`Requested Title: "${audiobook.title}"`);
await logger?.info(`Requested Author: "${audiobook.author}"`);
await logger?.info(`Top ${top3.length} results (out of ${filteredResults.length} above threshold):`);
await logger?.info(`--------------------------------------------------------`);
logger.info(`==================== RANKING DEBUG ====================`);
logger.info(`Requested Title: "${audiobook.title}"`);
logger.info(`Requested Author: "${audiobook.author}"`);
logger.info(`Top ${top3.length} results (out of ${filteredResults.length} above threshold):`);
logger.info(`--------------------------------------------------------`);
for (let i = 0; i < top3.length; i++) {
const result = top3[i];
await logger?.info(`${i + 1}. "${result.title}"`);
await logger?.info(` Indexer: ${result.indexer}${result.indexerId ? ` (ID: ${result.indexerId})` : ''}`);
await logger?.info(``);
await logger?.info(` Base Score: ${result.score.toFixed(1)}/100`);
await logger?.info(` - Title/Author Match: ${result.breakdown.matchScore.toFixed(1)}/60`);
await logger?.info(` - Format Quality: ${result.breakdown.formatScore.toFixed(1)}/25 (${result.format || 'unknown'})`);
await logger?.info(` - Seeder Count: ${result.breakdown.seederScore.toFixed(1)}/15 (${result.seeders !== undefined ? result.seeders + ' seeders' : 'N/A for Usenet'})`);
await logger?.info(``);
await logger?.info(` Bonus Points: +${result.bonusPoints.toFixed(1)}`);
logger.info(`${i + 1}. "${result.title}"`);
logger.info(` Indexer: ${result.indexer}${result.indexerId ? ` (ID: ${result.indexerId})` : ''}`);
logger.info(``);
logger.info(` Base Score: ${result.score.toFixed(1)}/100`);
logger.info(` - Title/Author Match: ${result.breakdown.matchScore.toFixed(1)}/60`);
logger.info(` - Format Quality: ${result.breakdown.formatScore.toFixed(1)}/25 (${result.format || 'unknown'})`);
logger.info(` - Seeder Count: ${result.breakdown.seederScore.toFixed(1)}/15 (${result.seeders !== undefined ? result.seeders + ' seeders' : 'N/A for Usenet'})`);
logger.info(``);
logger.info(` Bonus Points: +${result.bonusPoints.toFixed(1)}`);
if (result.bonusModifiers.length > 0) {
for (const mod of result.bonusModifiers) {
await logger?.info(` - ${mod.reason}: +${mod.points.toFixed(1)}`);
logger.info(` - ${mod.reason}: +${mod.points.toFixed(1)}`);
}
}
await logger?.info(``);
await logger?.info(` Final Score: ${result.finalScore.toFixed(1)}`);
logger.info(``);
logger.info(` Final Score: ${result.finalScore.toFixed(1)}`);
if (result.breakdown.notes.length > 0) {
await logger?.info(` Notes: ${result.breakdown.notes.join(', ')}`);
logger.info(` Notes: ${result.breakdown.notes.join(', ')}`);
}
if (i < top3.length - 1) {
await logger?.info(`--------------------------------------------------------`);
logger.info(`--------------------------------------------------------`);
}
}
await logger?.info(`========================================================`);
await logger?.info(`Selected best result: ${bestResult.title} (final score: ${bestResult.finalScore.toFixed(1)})`);
logger.info(`========================================================`);
logger.info(`Selected best result: ${bestResult.title} (final score: ${bestResult.finalScore.toFixed(1)})`);
// Trigger download job with best result
const jobQueue = getJobQueueService();
@@ -202,7 +202,7 @@ export async function processSearchIndexers(payload: SearchIndexersPayload): Pro
},
};
} catch (error) {
await logger?.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
logger.error(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`);
await prisma.request.update({
where: { id: requestId },