diff --git a/apps/electron-backend/build-worker.js b/apps/electron-backend/build-worker.js index 5fd6b8413..e23285b81 100644 --- a/apps/electron-backend/build-worker.js +++ b/apps/electron-backend/build-worker.js @@ -12,6 +12,11 @@ const nodeBuiltins = [ 'v8', 'vm', 'worker_threads', 'zlib' ]; +// Native modules that must be externalized (loaded at runtime) +const nativeModules = [ + 'better-sqlite3', +]; + const isProduction = process.env.NODE_ENV === 'production'; async function buildWorker() { @@ -25,12 +30,17 @@ async function buildWorker() { target: 'node18', format: 'cjs', outfile: path.join(__dirname, '../../dist/apps/electron-backend/workers/epg-parser.worker.js'), - external: nodeBuiltins.map(m => `node:${m}`).concat(nodeBuiltins), + external: [ + ...nodeBuiltins.map(m => `node:${m}`), + ...nodeBuiltins, + ...nativeModules, + ], sourcemap: !isProduction, minify: isProduction, // Resolve workspace libraries from tsconfig paths alias: { 'shared-interfaces': path.join(__dirname, '../../libs/shared/interfaces/src/index.ts'), + 'database': path.join(__dirname, '../../libs/shared/database/src/index.ts'), }, }); diff --git a/apps/electron-backend/src/app/api/main.preload.ts b/apps/electron-backend/src/app/api/main.preload.ts index dc1434a28..bf51bc1ee 100644 --- a/apps/electron-backend/src/app/api/main.preload.ts +++ b/apps/electron-backend/src/app/api/main.preload.ts @@ -17,6 +17,17 @@ contextBridge.exposeInMainWorld('electron', { ) => { ipcRenderer.on('player-error', (_event, data) => callback(data)); }, + // EPG progress listener + onEpgProgress: ( + callback: (data: { + url: string; + status: 'loading' | 'complete' | 'error'; + stats?: { totalChannels: number; totalPrograms: number }; + error?: string; + }) => void + ) => { + ipcRenderer.on('EPG_PROGRESS_UPDATE', (_event, data) => callback(data)); + }, getAppVersion: () => ipcRenderer.invoke('get-app-version'), platform: process.platform, fetchPlaylistByUrl: (url: string, title?: string) => @@ -72,6 +83,11 @@ contextBridge.exposeInMainWorld('electron', { getEpgChannelsByRange: (skip: number, limit: number) => ipcRenderer.invoke('EPG_GET_CHANNELS_BY_RANGE', { skip, limit }), forceFetchEpg: (url: string) => ipcRenderer.invoke('EPG_FORCE_FETCH', url), + clearEpgData: () => ipcRenderer.invoke('EPG_CLEAR_ALL'), + checkEpgFreshness: (urls: string[], maxAgeHours?: number) => + ipcRenderer.invoke('EPG_CHECK_FRESHNESS', { urls, maxAgeHours }), + searchEpgPrograms: (searchTerm: string, limit?: number) => + ipcRenderer.invoke('EPG_DB_SEARCH_PROGRAMS', searchTerm, limit), setMpvPlayerPath: (mpvPlayerPath: string) => ipcRenderer.invoke('SET_MPV_PLAYER_PATH', mpvPlayerPath), setVlcPlayerPath: (vlcPlayerPath: string) => diff --git a/apps/electron-backend/src/app/database/schema.ts b/apps/electron-backend/src/app/database/schema.ts index 752ad0784..cd9c38421 100644 --- a/apps/electron-backend/src/app/database/schema.ts +++ b/apps/electron-backend/src/app/database/schema.ts @@ -10,6 +10,8 @@ export { content, recentlyViewed, favorites, + epgChannels, + epgPrograms, // Types type Playlist, type NewPlaylist, @@ -21,5 +23,9 @@ export { type NewRecentlyViewed, type Favorite, type NewFavorite, + type EpgChannel, + type NewEpgChannel, + type EpgProgramDb, + type NewEpgProgramDb, } from 'database'; diff --git a/apps/electron-backend/src/app/events/database.events.ts b/apps/electron-backend/src/app/events/database.events.ts index fce5dadf2..cbe25b94e 100644 --- a/apps/electron-backend/src/app/events/database.events.ts +++ b/apps/electron-backend/src/app/events/database.events.ts @@ -7,6 +7,7 @@ import { ipcMain } from 'electron'; import './database/category.events'; import './database/content.events'; +import './database/epg-db.events'; import './database/favorites.events'; import './database/playlist.events'; import './database/recently-viewed.events'; diff --git a/apps/electron-backend/src/app/events/database/epg-db.events.ts b/apps/electron-backend/src/app/events/database/epg-db.events.ts new file mode 100644 index 000000000..2dfedb00e --- /dev/null +++ b/apps/electron-backend/src/app/events/database/epg-db.events.ts @@ -0,0 +1,391 @@ +/** + * EPG Database IPC event handlers + * Operations for storing and querying EPG data in SQLite with FTS5 support + */ + +import { and, eq, gte, lte, sql } from 'drizzle-orm'; +import { ipcMain } from 'electron'; +import { getDatabase } from '../../database/connection'; +import * as schema from '../../database/schema'; + +const loggerLabel = '[EPG DB]'; + +/** + * Save EPG channels in bulk (upsert) + */ +ipcMain.handle( + 'EPG_DB_SAVE_CHANNELS', + async ( + _event, + sourceUrl: string, + channels: Array<{ + id: string; + displayName: string; + iconUrl?: string; + url?: string; + }> + ) => { + try { + const db = await getDatabase(); + + // Delete existing channels from this source first + await db + .delete(schema.epgChannels) + .where(eq(schema.epgChannels.sourceUrl, sourceUrl)); + + if (channels.length === 0) return { success: true, count: 0 }; + + // Insert in chunks + const CHUNK_SIZE = 500; + let totalInserted = 0; + + for (let i = 0; i < channels.length; i += CHUNK_SIZE) { + const chunk = channels.slice(i, i + CHUNK_SIZE); + const values = chunk.map((ch) => ({ + id: ch.id, + displayName: ch.displayName, + iconUrl: ch.iconUrl || null, + url: ch.url || null, + sourceUrl, + })); + + await db.insert(schema.epgChannels).values(values); + totalInserted += chunk.length; + } + + console.log( + loggerLabel, + `Saved ${totalInserted} channels from ${sourceUrl}` + ); + return { success: true, count: totalInserted }; + } catch (error) { + console.error(loggerLabel, 'Error saving EPG channels:', error); + throw error; + } + } +); + +/** + * Save EPG programs in bulk + */ +ipcMain.handle( + 'EPG_DB_SAVE_PROGRAMS', + async ( + _event, + programs: Array<{ + channelId: string; + start: string; + stop: string; + title: string; + description?: string; + category?: string; + iconUrl?: string; + rating?: string; + episodeNum?: string; + }> + ) => { + try { + const db = await getDatabase(); + + if (programs.length === 0) return { success: true, count: 0 }; + + // Insert in chunks for better performance + const CHUNK_SIZE = 500; + let totalInserted = 0; + + for (let i = 0; i < programs.length; i += CHUNK_SIZE) { + const chunk = programs.slice(i, i + CHUNK_SIZE); + const values = chunk.map((prog) => ({ + channelId: prog.channelId, + start: prog.start, + stop: prog.stop, + title: prog.title, + description: prog.description || null, + category: prog.category || null, + iconUrl: prog.iconUrl || null, + rating: prog.rating || null, + episodeNum: prog.episodeNum || null, + })); + + await db.insert(schema.epgPrograms).values(values); + totalInserted += chunk.length; + } + + console.log(loggerLabel, `Saved ${totalInserted} programs`); + return { success: true, count: totalInserted }; + } catch (error) { + console.error(loggerLabel, 'Error saving EPG programs:', error); + throw error; + } + } +); + +/** + * Get programs for a specific channel + */ +ipcMain.handle( + 'EPG_DB_GET_CHANNEL_PROGRAMS', + async (_event, channelId: string, fromTime?: string, toTime?: string) => { + try { + const db = await getDatabase(); + const now = new Date().toISOString(); + + let query = db + .select() + .from(schema.epgPrograms) + .where(eq(schema.epgPrograms.channelId, channelId)); + + // Apply time filters if provided + if (fromTime && toTime) { + query = db + .select() + .from(schema.epgPrograms) + .where( + and( + eq(schema.epgPrograms.channelId, channelId), + gte(schema.epgPrograms.stop, fromTime), + lte(schema.epgPrograms.start, toTime) + ) + ); + } else { + // Default: from now onwards + query = db + .select() + .from(schema.epgPrograms) + .where( + and( + eq(schema.epgPrograms.channelId, channelId), + gte(schema.epgPrograms.stop, now) + ) + ); + } + + const results = await query.orderBy(schema.epgPrograms.start); + return results; + } catch (error) { + console.error( + loggerLabel, + 'Error getting channel programs:', + error + ); + throw error; + } + } +); + +/** + * Get current program for a channel (what's on now) + */ +ipcMain.handle('EPG_DB_GET_CURRENT_PROGRAM', async (_event, channelId: string) => { + try { + const db = await getDatabase(); + const now = new Date().toISOString(); + + const result = await db + .select() + .from(schema.epgPrograms) + .where( + and( + eq(schema.epgPrograms.channelId, channelId), + lte(schema.epgPrograms.start, now), + gte(schema.epgPrograms.stop, now) + ) + ) + .limit(1); + + return result[0] || null; + } catch (error) { + console.error(loggerLabel, 'Error getting current program:', error); + throw error; + } +}); + +/** + * Full-text search EPG programs using FTS5 with LIKE fallback + * Handles Cyrillic and other Unicode text properly + * Includes channel display name via JOIN + */ +ipcMain.handle( + 'EPG_DB_SEARCH_PROGRAMS', + async (_event, searchTerm: string, limit = 50) => { + try { + const db = await getDatabase(); + const now = new Date().toISOString(); + const trimmedTerm = searchTerm.trim(); + + if (!trimmedTerm) { + return []; + } + + // Use LIKE for substring matching (works better with Cyrillic) + // This is more intuitive for users expecting exact substring matches + const likePattern = `%${trimmedTerm}%`; + + // JOIN with epg_channels to get channel display name + const results = await db.all(sql` + SELECT + p.*, + c.display_name as channel_name + FROM epg_programs p + LEFT JOIN epg_channels c ON p.channel_id = c.id + WHERE ( + p.title LIKE ${likePattern} + OR p.description LIKE ${likePattern} + OR p.category LIKE ${likePattern} + ) + AND p.stop >= ${now} + ORDER BY p.start + LIMIT ${limit} + `); + + return results; + } catch (error) { + console.error(loggerLabel, 'Error searching EPG programs:', error); + throw error; + } + } +); + +/** + * Get all EPG channels + */ +ipcMain.handle('EPG_DB_GET_CHANNELS', async () => { + try { + const db = await getDatabase(); + const results = await db + .select() + .from(schema.epgChannels) + .orderBy(schema.epgChannels.displayName); + return results; + } catch (error) { + console.error(loggerLabel, 'Error getting EPG channels:', error); + throw error; + } +}); + +/** + * Get channel by ID or display name + */ +ipcMain.handle('EPG_DB_GET_CHANNEL', async (_event, channelIdOrName: string) => { + try { + const db = await getDatabase(); + + // Try exact ID match first + let result = await db + .select() + .from(schema.epgChannels) + .where(eq(schema.epgChannels.id, channelIdOrName)) + .limit(1); + + if (result.length > 0) return result[0]; + + // Try display name match (case-insensitive) + result = await db + .select() + .from(schema.epgChannels) + .where( + sql`LOWER(${schema.epgChannels.displayName}) = LOWER(${channelIdOrName})` + ) + .limit(1); + + return result[0] || null; + } catch (error) { + console.error(loggerLabel, 'Error getting EPG channel:', error); + throw error; + } +}); + +/** + * Cleanup expired programs (older than specified hours) + */ +ipcMain.handle('EPG_DB_CLEANUP_EXPIRED', async (_event, hoursToKeep = 24) => { + try { + const db = await getDatabase(); + const cutoff = new Date( + Date.now() - hoursToKeep * 60 * 60 * 1000 + ).toISOString(); + + const result = await db + .delete(schema.epgPrograms) + .where(lte(schema.epgPrograms.stop, cutoff)); + + console.log( + loggerLabel, + `Cleaned up expired programs (older than ${hoursToKeep}h)` + ); + return { success: true }; + } catch (error) { + console.error(loggerLabel, 'Error cleaning up EPG programs:', error); + throw error; + } +}); + +/** + * Clear all EPG data for a specific source URL + */ +ipcMain.handle('EPG_DB_CLEAR_SOURCE', async (_event, sourceUrl: string) => { + try { + const db = await getDatabase(); + + // Deleting channels will cascade delete programs due to foreign key + await db + .delete(schema.epgChannels) + .where(eq(schema.epgChannels.sourceUrl, sourceUrl)); + + console.log(loggerLabel, `Cleared EPG data for source: ${sourceUrl}`); + return { success: true }; + } catch (error) { + console.error(loggerLabel, 'Error clearing EPG source:', error); + throw error; + } +}); + +/** + * Clear all EPG data + */ +ipcMain.handle('EPG_DB_CLEAR_ALL', async () => { + try { + const db = await getDatabase(); + + await db.delete(schema.epgPrograms); + await db.delete(schema.epgChannels); + + console.log(loggerLabel, 'Cleared all EPG data'); + return { success: true }; + } catch (error) { + console.error(loggerLabel, 'Error clearing all EPG data:', error); + throw error; + } +}); + +/** + * Get EPG statistics + */ +ipcMain.handle('EPG_DB_GET_STATS', async () => { + try { + const db = await getDatabase(); + + const channelCount = await db + .select({ count: sql`count(*)` }) + .from(schema.epgChannels); + + const programCount = await db + .select({ count: sql`count(*)` }) + .from(schema.epgPrograms); + + const now = new Date().toISOString(); + const futureProgramCount = await db + .select({ count: sql`count(*)` }) + .from(schema.epgPrograms) + .where(gte(schema.epgPrograms.stop, now)); + + return { + channels: channelCount[0].count, + programs: programCount[0].count, + futurePrograms: futureProgramCount[0].count, + }; + } catch (error) { + console.error(loggerLabel, 'Error getting EPG stats:', error); + throw error; + } +}); diff --git a/apps/electron-backend/src/app/events/epg.events.ts b/apps/electron-backend/src/app/events/epg.events.ts index 9d6d33824..01523102d 100644 --- a/apps/electron-backend/src/app/events/epg.events.ts +++ b/apps/electron-backend/src/app/events/epg.events.ts @@ -1,112 +1,220 @@ -import { app, ipcMain } from 'electron'; +import { and, eq, gte, sql } from 'drizzle-orm'; +import { app, BrowserWindow, ipcMain } from 'electron'; import * as path from 'path'; -import { - EpgChannel, - EpgChannelWithPrograms, - EpgData, - EpgProgram, -} from 'shared-interfaces'; import { Worker } from 'worker_threads'; import { pathToFileURL } from 'url'; +import { getDatabase } from '../database/connection'; +import * as schema from '../database/schema'; +import { EpgProgram } from 'shared-interfaces'; /** * EPG Events Handler - * Manages EPG data fetching and querying using worker threads + * Manages EPG data fetching and querying using worker threads. + * Database operations are performed in the worker thread to avoid blocking the main thread. */ export default class EpgEvents { - private static epgData: EpgData = { - channels: [], - programs: [], - }; - - private static epgDataMerged: Map = - new Map(); private static fetchedUrls: Set = new Set(); private static workers: Map = new Map(); private static readonly loggerLabel = '[EPG Events]'; - // Display name index for O(1) lookups (lowercase name -> channel id) - private static displayNameIndex: Map = new Map(); + /** + * 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[] }) => { + 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 }) => { + async (_event, args: { channelId: string }) => { return this.handleGetChannelPrograms(args.channelId); } ); - // Get all channels + // Get all channels from database ipcMain.handle('EPG_GET_CHANNELS', async () => { - return this.epgData; + return this.handleGetAllChannels(); }); // Get channels by range (pagination) ipcMain.handle( 'EPG_GET_CHANNELS_BY_RANGE', - async (event, args: { skip: number; limit: number }) => { + async (_event, args: { skip: number; limit: number }) => { return this.handleGetChannelsByRange(args.skip, args.limit); } ); // Force fetch (ignore cache) - ipcMain.handle('EPG_FORCE_FETCH', async (event, url: string) => { + ipcMain.handle('EPG_FORCE_FETCH', async (_event, url: string) => { this.fetchedUrls.delete(url); return await this.handleFetchEpg([url]); }); - // Cleanup expired programs + // Cleanup expired programs (uses worker thread) ipcMain.handle( 'EPG_CLEANUP_EXPIRED', - async (event, hoursToKeep?: number) => { - return this.cleanupExpiredPrograms(hoursToKeep); + async (_event, hoursToKeep?: number) => { + return this.runCleanupInWorker(hoursToKeep ?? 24); + } + ); + + // 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 ( + _event, + args: { urls: string[]; maxAgeHours?: number } + ): Promise<{ staleUrls: string[]; freshUrls: string[] }> => { + return this.checkEpgFreshness(args.urls, args.maxAgeHours ?? 12); } ); return ipcMain; } + /** + * Check which EPG URLs have fresh data vs stale/missing data + * @param urls - EPG source URLs to check + * @param maxAgeHours - Maximum age in hours before data is considered stale + */ + private static async checkEpgFreshness( + urls: string[], + maxAgeHours: number + ): Promise<{ staleUrls: string[]; freshUrls: string[] }> { + const staleUrls: string[] = []; + const freshUrls: string[] = []; + const cutoffTime = new Date( + Date.now() - maxAgeHours * 60 * 60 * 1000 + ).toISOString(); + + try { + const db = await getDatabase(); + + for (const url of urls) { + if (!url?.trim()) continue; + + const result = await db + .select({ updatedAt: schema.epgChannels.updatedAt }) + .from(schema.epgChannels) + .where(eq(schema.epgChannels.sourceUrl, url)) + .limit(1); + + const isFresh = + result.length > 0 && + result[0].updatedAt && + result[0].updatedAt >= cutoffTime; + + if (isFresh) { + freshUrls.push(url); + this.fetchedUrls.add(url); + } else { + staleUrls.push(url); + } + } + } catch (error) { + console.error(this.loggerLabel, 'Error checking EPG freshness:', error); + return { staleUrls: urls, freshUrls: [] }; + } + + // Log summary + if (freshUrls.length > 0) { + console.log(this.loggerLabel, `EPG fresh (skipping): ${freshUrls.length} source(s)`); + } + if (staleUrls.length > 0) { + console.log(this.loggerLabel, `EPG stale (will fetch): ${staleUrls.length} source(s)`); + } + + return { staleUrls, freshUrls }; + } + /** * Handle EPG fetch from URLs + * Automatically skips URLs with fresh data (less than 12 hours old) + * Processes URLs sequentially to avoid SQLite database locking issues */ private static async handleFetchEpg( urls: string[] - ): Promise<{ success: boolean; message?: string }> { + ): Promise<{ success: boolean; message?: string; skipped?: string[] }> { const validUrls = urls.filter((url) => url?.trim()); if (validUrls.length === 0) { return { success: false, message: 'No valid URLs provided' }; } - const promises = validUrls.map((url) => this.fetchEpgFromUrl(url)); + // Check which URLs have fresh data and can be skipped + const { staleUrls, freshUrls } = await this.checkEpgFreshness(validUrls, 12); - try { - await Promise.all(promises); - return { success: true }; - } catch (error) { - console.error(this.loggerLabel, 'Error fetching EPG:', error); + if (staleUrls.length === 0) { + return { success: true, message: 'All EPG data is fresh', skipped: freshUrls }; + } + + // Send queued status for all stale URLs first + staleUrls.forEach((url, index) => { + this.sendProgressToRenderer(url, 'queued', undefined, undefined, index + 1); + }); + + // Process only stale URLs sequentially to avoid database locking + const errors: string[] = []; + for (let i = 0; i < staleUrls.length; i++) { + const url = staleUrls[i]; + try { + await this.fetchEpgFromUrl(url); + } catch (error) { + console.error(this.loggerLabel, `Error fetching EPG from ${url}:`, error); + errors.push(error instanceof Error ? error.message : String(error)); + } + } + + if (errors.length > 0) { return { - success: false, - message: error instanceof Error ? error.message : String(error), + success: errors.length < staleUrls.length, // Partial success if some worked + message: errors.join('; '), + skipped: freshUrls, }; } + + 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 + // Skip if already fetched this session if (this.fetchedUrls.has(url)) { + console.log(this.loggerLabel, `Skipping already fetched URL: ${url}`); return; } @@ -114,7 +222,6 @@ export default class EpgEvents { let workerPath: string; if (app.isPackaged) { - // In packaged app: Resources/dist/apps/electron-backend/workers/epg-parser.worker.js const resourcesPath = path.dirname(app.getAppPath()); workerPath = path.join( resourcesPath, @@ -125,27 +232,20 @@ export default class EpgEvents { 'epg-parser.worker.js' ); } else { - // In development: dist/apps/electron-backend/workers/epg-parser.worker.js workerPath = path.join(__dirname, 'workers', 'epg-parser.worker.js'); } let worker: Worker; try { - // Worker threads require file:// URLs wrapped in URL object for packaged apps const workerURL = pathToFileURL(workerPath); worker = new Worker(workerURL, { - // Increase memory limits for large EPG files resourceLimits: { - maxOldGenerationSizeMb: 4096, // 4GB for parsed data - maxYoungGenerationSizeMb: 512, // 512MB for temporary allocations + maxOldGenerationSizeMb: 4096, + maxYoungGenerationSizeMb: 512, }, }); } catch (error) { - console.error( - this.loggerLabel, - 'Failed to create worker:', - error - ); + console.error(this.loggerLabel, 'Failed to create worker:', error); reject(error); return; } @@ -154,68 +254,70 @@ export default class EpgEvents { worker.on( 'message', - (message: { + async (message: { type: string; - channels?: EpgChannel[]; - programs?: EpgProgram[]; error?: string; url?: string; stats?: { totalChannels: number; totalPrograms: number }; }) => { - switch (message.type) { - case 'READY': - worker.postMessage({ type: 'FETCH_EPG', url }); - break; + 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_CHANNELS_BATCH': - // Add channels incrementally - if (message.channels) { - this.addChannelsBatch(message.channels); - } - break; + case 'EPG_PROGRESS': + if (message.stats) { + console.log( + this.loggerLabel, + `Progress: ${message.stats.totalChannels} channels, ${message.stats.totalPrograms} programs` + ); + // Forward progress to renderer + this.sendProgressToRenderer(url, 'loading', message.stats); + } + break; - case 'EPG_PROGRAMS_BATCH': - // Add programs incrementally - if (message.programs) { - this.addProgramsBatch(message.programs); - } - break; - - case 'EPG_PROGRESS': - // Log progress for large files - if (message.stats) { + case 'EPG_COMPLETE': console.log( this.loggerLabel, - `Progress: ${message.stats.totalChannels} channels, ${message.stats.totalPrograms} programs` + `EPG parsing complete for ${url}:`, + message.stats ); - } - break; + // Notify renderer of completion + this.sendProgressToRenderer(url, 'complete', message.stats); + this.fetchedUrls.add(url); + // Trigger cleanup in worker thread (non-blocking) + worker.postMessage({ type: 'CLEANUP_EXPIRED', hoursToKeep: 24 }); + break; - case 'EPG_COMPLETE': - console.log( - this.loggerLabel, - `EPG parsing complete for ${url}:`, - message.stats - ); - this.rebuildMergedData(); - // Auto-cleanup programs older than 24 hours - this.cleanupExpiredPrograms(24); - this.fetchedUrls.add(url); - worker.terminate(); - this.workers.delete(url); - resolve(); - break; + case 'CLEANUP_COMPLETE': + console.log( + this.loggerLabel, + 'Cleanup complete, terminating worker' + ); + worker.terminate(); + this.workers.delete(url); + resolve(); + break; - case 'EPG_ERROR': - console.error( - this.loggerLabel, - 'Worker error:', - message.error - ); - worker.terminate(); - this.workers.delete(url); - reject(new Error(message.error || 'Unknown error')); - 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); + reject(new Error(message.error || 'Unknown error')); + break; + } + } catch (err) { + console.error(this.loggerLabel, 'Error handling message:', err); + reject(err); } } ); @@ -239,154 +341,263 @@ export default class EpgEvents { } /** - * Add a batch of channels (streaming mode) + * Transform database row to flat EpgProgram interface */ - private static addChannelsBatch(channels: EpgChannel[]): void { - // Use Set for O(1) duplicate checking - const existingIds = new Set(this.epgData.channels.map((c) => c.id)); - const newChannels = channels.filter((c) => !existingIds.has(c.id)); - this.epgData.channels = this.epgData.channels.concat(newChannels); + 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, + }; } /** - * Add a batch of programs (streaming mode) + * Get programs for a specific channel from database */ - private static addProgramsBatch(programs: EpgProgram[]): void { - // Simply append programs - no duplicate check needed as each program is unique - this.epgData.programs = this.epgData.programs.concat(programs); + private static async handleGetChannelPrograms(channelId: string): Promise { + try { + const db = await getDatabase(); + const now = new Date().toISOString(); + + // Try exact channel ID match first + let results = await db + .select() + .from(schema.epgPrograms) + .where( + and( + eq(schema.epgPrograms.channelId, channelId), + gte(schema.epgPrograms.stop, now) + ) + ) + .orderBy(schema.epgPrograms.start) + .limit(100); + + if (results.length > 0) { + return results.map(this.transformDbRowToEpgProgram); + } + + // Try to find channel by display name + const channel = await db + .select() + .from(schema.epgChannels) + .where( + sql`LOWER(${schema.epgChannels.displayName}) LIKE LOWER(${'%' + channelId + '%'})` + ) + .limit(1); + + if (channel.length > 0) { + results = await db + .select() + .from(schema.epgPrograms) + .where( + and( + eq(schema.epgPrograms.channelId, channel[0].id), + gte(schema.epgPrograms.stop, now) + ) + ) + .orderBy(schema.epgPrograms.start) + .limit(100); + + return results.map(this.transformDbRowToEpgProgram); + } + + return []; + } catch (error) { + console.error(this.loggerLabel, 'Error getting channel programs:', error); + return []; + } } /** - * Rebuild merged data structure (channels with programs) - * Optimized for large datasets + * Get all channels from database */ - private static rebuildMergedData(): void { - this.epgDataMerged.clear(); - this.displayNameIndex.clear(); + 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); - // Create channel lookup map for O(1) access - const channelMap = new Map(this.epgData.channels.map((c) => [c.id, c])); - - // Build display name index for fast lookups - for (const channel of this.epgData.channels) { - for (const name of channel.displayName) { - if (name.value) { - this.displayNameIndex.set( - name.value.toLowerCase(), - channel.id - ); - } - } + return { channels, programs: [] }; + } catch (error) { + console.error(this.loggerLabel, 'Error getting all channels:', error); + return { channels: [], programs: [] }; } - - // Group programs by channel in a single pass - const programsByChannel = new Map(); - - for (const program of this.epgData.programs) { - const channelId = program.channel; - if (!programsByChannel.has(channelId)) { - programsByChannel.set(channelId, []); - } - programsByChannel.get(channelId)!.push(program); - } - - // Build merged structure - for (const [channelId, programs] of programsByChannel.entries()) { - const channel = channelMap.get(channelId); - if (channel) { - this.epgDataMerged.set(channelId, { - ...channel, - programs, - }); - } - } - - // Free memory - programs are now stored in epgDataMerged - this.epgData.programs = []; - - console.log( - this.loggerLabel, - `Merged data rebuilt: ${this.epgDataMerged.size} channels, ${this.displayNameIndex.size} display names indexed` - ); } /** - * Get programs for a specific channel + * Get channels by range (for pagination) with their programs */ - private static handleGetChannelPrograms(channelId: string): EpgProgram[] { - // First try exact ID match - let channelData = this.epgDataMerged.get(channelId); - - // If not found, try display name index (O(1) lookup) - if (!channelData) { - const indexedId = this.displayNameIndex.get(channelId.toLowerCase()); - if (indexedId) { - channelData = this.epgDataMerged.get(indexedId); - } - } - - // Fallback: partial match (only if index lookup failed) - if (!channelData) { - const searchTerm = channelId.toLowerCase(); - for (const [name, id] of this.displayNameIndex.entries()) { - if (name.includes(searchTerm) || searchTerm.includes(name)) { - channelData = this.epgDataMerged.get(id); - if (channelData) break; - } - } - } - - return channelData?.programs || []; - } - - /** - * Get channels by range (for pagination) - */ - private static handleGetChannelsByRange( + private static async handleGetChannelsByRange( skip: number, limit: number - ): EpgChannelWithPrograms[] { - const channels = Array.from(this.epgDataMerged.values()); - return channels.slice(skip, skip + limit); - } + ): Promise> { + 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); - /** - * Clean up expired programs (older than specified hours) - * Call this periodically to free memory from stale EPG data - */ - static cleanupExpiredPrograms(hoursToKeep = 24): number { - const cutoff = new Date( - Date.now() - hoursToKeep * 60 * 60 * 1000 - ).toISOString(); - let removedCount = 0; + // 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); - for (const channel of this.epgDataMerged.values()) { - const originalLength = channel.programs.length; - channel.programs = channel.programs.filter((p) => p.stop > cutoff); - removedCount += originalLength - channel.programs.length; - } - - if (removedCount > 0) { - console.log( - this.loggerLabel, - `Cleaned up ${removedCount} expired programs (older than ${hoursToKeep}h)` + return { + ...channel, + programs: programs.map(this.transformDbRowToEpgProgram), + }; + }) ); - } - return removedCount; + return channelsWithPrograms; + } catch (error) { + console.error(this.loggerLabel, 'Error getting channels by range:', error); + return []; + } } /** - * Clear all EPG data + * Run cleanup in worker thread to avoid blocking main thread */ - static clearEpgData(): void { - this.epgData = { channels: [], programs: [] }; - this.epgDataMerged.clear(); - this.displayNameIndex.clear(); - this.fetchedUrls.clear(); + private static async runCleanupInWorker(hoursToKeep: number): Promise<{ success: boolean }> { + return new Promise((resolve, reject) => { + let workerPath: string; - // Terminate all workers - this.workers.forEach((worker) => worker.terminate()); - this.workers.clear(); + if (app.isPackaged) { + const resourcesPath = path.dirname(app.getAppPath()); + workerPath = path.join( + resourcesPath, + 'dist', + 'apps', + 'electron-backend', + 'workers', + 'epg-parser.worker.js' + ); + } else { + workerPath = path.join(__dirname, 'workers', 'epg-parser.worker.js'); + } + + let worker: Worker; + try { + const workerURL = pathToFileURL(workerPath); + worker = new Worker(workerURL); + } catch (error) { + console.error(this.loggerLabel, 'Failed to create worker for cleanup:', error); + reject(error); + return; + } + + worker.on('message', (message: { type: string; error?: string }) => { + if (message.type === 'READY') { + worker.postMessage({ type: 'CLEANUP_EXPIRED', hoursToKeep }); + } else if (message.type === 'CLEANUP_COMPLETE') { + worker.terminate(); + resolve({ success: true }); + } else if (message.type === 'EPG_ERROR') { + console.error(this.loggerLabel, 'Worker cleanup error:', message.error); + worker.terminate(); + reject(new Error(message.error || 'Cleanup failed')); + } + }); + + worker.on('error', (error) => { + console.error(this.loggerLabel, 'Worker error during cleanup:', error); + worker.terminate(); + reject(error); + }); + }); + } + + /** + * Clear all EPG data using worker thread to avoid blocking main thread + */ + static async clearEpgData(): Promise { + return new Promise((resolve, reject) => { + let workerPath: string; + + if (app.isPackaged) { + const resourcesPath = path.dirname(app.getAppPath()); + workerPath = path.join( + resourcesPath, + 'dist', + 'apps', + 'electron-backend', + 'workers', + 'epg-parser.worker.js' + ); + } else { + workerPath = path.join(__dirname, 'workers', 'epg-parser.worker.js'); + } + + let worker: Worker; + try { + const workerURL = pathToFileURL(workerPath); + worker = new Worker(workerURL); + } 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); + }); + }); } } diff --git a/apps/electron-backend/src/app/workers/epg-parser.worker.ts b/apps/electron-backend/src/app/workers/epg-parser.worker.ts index 2943c24b8..b0af79482 100644 --- a/apps/electron-backend/src/app/workers/epg-parser.worker.ts +++ b/apps/electron-backend/src/app/workers/epg-parser.worker.ts @@ -1,30 +1,78 @@ -import { EpgChannel, EpgProgram } from 'shared-interfaces'; import { parentPort } from 'worker_threads'; import { createGunzip } from 'zlib'; import { SaxesParser, SaxesTagPlain } from 'saxes'; import { Readable } from 'stream'; +import Database from 'better-sqlite3'; +import { existsSync, mkdirSync } from 'fs'; +import { homedir } from 'os'; +import { join } from 'path'; + +/** + * Internal parsing types with arrays for XML parsing + * These are different from the flat EpgProgram interface used by the frontend + */ +interface ParsedTextValue { + lang: string; + value: string; +} + +interface ParsedIcon { + src: string; + width?: number; + height?: number; +} + +interface ParsedRating { + system: string; + value: string; +} + +interface ParsedEpisodeNum { + system: string; + value: string; +} + +interface ParsedChannel { + id: string; + displayName: ParsedTextValue[]; + icon: ParsedIcon[]; + url: string[]; +} + +interface ParsedProgram { + start: string; + stop: string; + channel: string; + title: ParsedTextValue[]; + desc: ParsedTextValue[]; + category: ParsedTextValue[]; + date: string; + episodeNum: ParsedEpisodeNum[]; + icon: ParsedIcon[]; + rating: ParsedRating[]; +} /** * Streaming EPG Parser Worker * Uses SAX parsing to process XML incrementally without loading entire file into memory. * Supports both regular and gzipped EPG files. + * Performs database operations directly to avoid blocking the main thread. */ interface WorkerMessage { - type: 'FETCH_EPG' | 'FORCE_FETCH'; - url: string; + type: 'FETCH_EPG' | 'FORCE_FETCH' | 'CLEAR_EPG' | 'CLEANUP_EXPIRED'; + url?: string; + hoursToKeep?: number; } interface WorkerResponse { type: - | 'EPG_CHANNELS_BATCH' - | 'EPG_PROGRAMS_BATCH' | 'EPG_COMPLETE' | 'EPG_ERROR' | 'EPG_PROGRESS' + | 'CLEAR_COMPLETE' + | 'CLEANUP_COMPLETE' | 'READY'; - channels?: EpgChannel[]; - programs?: EpgProgram[]; error?: string; url?: string; stats?: { @@ -35,10 +83,174 @@ interface WorkerResponse { const loggerLabel = '[EPG Worker]'; -// Batch size for sending data back to main thread +// Batch size for database inserts const CHANNEL_BATCH_SIZE = 100; const PROGRAM_BATCH_SIZE = 1000; +// Skip programs that ended more than this many hours ago +const SKIP_PROGRAMS_OLDER_THAN_HOURS = 2; + +/** + * Calculate the cutoff time for filtering old programs + * Programs that ended before this time will be skipped + */ +function getProgramCutoffTime(): Date { + return new Date(Date.now() - SKIP_PROGRAMS_OLDER_THAN_HOURS * 60 * 60 * 1000); +} + +/** + * Check if a program has already ended (is expired) + * Handles timezone-aware ISO date strings + */ +function isProgramExpired(stopTime: string, cutoff: Date): boolean { + if (!stopTime) return true; // Skip programs without stop time + + try { + const stopDate = new Date(stopTime); + // Check if the date is valid + if (isNaN(stopDate.getTime())) return false; // Don't skip if we can't parse + return stopDate < cutoff; + } catch { + return false; // Don't skip if parsing fails + } +} + +/** + * Get database file path (same as main app) + */ +function getDatabasePath(): string { + const dbDir = join(homedir(), '.iptvnator', 'databases'); + if (!existsSync(dbDir)) { + mkdirSync(dbDir, { recursive: true }); + } + return join(dbDir, 'iptvnator.db'); +} + +/** + * Database helper class for EPG operations + * Creates its own connection to avoid blocking main thread + */ +class EpgDatabase { + private db: Database.Database; + private knownChannelIds: Set = new Set(); + + // Prepared statements for better performance + private insertChannelStmt: Database.Statement; + private insertProgramStmt: Database.Statement; + private deleteChannelsStmt: Database.Statement; + + constructor() { + const dbPath = getDatabasePath(); + this.db = new Database(dbPath); + this.db.pragma('foreign_keys = ON'); + this.db.pragma('journal_mode = WAL'); // Better concurrent write performance + + // Prepare statements + // Use INSERT OR REPLACE to update existing channels and refresh updated_at + // Use strftime with ISO format for consistent date comparison + this.insertChannelStmt = this.db.prepare(` + INSERT INTO epg_channels (id, display_name, icon_url, url, source_url, updated_at) + VALUES (?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) + ON CONFLICT(id) DO UPDATE SET + display_name = excluded.display_name, + icon_url = excluded.icon_url, + url = excluded.url, + source_url = excluded.source_url, + updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') + `); + + this.insertProgramStmt = this.db.prepare(` + INSERT INTO epg_programs (channel_id, start, stop, title, description, category, icon_url, rating, episode_num) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + `); + + this.deleteChannelsStmt = this.db.prepare(` + DELETE FROM epg_channels WHERE source_url = ? + `); + } + + /** + * Clear existing EPG data for a source URL + */ + clearSourceData(sourceUrl: string): void { + this.deleteChannelsStmt.run(sourceUrl); + this.knownChannelIds.clear(); + } + + /** + * Insert a batch of channels + */ + insertChannels(channels: ParsedChannel[], sourceUrl: string): void { + const insertMany = this.db.transaction((channels: ParsedChannel[]) => { + for (const channel of channels) { + const displayName = channel.displayName?.[0]?.value || channel.id; + const iconUrl = channel.icon?.[0]?.src || null; + const url = channel.url?.[0] || null; + + this.insertChannelStmt.run( + channel.id, + displayName, + iconUrl, + url, + sourceUrl + ); + this.knownChannelIds.add(channel.id); + } + }); + + insertMany(channels); + } + + /** + * Insert a batch of programs + * Only inserts programs for known channels to avoid FK constraint failures + */ + insertPrograms(programs: ParsedProgram[]): number { + let insertedCount = 0; + + const insertMany = this.db.transaction((programs: ParsedProgram[]) => { + for (const prog of programs) { + // Skip if channel not known + if (!this.knownChannelIds.has(prog.channel)) continue; + + const title = prog.title?.[0]?.value || 'Unknown'; + const description = prog.desc?.[0]?.value || null; + const category = prog.category?.[0]?.value || null; + const iconUrl = prog.icon?.[0]?.src || null; + const rating = prog.rating?.[0]?.value || null; + const episodeNum = prog.episodeNum?.[0]?.value || null; + + try { + this.insertProgramStmt.run( + prog.channel, + prog.start, + prog.stop, + title, + description, + category, + iconUrl, + rating, + episodeNum + ); + insertedCount++; + } catch (err) { + // Skip individual failures (e.g., FK constraint) + } + } + }); + + insertMany(programs); + return insertedCount; + } + + /** + * Close the database connection + */ + close(): void { + this.db.close(); + } +} + /** * Parse XMLTV datetime format to ISO string * Format: YYYYMMDDHHmmss +HHMM or YYYYMMDDHHmmss @@ -46,7 +258,6 @@ const PROGRAM_BATCH_SIZE = 1000; function parseXmltvDate(dateStr: string): string { if (!dateStr) return ''; - // Extract date parts: 20231225060000 +0000 const match = dateStr.match( /^(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})\s*([+-]\d{4})?$/ ); @@ -55,11 +266,9 @@ function parseXmltvDate(dateStr: string): string { const [, year, month, day, hour, minute, second, tz] = match; - // Build ISO string let isoString = `${year}-${month}-${day}T${hour}:${minute}:${second}`; if (tz) { - // Convert +0000 to +00:00 isoString += `${tz.slice(0, 3)}:${tz.slice(3)}`; } else { isoString += 'Z'; @@ -73,22 +282,31 @@ function parseXmltvDate(dateStr: string): string { */ class StreamingEpgParser { private parser: SaxesParser; - private channels: EpgChannel[] = []; - private programs: EpgProgram[] = []; + private channels: ParsedChannel[] = []; + private programs: ParsedProgram[] = []; private totalChannels = 0; private totalPrograms = 0; + private skippedPrograms = 0; + + // Cutoff time for filtering old programs (calculated once at start) + private readonly cutoffTime: Date; // Current element being parsed - private currentChannel: Partial | null = null; - private currentProgram: Partial | null = null; + private currentChannel: Partial | null = null; + private currentProgram: Partial | null = null; private currentTextContent = ''; private currentLang = ''; // For nested elements private elementStack: string[] = []; - constructor(private onChannelsBatch: (channels: EpgChannel[]) => void, private onProgramsBatch: (programs: EpgProgram[]) => void, private onProgress: (channels: number, programs: number) => void) { + constructor( + private onChannelsBatch: (channels: ParsedChannel[]) => void, + private onProgramsBatch: (programs: ParsedProgram[]) => void, + private onProgress: (channels: number, programs: number) => void + ) { this.parser = new SaxesParser(); + this.cutoffTime = getProgramCutoffTime(); this.setupParser(); } @@ -121,12 +339,8 @@ class StreamingEpgParser { category: [], date: '', episodeNum: [], - previouslyShown: [], - subtitles: [], icon: [], rating: [], - credits: [], - audio: [], }; break; @@ -163,10 +377,8 @@ class StreamingEpgParser { case 'rating': if (this.currentProgram) { - // Rating has system attribute, value comes as child element const system = (tag.attributes['system'] as string) || ''; - // We'll complete this when we see the value element this.currentProgram.rating!.push({ system, value: '' }); } break; @@ -203,8 +415,7 @@ class StreamingEpgParser { if (text) this.currentChannel.url!.push(text); break; case 'channel': - // Channel complete, add to batch - this.channels.push(this.currentChannel as EpgChannel); + this.channels.push(this.currentChannel as ParsedChannel); this.totalChannels++; this.currentChannel = null; @@ -239,7 +450,6 @@ class StreamingEpgParser { this.currentProgram.date = text; break; case 'value': - // This is for rating value if ( this.elementStack.includes('rating') && this.currentProgram.rating!.length > 0 @@ -257,14 +467,18 @@ class StreamingEpgParser { } break; case 'programme': - // Program complete, add to batch - this.programs.push(this.currentProgram as EpgProgram); - this.totalPrograms++; - this.currentProgram = null; + // Skip programs that have already ended + if (isProgramExpired(this.currentProgram.stop!, this.cutoffTime)) { + this.skippedPrograms++; + } else { + this.programs.push(this.currentProgram as ParsedProgram); + this.totalPrograms++; - if (this.programs.length >= PROGRAM_BATCH_SIZE) { - this.flushPrograms(); + if (this.programs.length >= PROGRAM_BATCH_SIZE) { + this.flushPrograms(); + } } + this.currentProgram = null; break; } } @@ -298,20 +512,29 @@ class StreamingEpgParser { this.parser.write(chunk); } - finish(): { totalChannels: number; totalPrograms: number } { + finish(): { totalChannels: number; totalPrograms: number; skippedPrograms: number } { this.parser.close(); - // Flush any remaining data this.flushChannels(); this.flushPrograms(); + + if (this.skippedPrograms > 0) { + console.log( + loggerLabel, + `Skipped ${this.skippedPrograms} expired programs (ended more than ${SKIP_PROGRAMS_OLDER_THAN_HOURS}h ago)` + ); + } + return { totalChannels: this.totalChannels, totalPrograms: this.totalPrograms, + skippedPrograms: this.skippedPrograms, }; } } /** * Fetches and parses EPG data from URL using streaming + * Inserts directly into SQLite to avoid blocking main thread */ async function fetchAndParseEpgStreaming(url: string): Promise { const isGzipped = url.endsWith('.gz'); @@ -321,90 +544,176 @@ async function fetchAndParseEpgStreaming(url: string): Promise { `Fetching EPG from ${url} (gzipped: ${isGzipped})` ); - const response = await fetch(url.trim()); + // Create database connection in worker + const epgDb = new EpgDatabase(); - if (!response.ok) { - throw new Error(`HTTP error! status: ${response.status}`); - } + try { + // Clear existing data for this source + console.log(loggerLabel, `Clearing existing data for ${url}`); + epgDb.clearSourceData(url); - if (!response.body) { - throw new Error('Response body is null'); - } + const response = await fetch(url.trim()); - const parser = new StreamingEpgParser( - (channels) => { - const response: WorkerResponse = { - type: 'EPG_CHANNELS_BATCH', - channels, - }; - parentPort?.postMessage(response); - }, - (programs) => { - const response: WorkerResponse = { - type: 'EPG_PROGRAMS_BATCH', - programs, - }; - parentPort?.postMessage(response); - }, - (totalChannels, totalPrograms) => { - const response: WorkerResponse = { - type: 'EPG_PROGRESS', - stats: { totalChannels, totalPrograms }, - }; - parentPort?.postMessage(response); - } - ); - - // Convert web stream to Node.js stream - const nodeStream = Readable.fromWeb(response.body as any); - - return new Promise((resolve, reject) => { - let dataStream: Readable = nodeStream; - - if (isGzipped) { - const gunzip = createGunzip(); - dataStream = nodeStream.pipe(gunzip); - - gunzip.on('error', (err) => { - console.error(loggerLabel, 'Gunzip error:', err); - reject(err); - }); + if (!response.ok) { + throw new Error(`HTTP error! status: ${response.status}`); } - dataStream.on('data', (chunk: Buffer) => { - try { - parser.write(chunk.toString('utf-8')); - } catch (err) { - console.error(loggerLabel, 'Parse error:', err); - reject(err); - } - }); - - dataStream.on('end', () => { - try { - const stats = parser.finish(); - console.log( - loggerLabel, - `Parsing complete: ${stats.totalChannels} channels, ${stats.totalPrograms} programs` - ); + if (!response.body) { + throw new Error('Response body is null'); + } + const parser = new StreamingEpgParser( + (channels) => { + // Insert channels directly into database + epgDb.insertChannels(channels, url); + }, + (programs) => { + // Insert programs directly into database + epgDb.insertPrograms(programs); + }, + (totalChannels, totalPrograms) => { + // Send progress to main thread (lightweight) const response: WorkerResponse = { - type: 'EPG_COMPLETE', - url, - stats, + type: 'EPG_PROGRESS', + stats: { totalChannels, totalPrograms }, }; parentPort?.postMessage(response); - resolve(); - } catch (err) { - reject(err); } - }); + ); - dataStream.on('error', (err) => { - console.error(loggerLabel, 'Stream error:', err); - reject(err); + // Convert web stream to Node.js stream + const nodeStream = Readable.fromWeb(response.body as any); + + return new Promise((resolve, reject) => { + let dataStream: Readable = nodeStream; + + if (isGzipped) { + const gunzip = createGunzip(); + dataStream = nodeStream.pipe(gunzip); + + gunzip.on('error', (err) => { + console.error(loggerLabel, 'Gunzip error:', err); + epgDb.close(); + reject(err); + }); + } + + dataStream.on('data', (chunk: Buffer) => { + try { + parser.write(chunk.toString('utf-8')); + } catch (err) { + console.error(loggerLabel, 'Parse error:', err); + epgDb.close(); + reject(err); + } + }); + + dataStream.on('end', () => { + try { + const stats = parser.finish(); + console.log( + loggerLabel, + `Parsing complete: ${stats.totalChannels} channels, ${stats.totalPrograms} programs` + + (stats.skippedPrograms > 0 + ? ` (skipped ${stats.skippedPrograms} expired)` + : '') + ); + + // Close database connection + epgDb.close(); + + const response: WorkerResponse = { + type: 'EPG_COMPLETE', + url, + stats: { + totalChannels: stats.totalChannels, + totalPrograms: stats.totalPrograms, + }, + }; + parentPort?.postMessage(response); + resolve(); + } catch (err) { + epgDb.close(); + reject(err); + } + }); + + dataStream.on('error', (err) => { + console.error(loggerLabel, 'Stream error:', err); + epgDb.close(); + reject(err); + }); }); - }); + } catch (error) { + epgDb.close(); + throw error; + } +} + +/** + * Clears all EPG data from the database + * Runs in worker thread to avoid blocking main thread + */ +function clearAllEpgData(): void { + const dbPath = getDatabasePath(); + const db = new Database(dbPath); + + try { + console.log(loggerLabel, 'Clearing all EPG data...'); + + // Delete programs first (foreign key constraint) + db.exec('DELETE FROM epg_programs'); + // Then delete channels + db.exec('DELETE FROM epg_channels'); + + console.log(loggerLabel, 'All EPG data cleared'); + + const response: WorkerResponse = { type: 'CLEAR_COMPLETE' }; + parentPort?.postMessage(response); + } catch (error) { + console.error(loggerLabel, 'Error clearing EPG data:', error); + const errorResponse: WorkerResponse = { + type: 'EPG_ERROR', + error: error instanceof Error ? error.message : String(error), + }; + parentPort?.postMessage(errorResponse); + } finally { + db.close(); + } +} + +/** + * Cleans up expired EPG programs from the database + * Runs in worker thread to avoid blocking main thread + */ +function cleanupExpiredPrograms(hoursToKeep: number): void { + const dbPath = getDatabasePath(); + const db = new Database(dbPath); + + try { + const cutoff = new Date( + Date.now() - hoursToKeep * 60 * 60 * 1000 + ).toISOString(); + + console.log(loggerLabel, `Cleaning up programs older than ${hoursToKeep}h (before ${cutoff})...`); + + const stmt = db.prepare('DELETE FROM epg_programs WHERE stop <= ?'); + const result = stmt.run(cutoff); + + console.log(loggerLabel, `Cleaned up ${result.changes} expired programs`); + + const response: WorkerResponse = { type: 'CLEANUP_COMPLETE' }; + parentPort?.postMessage(response); + } catch (error) { + console.error(loggerLabel, 'Error cleaning up expired programs:', error); + const errorResponse: WorkerResponse = { + type: 'EPG_ERROR', + error: error instanceof Error ? error.message : String(error), + }; + parentPort?.postMessage(errorResponse); + } finally { + db.close(); + } } /** @@ -417,7 +726,11 @@ if (parentPort) { message.type === 'FETCH_EPG' || message.type === 'FORCE_FETCH' ) { - await fetchAndParseEpgStreaming(message.url); + await fetchAndParseEpgStreaming(message.url!); + } else if (message.type === 'CLEAR_EPG') { + clearAllEpgData(); + } else if (message.type === 'CLEANUP_EXPIRED') { + cleanupExpiredPrograms(message.hoursToKeep ?? 24); } } catch (error) { console.error(loggerLabel, 'Worker error:', error); diff --git a/global.d.ts b/global.d.ts index f84587d6d..4f3adfed2 100644 --- a/global.d.ts +++ b/global.d.ts @@ -57,6 +57,15 @@ declare global { forceFetchEpg: ( url: string ) => Promise<{ success: boolean; message?: string }>; + clearEpgData: () => Promise<{ success: boolean }>; + checkEpgFreshness: ( + urls: string[], + maxAgeHours?: number + ) => Promise<{ staleUrls: string[]; freshUrls: string[] }>; + searchEpgPrograms: ( + searchTerm: string, + limit?: number + ) => Promise; updateSettings: (settings: any) => Promise; setMpvPlayerPath: (mpvPlayerPath: string) => Promise; setVlcPlayerPath: (vlcPlayerPath: string) => Promise; @@ -173,6 +182,16 @@ declare global { }) => void ) => void; getLocalIpAddresses: () => Promise; + // EPG progress listener + onEpgProgress?: ( + callback: (data: { + url: string; + status: 'loading' | 'complete' | 'error'; + stats?: { totalChannels: number; totalPrograms: number }; + error?: string; + }) => void + ) => void; + dbDeleteAllPlaylists: () => Promise<{ success: boolean }>; }; process: NodeJS.Process; require: NodeRequire; diff --git a/libs/shared/database/src/lib/connection.ts b/libs/shared/database/src/lib/connection.ts index 5aabfd38a..b25863ca6 100644 --- a/libs/shared/database/src/lib/connection.ts +++ b/libs/shared/database/src/lib/connection.ts @@ -105,6 +105,57 @@ const CREATE_TABLE_STATEMENTS = [ `CREATE UNIQUE INDEX IF NOT EXISTS recently_viewed_content_playlist_unique ON recently_viewed(content_id, playlist_id)`, `CREATE INDEX IF NOT EXISTS recently_viewed_playlist_idx ON recently_viewed(playlist_id)`, `CREATE INDEX IF NOT EXISTS recently_viewed_viewed_at_idx ON recently_viewed(viewed_at)`, + // EPG tables + `CREATE TABLE IF NOT EXISTS epg_channels ( + id TEXT PRIMARY KEY, + display_name TEXT NOT NULL, + icon_url TEXT, + url TEXT, + source_url TEXT NOT NULL, + updated_at TEXT DEFAULT (datetime('now')) + )`, + `CREATE TABLE IF NOT EXISTS epg_programs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + channel_id TEXT NOT NULL, + start TEXT NOT NULL, + stop TEXT NOT NULL, + title TEXT NOT NULL, + description TEXT, + category TEXT, + icon_url TEXT, + rating TEXT, + episode_num TEXT, + FOREIGN KEY (channel_id) REFERENCES epg_channels(id) ON DELETE CASCADE + )`, + // EPG indexes + `CREATE INDEX IF NOT EXISTS idx_epg_channels_source ON epg_channels(source_url)`, + `CREATE INDEX IF NOT EXISTS idx_epg_channels_name ON epg_channels(display_name)`, + `CREATE INDEX IF NOT EXISTS idx_epg_programs_channel ON epg_programs(channel_id)`, + `CREATE INDEX IF NOT EXISTS idx_epg_programs_start ON epg_programs(start)`, + `CREATE INDEX IF NOT EXISTS idx_epg_programs_time_range ON epg_programs(channel_id, start, stop)`, + // FTS5 virtual table for full-text search on EPG programs + `CREATE VIRTUAL TABLE IF NOT EXISTS epg_programs_fts USING fts5( + title, + description, + category, + content='epg_programs', + content_rowid='id' + )`, + // Triggers to keep FTS index in sync with epg_programs table + `CREATE TRIGGER IF NOT EXISTS epg_programs_ai AFTER INSERT ON epg_programs BEGIN + INSERT INTO epg_programs_fts(rowid, title, description, category) + VALUES (new.id, new.title, new.description, new.category); + END`, + `CREATE TRIGGER IF NOT EXISTS epg_programs_ad AFTER DELETE ON epg_programs BEGIN + INSERT INTO epg_programs_fts(epg_programs_fts, rowid, title, description, category) + VALUES ('delete', old.id, old.title, old.description, old.category); + END`, + `CREATE TRIGGER IF NOT EXISTS epg_programs_au AFTER UPDATE ON epg_programs BEGIN + INSERT INTO epg_programs_fts(epg_programs_fts, rowid, title, description, category) + VALUES ('delete', old.id, old.title, old.description, old.category); + INSERT INTO epg_programs_fts(rowid, title, description, category) + VALUES (new.id, new.title, new.description, new.category); + END`, ]; /** diff --git a/libs/shared/database/src/lib/schema.ts b/libs/shared/database/src/lib/schema.ts index 968c96893..3e21f0167 100644 --- a/libs/shared/database/src/lib/schema.ts +++ b/libs/shared/database/src/lib/schema.ts @@ -123,6 +123,51 @@ export const favorites = sqliteTable( }) ); +// EPG Channels table +export const epgChannels = sqliteTable( + 'epg_channels', + { + id: text('id').primaryKey(), // Channel ID from EPG source + displayName: text('display_name').notNull(), + iconUrl: text('icon_url'), + url: text('url'), + sourceUrl: text('source_url').notNull(), // Which EPG URL this came from + updatedAt: text('updated_at').default(sql`CURRENT_TIMESTAMP`), + }, + (table) => ({ + sourceIdx: index('idx_epg_channels_source').on(table.sourceUrl), + nameIdx: index('idx_epg_channels_name').on(table.displayName), + }) +); + +// EPG Programs table +export const epgPrograms = sqliteTable( + 'epg_programs', + { + id: integer('id').primaryKey({ autoIncrement: true }), + channelId: text('channel_id') + .notNull() + .references(() => epgChannels.id, { onDelete: 'cascade' }), + start: text('start').notNull(), // ISO datetime + stop: text('stop').notNull(), // ISO datetime + title: text('title').notNull(), + description: text('description'), + category: text('category'), + iconUrl: text('icon_url'), + rating: text('rating'), + episodeNum: text('episode_num'), + }, + (table) => ({ + channelIdx: index('idx_epg_programs_channel').on(table.channelId), + startIdx: index('idx_epg_programs_start').on(table.start), + timeRangeIdx: index('idx_epg_programs_time_range').on( + table.channelId, + table.start, + table.stop + ), + }) +); + // Type exports for TypeScript export type Playlist = typeof playlists.$inferSelect; export type NewPlaylist = typeof playlists.$inferInsert; @@ -138,3 +183,9 @@ export type NewRecentlyViewed = typeof recentlyViewed.$inferInsert; export type Favorite = typeof favorites.$inferSelect; export type NewFavorite = typeof favorites.$inferInsert; + +export type EpgChannel = typeof epgChannels.$inferSelect; +export type NewEpgChannel = typeof epgChannels.$inferInsert; + +export type EpgProgramDb = typeof epgPrograms.$inferSelect; +export type NewEpgProgramDb = typeof epgPrograms.$inferInsert;