From 4847eb52413a1f428bca7dab042e13a955c5b528 Mon Sep 17 00:00:00 2001 From: 4gray <4gray@users.noreply.github.com> Date: Wed, 27 May 2026 22:05:40 +0200 Subject: [PATCH] refactor(electron): split epg event services Split EPG IPC orchestration, worker lifecycle, and query logic into focused services. Harden clear-worker lifecycle after review with timeout/exit handling and regression coverage. --- CLAUDE.md | 4 +- .../src/app/events/epg-query.service.ts | 400 ++++++++++ .../src/app/events/epg-worker.service.ts | 330 ++++++++ .../src/app/events/epg.events.spec.ts | 29 + .../src/app/events/epg.events.ts | 717 +----------------- docs/architecture/sqlite-db-worker.md | 24 +- 6 files changed, 800 insertions(+), 704 deletions(-) create mode 100644 apps/electron-backend/src/app/events/epg-query.service.ts create mode 100644 apps/electron-backend/src/app/events/epg-worker.service.ts diff --git a/CLAUDE.md b/CLAUDE.md index 757c186cb..10c11a606 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -546,7 +546,7 @@ This project uses modern Angular signal-based APIs and patterns. **ALWAYS** use - **Event handlers**: `apps/electron-backend/src/app/events/` - `database.events.ts` - Database CRUD operations - `playlist.events.ts` - Playlist import/update - - `epg.events.ts` - EPG fetch and parsing (uses worker) + - `epg.events.ts` - EPG IPC registration and freshness/fetch orchestration; worker lifecycle lives in `epg-worker.service.ts`, DB lookups in `epg-query.service.ts` - `xtream.events.ts` - Xtream Codes API - `stalker.events.ts` - Stalker portal API - `player.events.ts` - External player IPC registration; MPV/VLC lifecycle logic lives in `mpv-session.service.ts`, `vlc-session.service.ts`, and shared `external-player-*` helpers @@ -555,7 +555,7 @@ This project uses modern Angular signal-based APIs and patterns. **ALWAYS** use **Workers**: -- EPG parsing runs in worker thread: `apps/electron-backend/src/app/workers/epg-parser.worker.ts` +- EPG parsing runs in worker thread: `apps/electron-backend/src/app/workers/epg-parser.worker.ts`; main-process worker lifecycle is coordinated from `apps/electron-backend/src/app/events/epg-worker.service.ts` ### Key Features diff --git a/apps/electron-backend/src/app/events/epg-query.service.ts b/apps/electron-backend/src/app/events/epg-query.service.ts new file mode 100644 index 000000000..a28bf823a --- /dev/null +++ b/apps/electron-backend/src/app/events/epg-query.service.ts @@ -0,0 +1,400 @@ +import { and, eq, gte, inArray, lte, sql } from 'drizzle-orm'; +import { EpgChannelMetadata, EpgProgram } from '@iptvnator/shared/interfaces'; +import { getDatabase } from '../database/connection'; +import * as schema from '../database/schema'; + +interface EpgProgramRow { + id: number; + channelId: string; + start: string; + stop: string; + title: string; + description: string | null; + category: string | null; + iconUrl: string | null; + rating: string | null; + episodeNum: string | null; +} + +export class EpgQueryService { + constructor(private readonly loggerLabel = '[EPG Events]') {} + + async getChannelPrograms(channelId: string): Promise { + try { + const db = await getDatabase(); + const trimmedChannelId = channelId.trim(); + + if (!trimmedChannelId) { + return []; + } + + let results = await db + .select() + .from(schema.epgPrograms) + .where(eq(schema.epgPrograms.channelId, trimmedChannelId)) + .orderBy(schema.epgPrograms.start) + .limit(500); + + if (results.length > 0) { + return results + .map(this.transformDbRowToEpgProgram) + .filter(this.isValidEpgProgram); + } + + let channel = await db + .select() + .from(schema.epgChannels) + .where( + sql`${schema.epgChannels.id} = ${trimmedChannelId} COLLATE NOCASE` + ) + .limit(1); + + if (channel.length > 0) { + results = await db + .select() + .from(schema.epgPrograms) + .where(eq(schema.epgPrograms.channelId, channel[0].id)) + .orderBy(schema.epgPrograms.start) + .limit(500); + + if (results.length > 0) { + return results + .map(this.transformDbRowToEpgProgram) + .filter(this.isValidEpgProgram); + } + } + + channel = await db + .select() + .from(schema.epgChannels) + .where(eq(schema.epgChannels.displayName, trimmedChannelId)) + .limit(1); + + if (channel.length === 0) { + channel = await db + .select() + .from(schema.epgChannels) + .where( + sql`${schema.epgChannels.displayName} = ${trimmedChannelId} COLLATE NOCASE` + ) + .limit(1); + } + + if (channel.length > 0) { + results = await db + .select() + .from(schema.epgPrograms) + .where(eq(schema.epgPrograms.channelId, channel[0].id)) + .orderBy(schema.epgPrograms.start) + .limit(500); + + return results + .map(this.transformDbRowToEpgProgram) + .filter(this.isValidEpgProgram); + } + + return []; + } catch (error) { + console.error( + this.loggerLabel, + 'Error getting channel programs:', + error + ); + return []; + } + } + + async getCurrentProgramsBatch( + channelIds: string[] + ): Promise> { + const result: Record = {}; + if (!Array.isArray(channelIds) || channelIds.length === 0) { + return result; + } + + const validIds = Array.from( + new Set( + channelIds + .map((id) => id?.trim()) + .filter((id): id is string => Boolean(id)) + ) + ); + if (validIds.length === 0) { + return result; + } + + try { + const db = await getDatabase(); + const now = new Date().toISOString(); + + const rows = await db + .select() + .from(schema.epgPrograms) + .where( + and( + inArray(schema.epgPrograms.channelId, validIds), + lte(schema.epgPrograms.start, now), + gte(schema.epgPrograms.stop, now) + ) + ); + + for (const row of rows) { + if (!result[row.channelId]) { + const program = this.transformDbRowToEpgProgram(row); + if (this.isValidEpgProgram(program)) { + result[row.channelId] = program; + } + } + } + + const unmatchedIds = validIds.filter((id) => !(id in result)); + for (const channelId of unmatchedIds) { + result[channelId] = null; + + let channel = await db + .select() + .from(schema.epgChannels) + .where( + sql`${schema.epgChannels.id} = ${channelId} COLLATE NOCASE` + ) + .limit(1); + + if (channel.length === 0) { + channel = await db + .select() + .from(schema.epgChannels) + .where( + sql`${schema.epgChannels.displayName} = ${channelId} COLLATE NOCASE` + ) + .limit(1); + } + + if (channel.length === 0) { + continue; + } + + const programRows = await db + .select() + .from(schema.epgPrograms) + .where( + and( + eq(schema.epgPrograms.channelId, channel[0].id), + lte(schema.epgPrograms.start, now), + gte(schema.epgPrograms.stop, now) + ) + ) + .limit(1); + + if (programRows.length > 0) { + const program = this.transformDbRowToEpgProgram( + programRows[0] + ); + if (this.isValidEpgProgram(program)) { + result[channelId] = program; + } + } + } + + return result; + } catch (error) { + console.error( + this.loggerLabel, + 'Error getting batch current programs:', + error + ); + return result; + } + } + + async getAllChannels(): Promise<{ + channels: Array<{ id: string; displayName: string }>; + programs: never[]; + }> { + try { + const db = await getDatabase(); + const channels = await db + .select({ + id: schema.epgChannels.id, + displayName: schema.epgChannels.displayName, + }) + .from(schema.epgChannels) + .orderBy(schema.epgChannels.displayName); + + return { channels, programs: [] }; + } catch (error) { + console.error( + this.loggerLabel, + 'Error getting all channels:', + error + ); + return { channels: [], programs: [] }; + } + } + + async getChannelMetadata( + channelIds: string[] + ): Promise> { + try { + const normalizedChannelIds = + this.normalizeChannelLookupKeys(channelIds); + + if (normalizedChannelIds.length === 0) { + return {}; + } + + const db = await getDatabase(); + const lowerKeys = Array.from( + new Set( + normalizedChannelIds.map((channelId) => + channelId.toLowerCase() + ) + ) + ); + const lowerKeyValues = lowerKeys.map((key) => sql`${key}`); + + const candidates = await db + .select({ + id: schema.epgChannels.id, + displayName: schema.epgChannels.displayName, + iconUrl: schema.epgChannels.iconUrl, + }) + .from(schema.epgChannels).where(sql` + LOWER(${schema.epgChannels.id}) IN (${sql.join(lowerKeyValues, sql`, `)}) + OR LOWER(${schema.epgChannels.displayName}) IN (${sql.join(lowerKeyValues, sql`, `)}) + `); + + return Object.fromEntries( + normalizedChannelIds.map((channelId) => [ + channelId, + this.resolveChannelMetadataCandidate(channelId, candidates), + ]) + ); + } catch (error) { + console.error( + this.loggerLabel, + 'Error getting channel metadata:', + error + ); + return {}; + } + } + + async getChannelsByRange( + skip: number, + limit: number + ): Promise< + Array<{ + id: string; + displayName: string; + iconUrl: string | null; + programs: EpgProgram[]; + }> + > { + try { + const db = await getDatabase(); + const channels = await db + .select({ + id: schema.epgChannels.id, + displayName: schema.epgChannels.displayName, + iconUrl: schema.epgChannels.iconUrl, + }) + .from(schema.epgChannels) + .orderBy(schema.epgChannels.displayName) + .offset(skip) + .limit(limit); + + return Promise.all( + channels.map(async (channel) => { + const programs = await db + .select() + .from(schema.epgPrograms) + .where(eq(schema.epgPrograms.channelId, channel.id)) + .orderBy(schema.epgPrograms.start); + + return { + ...channel, + programs: programs.map(this.transformDbRowToEpgProgram), + }; + }) + ); + } catch (error) { + console.error( + this.loggerLabel, + 'Error getting channels by range:', + error + ); + return []; + } + } + + private normalizeChannelLookupKeys(channelIds: string[]): string[] { + return Array.from( + new Set( + channelIds + .map((channelId) => channelId.trim()) + .filter((channelId) => channelId.length > 0) + ) + ); + } + + private resolveChannelMetadataCandidate( + channelId: string, + candidates: EpgChannelMetadata[] + ): EpgChannelMetadata | null { + const lowerChannelId = channelId.toLowerCase(); + + const exactIdMatch = + candidates.find((candidate) => candidate.id === channelId) ?? null; + if (exactIdMatch) { + return exactIdMatch; + } + + const caseInsensitiveIdMatch = + candidates.find( + (candidate) => candidate.id.toLowerCase() === lowerChannelId + ) ?? null; + if (caseInsensitiveIdMatch) { + return caseInsensitiveIdMatch; + } + + const exactDisplayNameMatch = + candidates.find( + (candidate) => candidate.displayName === channelId + ) ?? null; + if (exactDisplayNameMatch) { + return exactDisplayNameMatch; + } + + return ( + candidates.find( + (candidate) => + candidate.displayName.toLowerCase() === lowerChannelId + ) ?? null + ); + } + + private transformDbRowToEpgProgram(row: EpgProgramRow): EpgProgram { + return { + start: row.start, + stop: row.stop, + channel: row.channelId, + title: row.title, + desc: row.description, + category: row.category, + iconUrl: row.iconUrl, + rating: row.rating, + episodeNum: row.episodeNum, + }; + } + + private isValidEpgProgram(program: EpgProgram): boolean { + return Boolean( + program.start && + program.stop && + !Number.isNaN(new Date(program.start).getTime()) && + !Number.isNaN(new Date(program.stop).getTime()) + ); + } +} + +export const epgQueryService = new EpgQueryService(); diff --git a/apps/electron-backend/src/app/events/epg-worker.service.ts b/apps/electron-backend/src/app/events/epg-worker.service.ts new file mode 100644 index 000000000..cb44cd1ce --- /dev/null +++ b/apps/electron-backend/src/app/events/epg-worker.service.ts @@ -0,0 +1,330 @@ +import { app, BrowserWindow } from 'electron'; +import * as path from 'path'; +import { pathToFileURL } from 'url'; +import { Worker } from 'worker_threads'; +import { resolveWorkerRuntimeBootstrap } from '../workers/worker-runtime-paths'; + +export type EpgProgressStatus = 'queued' | 'loading' | 'complete' | 'error'; + +export interface EpgProgressStats { + totalChannels: number; + totalPrograms: number; +} + +interface EpgWorkerMessage { + type: string; + error?: string; + url?: string; + stats?: EpgProgressStats; +} + +export class EpgWorkerService { + private readonly fetchedUrls = new Set(); + private readonly workers = new Map(); + + constructor( + private readonly loggerLabel = '[EPG Events]', + private readonly fetchTimeoutMs = 5 * 60 * 1000 + ) {} + + hasFetchedUrl(url: string): boolean { + return this.fetchedUrls.has(url); + } + + markFetchedUrl(url: string): void { + this.fetchedUrls.add(url); + } + + deleteFetchedUrl(url: string): void { + this.fetchedUrls.delete(url); + } + + sendProgressToRenderer( + url: string, + status: EpgProgressStatus, + stats?: EpgProgressStats, + error?: string, + queuePosition?: number + ): void { + const windows = BrowserWindow.getAllWindows(); + windows.forEach((win) => { + win.webContents.send('EPG_PROGRESS_UPDATE', { + url, + status, + stats, + error, + queuePosition, + }); + }); + } + + async fetchEpgFromUrl(url: string): Promise { + if (this.fetchedUrls.has(url)) { + console.log( + this.loggerLabel, + `Skipping already fetched URL: ${url}` + ); + return; + } + + return new Promise((resolve, reject) => { + let worker: Worker; + try { + worker = this.createEpgWorker(); + } catch (error) { + console.error( + this.loggerLabel, + 'Failed to create worker:', + error + ); + reject(error); + return; + } + + this.workers.set(url, worker); + + // Guards against double-settling and keeps the outer loop moving + // when the worker dies or hangs without sending EPG_COMPLETE/EPG_ERROR. + let settled = false; + const settle = (fn: () => void) => { + if (settled) return; + settled = true; + clearTimeout(timeoutId); + fn(); + }; + + const timeoutId = setTimeout(() => { + const errorMessage = `EPG fetch timed out after ${ + this.fetchTimeoutMs / 1000 + }s`; + console.error(this.loggerLabel, `${errorMessage}: ${url}`); + this.sendProgressToRenderer( + url, + 'error', + undefined, + errorMessage + ); + worker.terminate(); + this.workers.delete(url); + settle(() => reject(new Error(errorMessage))); + }, this.fetchTimeoutMs); + + worker.on('message', async (message: EpgWorkerMessage) => { + try { + switch (message.type) { + case 'READY': + this.sendProgressToRenderer(url, 'loading', { + totalChannels: 0, + totalPrograms: 0, + }); + worker.postMessage({ type: 'FETCH_EPG', url }); + break; + + case 'EPG_PROGRESS': + if (message.stats) { + this.sendProgressToRenderer( + url, + 'loading', + message.stats + ); + } + break; + + case 'EPG_COMPLETE': + console.log( + this.loggerLabel, + `EPG parsing complete for ${url}:`, + message.stats + ); + this.sendProgressToRenderer( + url, + 'complete', + message.stats + ); + this.fetchedUrls.add(url); + worker.terminate(); + this.workers.delete(url); + settle(() => resolve()); + break; + + case 'EPG_ERROR': + console.error( + this.loggerLabel, + 'Worker error:', + message.error + ); + this.sendProgressToRenderer( + url, + 'error', + undefined, + message.error + ); + worker.terminate(); + this.workers.delete(url); + settle(() => + reject( + new Error(message.error || 'Unknown error') + ) + ); + break; + } + } catch (err) { + console.error( + this.loggerLabel, + 'Error handling message:', + err + ); + this.sendProgressToRenderer( + url, + 'error', + undefined, + err instanceof Error ? err.message : String(err) + ); + worker.terminate(); + this.workers.delete(url); + settle(() => reject(err)); + } + }); + + worker.on('error', (error) => { + console.error(this.loggerLabel, 'Worker error event:', error); + this.sendProgressToRenderer( + url, + 'error', + undefined, + error.message + ); + worker.terminate(); + this.workers.delete(url); + settle(() => reject(error)); + }); + + worker.on('exit', (code) => { + if (settled) return; + const errorMessage = `Worker exited unexpectedly (code ${code})`; + console.error(this.loggerLabel, `${errorMessage}: ${url}`); + this.sendProgressToRenderer( + url, + 'error', + undefined, + errorMessage + ); + this.workers.delete(url); + settle(() => reject(new Error(errorMessage))); + }); + }); + } + + async clearEpgData(): Promise { + return new Promise((resolve, reject) => { + let worker: Worker; + try { + worker = this.createEpgWorker(); + } catch (error) { + console.error( + this.loggerLabel, + 'Failed to create worker for clear:', + error + ); + reject(error); + return; + } + + let settled = false; + const settle = (fn: () => void) => { + if (settled) return; + settled = true; + clearTimeout(timeoutId); + fn(); + }; + + const timeoutId = setTimeout(() => { + const errorMessage = `EPG clear timed out after ${ + this.fetchTimeoutMs / 1000 + }s`; + console.error(this.loggerLabel, errorMessage); + settle(() => { + worker.terminate(); + reject(new Error(errorMessage)); + }); + }, this.fetchTimeoutMs); + + worker.on( + 'message', + (message: { type: string; error?: string }) => { + if (message.type === 'READY') { + worker.postMessage({ type: 'CLEAR_EPG' }); + } else if (message.type === 'CLEAR_COMPLETE') { + settle(() => { + console.log( + this.loggerLabel, + 'EPG data cleared via worker' + ); + this.fetchedUrls.clear(); + this.workers.forEach((runningWorker) => + runningWorker.terminate() + ); + this.workers.clear(); + worker.terminate(); + resolve(); + }); + } else if (message.type === 'EPG_ERROR') { + console.error( + this.loggerLabel, + 'Worker clear error:', + message.error + ); + settle(() => { + worker.terminate(); + reject(new Error(message.error || 'Clear failed')); + }); + } + } + ); + + worker.on('error', (error) => { + console.error( + this.loggerLabel, + 'Worker error during clear:', + error + ); + settle(() => { + worker.terminate(); + reject(error); + }); + }); + + worker.on('exit', (code) => { + if (settled) return; + const errorMessage = `Clear worker exited unexpectedly (code ${code})`; + console.error(this.loggerLabel, errorMessage); + settle(() => reject(new Error(errorMessage))); + }); + }); + } + + private createEpgWorker(): Worker { + const bootstrap = resolveWorkerRuntimeBootstrap({ + isPackaged: app.isPackaged, + workerFilename: 'epg-parser.worker.js', + developmentWorkerDir: path.join(__dirname, 'workers'), + resourcesPath: ( + process as NodeJS.Process & { resourcesPath?: string } + ).resourcesPath, + appPath: app.getAppPath(), + }); + + const workerURL = pathToFileURL(bootstrap.workerPath); + return new Worker(workerURL, { + resourceLimits: { + maxOldGenerationSizeMb: 4096, + maxYoungGenerationSizeMb: 512, + }, + workerData: { + nativeModuleSearchPaths: bootstrap.nativeModuleSearchPaths, + }, + }); + } +} + +export const epgWorkerService = new EpgWorkerService(); diff --git a/apps/electron-backend/src/app/events/epg.events.spec.ts b/apps/electron-backend/src/app/events/epg.events.spec.ts index ee0f84ee0..8d61ab6ac 100644 --- a/apps/electron-backend/src/app/events/epg.events.spec.ts +++ b/apps/electron-backend/src/app/events/epg.events.spec.ts @@ -45,6 +45,7 @@ jest.mock('../database/connection', () => ({ describe('EpgEvents', () => { let EpgEvents: typeof EpgEventsType; + let EpgWorkerService: typeof import('./epg-worker.service').EpgWorkerService; let consoleLogSpy: jest.SpyInstance; let consoleErrorSpy: jest.SpyInstance; @@ -63,9 +64,11 @@ describe('EpgEvents', () => { }); ({ default: EpgEvents } = await import('./epg.events')); + ({ EpgWorkerService } = await import('./epg-worker.service')); }); afterEach(() => { + jest.useRealTimers(); consoleLogSpy.mockRestore(); consoleErrorSpy.mockRestore(); getDatabase.mockReset(); @@ -131,6 +134,32 @@ describe('EpgEvents', () => { }); }); + it('rejects when the EPG clear worker exits before completion', async () => { + const workerService = new EpgWorkerService('[Test EPG]', 1000); + const clearPromise = workerService.clearEpgData(); + const worker = mockWorkerInstances[0]; + + worker.emit('exit', 0); + + await expect(clearPromise).rejects.toThrow( + 'Clear worker exited unexpectedly (code 0)' + ); + }); + + it('rejects when the EPG clear worker never responds', async () => { + jest.useFakeTimers(); + + const workerService = new EpgWorkerService('[Test EPG]', 25); + const clearPromise = workerService.clearEpgData(); + const worker = mockWorkerInstances[0]; + + worker.emit('message', { type: 'READY' }); + jest.advanceTimersByTime(25); + + await expect(clearPromise).rejects.toThrow('EPG clear timed out after'); + expect(worker.terminate).toHaveBeenCalled(); + }); + it('falls back to case-insensitive channel id lookup for EPG programs', async () => { const select = jest.fn(); const programLimitExact = jest.fn().mockResolvedValue([]); diff --git a/apps/electron-backend/src/app/events/epg.events.ts b/apps/electron-backend/src/app/events/epg.events.ts index adb08db1f..65781279a 100644 --- a/apps/electron-backend/src/app/events/epg.events.ts +++ b/apps/electron-backend/src/app/events/epg.events.ts @@ -1,79 +1,26 @@ -import { and, eq, gte, inArray, lte, sql } from 'drizzle-orm'; -import { app, BrowserWindow, ipcMain } from 'electron'; -import * as path from 'path'; +import { eq } from 'drizzle-orm'; +import { ipcMain } from 'electron'; import { EpgChannelMetadata, EpgProgram } from '@iptvnator/shared/interfaces'; -import { pathToFileURL } from 'url'; -import { Worker } from 'worker_threads'; import { getDatabase } from '../database/connection'; import * as schema from '../database/schema'; -import { resolveWorkerRuntimeBootstrap } from '../workers/worker-runtime-paths'; +import { epgQueryService } from './epg-query.service'; +import { epgWorkerService } from './epg-worker.service'; /** * EPG Events Handler - * Manages EPG data fetching and querying using worker threads. - * Database operations are performed in the worker thread to avoid blocking the main thread. + * Manages EPG IPC registration and delegates worker/query behavior. */ export default class EpgEvents { - private static fetchedUrls: Set = new Set(); - private static workers: Map = new Map(); private static readonly loggerLabel = '[EPG Events]'; - private static readonly FETCH_TIMEOUT_MS = 5 * 60 * 1000; - - private static createEpgWorker(): Worker { - const bootstrap = resolveWorkerRuntimeBootstrap({ - isPackaged: app.isPackaged, - workerFilename: 'epg-parser.worker.js', - developmentWorkerDir: path.join(__dirname, 'workers'), - resourcesPath: ( - process as NodeJS.Process & { resourcesPath?: string } - ).resourcesPath, - appPath: app.getAppPath(), - }); - - const workerURL = pathToFileURL(bootstrap.workerPath); - return new Worker(workerURL, { - resourceLimits: { - maxOldGenerationSizeMb: 4096, - maxYoungGenerationSizeMb: 512, - }, - workerData: { - nativeModuleSearchPaths: bootstrap.nativeModuleSearchPaths, - }, - }); - } - - /** - * Send EPG progress to all renderer windows - */ - private static sendProgressToRenderer( - url: string, - status: 'queued' | 'loading' | 'complete' | 'error', - stats?: { totalChannels: number; totalPrograms: number }, - error?: string, - queuePosition?: number - ): void { - const windows = BrowserWindow.getAllWindows(); - windows.forEach((win) => { - win.webContents.send('EPG_PROGRESS_UPDATE', { - url, - status, - stats, - error, - queuePosition, - }); - }); - } /** * Bootstrap EPG events */ static bootstrapEpgEvents(): Electron.IpcMain { - // Fetch EPG from URLs ipcMain.handle('FETCH_EPG', async (_event, args: { url: string[] }) => { return await this.handleFetchEpg(args.url); }); - // Get programs for a specific channel ipcMain.handle( 'GET_CHANNEL_PROGRAMS', async (_event, args: { channelId: string }) => { @@ -81,7 +28,6 @@ export default class EpgEvents { } ); - // Get current programs for many channels in a single batched query ipcMain.handle( 'GET_CURRENT_PROGRAMS_BATCH', async (_event, args: { channelIds: string[] }) => { @@ -89,7 +35,6 @@ export default class EpgEvents { } ); - // Get all channels from database ipcMain.handle('EPG_GET_CHANNELS', async () => { return this.handleGetAllChannels(); }); @@ -101,7 +46,6 @@ export default class EpgEvents { } ); - // Get channels by range (pagination) ipcMain.handle( 'EPG_GET_CHANNELS_BY_RANGE', async (_event, args: { skip: number; limit: number }) => { @@ -109,19 +53,16 @@ export default class EpgEvents { } ); - // Force fetch (ignore cache) ipcMain.handle('EPG_FORCE_FETCH', async (_event, url: string) => { - this.fetchedUrls.delete(url); + epgWorkerService.deleteFetchedUrl(url); return await this.handleFetchEpg([url]); }); - // Clear all EPG data ipcMain.handle('EPG_CLEAR_ALL', async () => { await this.clearEpgData(); return { success: true }; }); - // Check if EPG data for URLs is fresh (not stale) ipcMain.handle( 'EPG_CHECK_FRESHNESS', async ( @@ -172,7 +113,7 @@ export default class EpgEvents { if (isFresh) { freshUrls.push(url); - this.fetchedUrls.add(url); + epgWorkerService.markFetchedUrl(url); } else { staleUrls.push(url); } @@ -186,7 +127,6 @@ export default class EpgEvents { return { staleUrls: urls, freshUrls: [] }; } - // Log summary if (freshUrls.length > 0) { console.log( this.loggerLabel, @@ -217,7 +157,6 @@ export default class EpgEvents { return { success: false, message: 'No valid URLs provided' }; } - // Check which URLs have fresh data and can be skipped const { staleUrls, freshUrls } = await this.checkEpgFreshness( validUrls, 12 @@ -235,7 +174,7 @@ export default class EpgEvents { // a 'queued' status, then fetchEpgFromUrl silently skips the URL and no // completion update ever arrives, leaving the UI stuck at "queued". const urlsToFetch = staleUrls.filter( - (url) => !this.fetchedUrls.has(url) + (url) => !epgWorkerService.hasFetchedUrl(url) ); if (urlsToFetch.length === 0) { @@ -246,9 +185,8 @@ export default class EpgEvents { return { success: true, skipped: freshUrls }; } - // Send queued status for URLs that will actually be fetched urlsToFetch.forEach((url, index) => { - this.sendProgressToRenderer( + epgWorkerService.sendProgressToRenderer( url, 'queued', undefined, @@ -257,10 +195,8 @@ export default class EpgEvents { ); }); - // Process URLs sequentially to avoid database locking const errors: string[] = []; - for (let i = 0; i < urlsToFetch.length; i++) { - const url = urlsToFetch[i]; + for (const url of urlsToFetch) { try { await this.fetchEpgFromUrl(url); } catch (error) { @@ -277,7 +213,7 @@ export default class EpgEvents { if (errors.length > 0) { return { - success: errors.length < urlsToFetch.length, // Partial success if some worked + success: errors.length < urlsToFetch.length, message: errors.join('; '), skipped: freshUrls, }; @@ -286,561 +222,35 @@ export default class EpgEvents { return { success: true, skipped: freshUrls }; } - /** - * Fetch EPG from a single URL using worker thread - * The worker handles both parsing AND database operations - */ private static async fetchEpgFromUrl(url: string): Promise { - // Skip if already fetched this session - if (this.fetchedUrls.has(url)) { - console.log( - this.loggerLabel, - `Skipping already fetched URL: ${url}` - ); - return; - } - - return new Promise((resolve, reject) => { - let worker: Worker; - try { - worker = this.createEpgWorker(); - } catch (error) { - console.error( - this.loggerLabel, - 'Failed to create worker:', - error - ); - reject(error); - return; - } - - this.workers.set(url, worker); - - // Guards against double-settling and keeps the outer loop moving - // when the worker dies or hangs without sending EPG_COMPLETE/EPG_ERROR. - let settled = false; - const settle = (fn: () => void) => { - if (settled) return; - settled = true; - clearTimeout(timeoutId); - fn(); - }; - - const timeoutId = setTimeout(() => { - const errorMessage = `EPG fetch timed out after ${ - this.FETCH_TIMEOUT_MS / 1000 - }s`; - console.error(this.loggerLabel, `${errorMessage}: ${url}`); - this.sendProgressToRenderer( - url, - 'error', - undefined, - errorMessage - ); - worker.terminate(); - this.workers.delete(url); - settle(() => reject(new Error(errorMessage))); - }, this.FETCH_TIMEOUT_MS); - - worker.on( - 'message', - async (message: { - type: string; - error?: string; - url?: string; - stats?: { totalChannels: number; totalPrograms: number }; - }) => { - try { - switch (message.type) { - case 'READY': - // Notify renderer that loading started - this.sendProgressToRenderer(url, 'loading', { - totalChannels: 0, - totalPrograms: 0, - }); - worker.postMessage({ type: 'FETCH_EPG', url }); - break; - - case 'EPG_PROGRESS': - if (message.stats) { - // Forward progress to renderer - this.sendProgressToRenderer( - url, - 'loading', - message.stats - ); - } - break; - - case 'EPG_COMPLETE': - console.log( - this.loggerLabel, - `EPG parsing complete for ${url}:`, - message.stats - ); - // Notify renderer of completion - this.sendProgressToRenderer( - url, - 'complete', - message.stats - ); - this.fetchedUrls.add(url); - worker.terminate(); - this.workers.delete(url); - settle(() => resolve()); - break; - - case 'EPG_ERROR': - console.error( - this.loggerLabel, - 'Worker error:', - message.error - ); - // Notify renderer of error - this.sendProgressToRenderer( - url, - 'error', - undefined, - message.error - ); - worker.terminate(); - this.workers.delete(url); - settle(() => - reject( - new Error( - message.error || 'Unknown error' - ) - ) - ); - break; - } - } catch (err) { - console.error( - this.loggerLabel, - 'Error handling message:', - err - ); - this.sendProgressToRenderer( - url, - 'error', - undefined, - err instanceof Error ? err.message : String(err) - ); - worker.terminate(); - this.workers.delete(url); - settle(() => reject(err)); - } - } - ); - - worker.on('error', (error) => { - console.error(this.loggerLabel, 'Worker error event:', error); - this.sendProgressToRenderer( - url, - 'error', - undefined, - error.message - ); - worker.terminate(); - this.workers.delete(url); - settle(() => reject(error)); - }); - - worker.on('exit', (code) => { - // If the worker exits without emitting EPG_COMPLETE/EPG_ERROR, - // settle the promise so the outer sequential loop can advance - // instead of hanging forever. - if (settled) return; - const errorMessage = `Worker exited unexpectedly (code ${code})`; - console.error(this.loggerLabel, `${errorMessage}: ${url}`); - this.sendProgressToRenderer( - url, - 'error', - undefined, - errorMessage - ); - this.workers.delete(url); - settle(() => reject(new Error(errorMessage))); - }); - }); + return epgWorkerService.fetchEpgFromUrl(url); } - private static normalizeChannelLookupKeys(channelIds: string[]): string[] { - return Array.from( - new Set( - channelIds - .map((channelId) => channelId.trim()) - .filter((channelId) => channelId.length > 0) - ) - ); - } - - private static resolveChannelMetadataCandidate( - channelId: string, - candidates: EpgChannelMetadata[] - ): EpgChannelMetadata | null { - const lowerChannelId = channelId.toLowerCase(); - - const exactIdMatch = - candidates.find((candidate) => candidate.id === channelId) ?? null; - if (exactIdMatch) { - return exactIdMatch; - } - - const caseInsensitiveIdMatch = - candidates.find( - (candidate) => candidate.id.toLowerCase() === lowerChannelId - ) ?? null; - if (caseInsensitiveIdMatch) { - return caseInsensitiveIdMatch; - } - - const exactDisplayNameMatch = - candidates.find( - (candidate) => candidate.displayName === channelId - ) ?? null; - if (exactDisplayNameMatch) { - return exactDisplayNameMatch; - } - - return ( - candidates.find( - (candidate) => - candidate.displayName.toLowerCase() === lowerChannelId - ) ?? null - ); - } - - /** - * Transform database row to flat EpgProgram interface - */ - private static transformDbRowToEpgProgram(row: { - id: number; - channelId: string; - start: string; - stop: string; - title: string; - description: string | null; - category: string | null; - iconUrl: string | null; - rating: string | null; - episodeNum: string | null; - }) { - return { - start: row.start, - stop: row.stop, - channel: row.channelId, - title: row.title, - desc: row.description, - category: row.category, - iconUrl: row.iconUrl, - rating: row.rating, - episodeNum: row.episodeNum, - }; - } - - private static isValidEpgProgram(program: EpgProgram): boolean { - return Boolean( - program.start && - program.stop && - !Number.isNaN(new Date(program.start).getTime()) && - !Number.isNaN(new Date(program.stop).getTime()) - ); - } - - /** - * Get programs for a specific channel from database - */ private static async handleGetChannelPrograms( channelId: string ): Promise { - try { - const db = await getDatabase(); - const trimmedChannelId = channelId.trim(); - - if (!trimmedChannelId) { - return []; - } - - // Try exact channel ID match first - let results = await db - .select() - .from(schema.epgPrograms) - .where(eq(schema.epgPrograms.channelId, trimmedChannelId)) - .orderBy(schema.epgPrograms.start) - .limit(500); - - if (results.length > 0) { - return results - .map(this.transformDbRowToEpgProgram) - .filter(this.isValidEpgProgram); - } - - // Some playlists provide the right tvg-id with different casing than - // the XMLTV feed. Resolve the canonical channel row before giving up. - let channel = await db - .select() - .from(schema.epgChannels) - .where( - sql`${schema.epgChannels.id} = ${trimmedChannelId} COLLATE NOCASE` - ) - .limit(1); - - if (channel.length > 0) { - results = await db - .select() - .from(schema.epgPrograms) - .where(eq(schema.epgPrograms.channelId, channel[0].id)) - .orderBy(schema.epgPrograms.start) - .limit(500); - - if (results.length > 0) { - return results - .map(this.transformDbRowToEpgProgram) - .filter(this.isValidEpgProgram); - } - } - - // Try exact display name match before giving up. Using wildcard LIKE - // here can scan the whole table on the Electron main process. - channel = await db - .select() - .from(schema.epgChannels) - .where(eq(schema.epgChannels.displayName, trimmedChannelId)) - .limit(1); - - if (channel.length === 0) { - channel = await db - .select() - .from(schema.epgChannels) - .where( - sql`${schema.epgChannels.displayName} = ${trimmedChannelId} COLLATE NOCASE` - ) - .limit(1); - } - - if (channel.length > 0) { - results = await db - .select() - .from(schema.epgPrograms) - .where(eq(schema.epgPrograms.channelId, channel[0].id)) - .orderBy(schema.epgPrograms.start) - .limit(500); - - return results - .map(this.transformDbRowToEpgProgram) - .filter(this.isValidEpgProgram); - } - - return []; - } catch (error) { - console.error( - this.loggerLabel, - 'Error getting channel programs:', - error - ); - return []; - } + return epgQueryService.getChannelPrograms(channelId); } - /** - * Batch lookup of "currently playing" programs for many channels in a - * single SQL query. Replaces the N+1 pattern where the channel list - * fired one IPC + query per visible channel. - */ private static async handleGetCurrentProgramsBatch( channelIds: string[] ): Promise> { - const result: Record = {}; - if (!Array.isArray(channelIds) || channelIds.length === 0) { - return result; - } - - const validIds = Array.from( - new Set( - channelIds - .map((id) => id?.trim()) - .filter((id): id is string => Boolean(id)) - ) - ); - if (validIds.length === 0) { - return result; - } - - try { - const db = await getDatabase(); - const now = new Date().toISOString(); - - const rows = await db - .select() - .from(schema.epgPrograms) - .where( - and( - inArray(schema.epgPrograms.channelId, validIds), - lte(schema.epgPrograms.start, now), - gte(schema.epgPrograms.stop, now) - ) - ); - - for (const row of rows) { - if (!result[row.channelId]) { - const program = this.transformDbRowToEpgProgram(row); - if (this.isValidEpgProgram(program)) { - result[row.channelId] = program; - } - } - } - - // Per-channel fallback for IDs that didn't match by exact channel_id. - // Mirrors handleGetChannelPrograms's NOCASE-id and display-name resolution. - const unmatchedIds = validIds.filter((id) => !(id in result)); - for (const channelId of unmatchedIds) { - result[channelId] = null; - - let channel = await db - .select() - .from(schema.epgChannels) - .where( - sql`${schema.epgChannels.id} = ${channelId} COLLATE NOCASE` - ) - .limit(1); - - if (channel.length === 0) { - channel = await db - .select() - .from(schema.epgChannels) - .where( - sql`${schema.epgChannels.displayName} = ${channelId} COLLATE NOCASE` - ) - .limit(1); - } - - if (channel.length === 0) { - continue; - } - - const programRows = await db - .select() - .from(schema.epgPrograms) - .where( - and( - eq(schema.epgPrograms.channelId, channel[0].id), - lte(schema.epgPrograms.start, now), - gte(schema.epgPrograms.stop, now) - ) - ) - .limit(1); - - if (programRows.length > 0) { - const program = this.transformDbRowToEpgProgram( - programRows[0] - ); - if (this.isValidEpgProgram(program)) { - result[channelId] = program; - } - } - } - - return result; - } catch (error) { - console.error( - this.loggerLabel, - 'Error getting batch current programs:', - error - ); - return result; - } + return epgQueryService.getCurrentProgramsBatch(channelIds); } - /** - * Get all channels from database - */ private static async handleGetAllChannels(): Promise<{ channels: Array<{ id: string; displayName: string }>; programs: never[]; }> { - try { - const db = await getDatabase(); - const channels = await db - .select({ - id: schema.epgChannels.id, - displayName: schema.epgChannels.displayName, - }) - .from(schema.epgChannels) - .orderBy(schema.epgChannels.displayName); - - return { channels, programs: [] }; - } catch (error) { - console.error( - this.loggerLabel, - 'Error getting all channels:', - error - ); - return { channels: [], programs: [] }; - } + return epgQueryService.getAllChannels(); } - /** - * Resolve EPG channel metadata for a batch of lookup keys. - * - * Lookup precedence per key: - * 1. exact channel id - * 2. case-insensitive channel id - * 3. exact display name - * 4. case-insensitive display name - */ private static async handleGetChannelMetadata( channelIds: string[] ): Promise> { - try { - const normalizedChannelIds = - this.normalizeChannelLookupKeys(channelIds); - - if (normalizedChannelIds.length === 0) { - return {}; - } - - const db = await getDatabase(); - const lowerKeys = Array.from( - new Set( - normalizedChannelIds.map((channelId) => - channelId.toLowerCase() - ) - ) - ); - const lowerKeyValues = lowerKeys.map((key) => sql`${key}`); - - const candidates = await db - .select({ - id: schema.epgChannels.id, - displayName: schema.epgChannels.displayName, - iconUrl: schema.epgChannels.iconUrl, - }) - .from(schema.epgChannels) - .where(sql` - LOWER(${schema.epgChannels.id}) IN (${sql.join(lowerKeyValues, sql`, `)}) - OR LOWER(${schema.epgChannels.displayName}) IN (${sql.join(lowerKeyValues, sql`, `)}) - `); - - return Object.fromEntries( - normalizedChannelIds.map((channelId) => [ - channelId, - this.resolveChannelMetadataCandidate(channelId, candidates), - ]) - ); - } catch (error) { - console.error( - this.loggerLabel, - 'Error getting channel metadata:', - error - ); - return {}; - } + return epgQueryService.getChannelMetadata(channelIds); } - /** - * Get channels by range (for pagination) with their programs - */ private static async handleGetChannelsByRange( skip: number, limit: number @@ -852,101 +262,10 @@ export default class EpgEvents { programs: EpgProgram[]; }> > { - try { - const db = await getDatabase(); - const channels = await db - .select({ - id: schema.epgChannels.id, - displayName: schema.epgChannels.displayName, - iconUrl: schema.epgChannels.iconUrl, - }) - .from(schema.epgChannels) - .orderBy(schema.epgChannels.displayName) - .offset(skip) - .limit(limit); - - // Fetch programs for each channel - const channelsWithPrograms = await Promise.all( - channels.map(async (channel) => { - const programs = await db - .select() - .from(schema.epgPrograms) - .where(eq(schema.epgPrograms.channelId, channel.id)) - .orderBy(schema.epgPrograms.start); - - return { - ...channel, - programs: programs.map(this.transformDbRowToEpgProgram), - }; - }) - ); - - return channelsWithPrograms; - } catch (error) { - console.error( - this.loggerLabel, - 'Error getting channels by range:', - error - ); - return []; - } + return epgQueryService.getChannelsByRange(skip, limit); } - /** - * Clear all EPG data using worker thread to avoid blocking main thread - */ static async clearEpgData(): Promise { - return new Promise((resolve, reject) => { - let worker: Worker; - try { - worker = this.createEpgWorker(); - } catch (error) { - console.error( - this.loggerLabel, - 'Failed to create worker for clear:', - error - ); - reject(error); - return; - } - - worker.on( - 'message', - (message: { type: string; error?: string }) => { - if (message.type === 'READY') { - worker.postMessage({ type: 'CLEAR_EPG' }); - } else if (message.type === 'CLEAR_COMPLETE') { - console.log( - this.loggerLabel, - 'EPG data cleared via worker' - ); - this.fetchedUrls.clear(); - // Terminate any running fetch workers - this.workers.forEach((w) => w.terminate()); - this.workers.clear(); - worker.terminate(); - resolve(); - } else if (message.type === 'EPG_ERROR') { - console.error( - this.loggerLabel, - 'Worker clear error:', - message.error - ); - worker.terminate(); - reject(new Error(message.error || 'Clear failed')); - } - } - ); - - worker.on('error', (error) => { - console.error( - this.loggerLabel, - 'Worker error during clear:', - error - ); - worker.terminate(); - reject(error); - }); - }); + return epgWorkerService.clearEpgData(); } } diff --git a/docs/architecture/sqlite-db-worker.md b/docs/architecture/sqlite-db-worker.md index b9bb7f85c..66cdaf999 100644 --- a/docs/architecture/sqlite-db-worker.md +++ b/docs/architecture/sqlite-db-worker.md @@ -267,6 +267,20 @@ Current sources: 2. `apps/electron-backend/src/app/workers/database.worker-connection.ts` 3. `apps/electron-backend/src/app/workers/epg-parser.worker.ts` +Main-process EPG ownership is split across focused event modules: + +1. `apps/electron-backend/src/app/events/epg.events.ts` registers EPG IPC + handlers and owns freshness/fetch orchestration. +2. `apps/electron-backend/src/app/events/epg-worker.service.ts` owns EPG + worker creation, renderer progress updates, fetch worker lifecycle, and + clear-worker lifecycle. +3. `apps/electron-backend/src/app/events/epg-query.service.ts` owns EPG + channel/program database lookups, metadata resolution, and DB row mapping. + +Keep worker lifecycle state out of the IPC registration layer. Add new EPG DB +lookup behavior to `epg-query.service.ts`; add new EPG worker/progress behavior +to `epg-worker.service.ts`. + ## UI Behavior Changes ### Search @@ -291,9 +305,9 @@ Current contract: 2. Any DB-backed lookup that starts from an Xtream result card, favorite button, recent-item update, continue-watching flow, or detail route must resolve content by: - - `playlist_id` - - `xtream_id` - - `content.type` + - `playlist_id` + - `xtream_id` + - `content.type` 3. Mixed Xtream collection identity must key entries by `type + xtream_id`, not `xtream_id` alone. This includes favorites maps, recent-item lists, dashboard collection payloads, and other UI state keyed off persisted @@ -427,6 +441,10 @@ covers: 1. shared worker bootstrap usage for the EPG worker 2. `nativeModuleSearchPaths` forwarding into workerData 3. actionable worker-path resolution failures +4. EPG clear worker rejection on unexpected exit or timeout +5. case-insensitive EPG program lookup fallbacks +6. metadata lookup precedence for exact/case-insensitive id and display name +7. malformed EPG row filtering `apps/electron-backend/src/app/workers/worker-runtime-paths.spec.ts` covers: