diff --git a/apps/electron-backend/project.json b/apps/electron-backend/project.json index 288f4dc69..152ef13d8 100644 --- a/apps/electron-backend/project.json +++ b/apps/electron-backend/project.json @@ -17,7 +17,20 @@ "outputPath": "dist/apps/electron-backend", "main": "apps/electron-backend/src/main.ts", "tsConfig": "apps/electron-backend/tsconfig.app.json", - "assets": ["apps/electron-backend/src/assets"] + "assets": [ + "apps/electron-backend/src/assets", + { + "input": "apps/electron-backend/src/app/workers", + "glob": "*.worker.ts", + "output": "workers" + } + ], + "additionalEntryPoints": [ + { + "entryName": "workers/epg-parser.worker", + "entryPath": "apps/electron-backend/src/app/workers/epg-parser.worker.ts" + } + ] }, "configurations": { "production": { diff --git a/apps/electron-backend/src/app/api/main.preload.ts b/apps/electron-backend/src/app/api/main.preload.ts index ba4257ab3..547c84cff 100644 --- a/apps/electron-backend/src/app/api/main.preload.ts +++ b/apps/electron-backend/src/app/api/main.preload.ts @@ -50,6 +50,10 @@ contextBridge.exposeInMainWorld('electron', { ipcRenderer.invoke('FETCH_EPG', { url: urls }), getChannelPrograms: (channelId: string) => ipcRenderer.invoke('GET_CHANNEL_PROGRAMS', { channelId }), + getEpgChannels: () => ipcRenderer.invoke('EPG_GET_CHANNELS'), + getEpgChannelsByRange: (skip: number, limit: number) => + ipcRenderer.invoke('EPG_GET_CHANNELS_BY_RANGE', { skip, limit }), + forceFetchEpg: (url: string) => ipcRenderer.invoke('EPG_FORCE_FETCH', url), setMpvPlayerPath: (mpvPlayerPath: string) => ipcRenderer.invoke('SET_MPV_PLAYER_PATH', mpvPlayerPath), setVlcPlayerPath: (vlcPlayerPath: string) => diff --git a/apps/electron-backend/src/app/events/epg.events.ts b/apps/electron-backend/src/app/events/epg.events.ts index d54676598..3f13fa26e 100644 --- a/apps/electron-backend/src/app/events/epg.events.ts +++ b/apps/electron-backend/src/app/events/epg.events.ts @@ -1,15 +1,331 @@ import { ipcMain } from 'electron'; +import * as path from 'path'; +import { EpgChannelWithPrograms, EpgData, EpgProgram } from 'shared-interfaces'; +import { Worker } from 'worker_threads'; +/** + * EPG Events Handler + * Manages EPG data fetching and querying using worker threads + */ 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]'; + + /** + * Bootstrap EPG events + */ static bootstrapEpgEvents(): Electron.IpcMain { + console.log('[EPG Events] Bootstrapping EPG events...'); + + // 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 }) => { + return this.handleGetChannelPrograms(args.channelId); + } + ); + + // Get all channels + ipcMain.handle('EPG_GET_CHANNELS', async () => { + return this.epgData; + }); + + // Get channels by range (pagination) + ipcMain.handle( + 'EPG_GET_CHANNELS_BY_RANGE', + 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) => { + this.fetchedUrls.delete(url); + return await this.handleFetchEpg([url]); + }); + return ipcMain; } + + /** + * Handle EPG fetch from URLs + */ + private static async handleFetchEpg( + urls: string[] + ): Promise<{ success: boolean; message?: 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)); + + try { + await Promise.all(promises); + return { success: true }; + } catch (error) { + console.error('[EPG Events] Error fetching EPG:', error); + return { + success: false, + message: error instanceof Error ? error.message : String(error), + }; + } + } + + /** + * Fetch EPG from a single URL using worker thread + */ + private static async fetchEpgFromUrl(url: string): Promise { + // Skip if already fetched + if (this.fetchedUrls.has(url)) { + console.log(this.loggerLabel, 'URL already fetched:', url); + return; + } + + return new Promise((resolve, reject) => { + const workerPath = path.join( + __dirname, + 'workers', + 'epg-parser.worker.js' + ); + + console.log(this.loggerLabel, 'Creating worker for:', url); + console.log(this.loggerLabel, 'Worker path:', workerPath); + + let worker: Worker; + try { + worker = new Worker(workerPath); + console.log(this.loggerLabel, 'Worker created successfully'); + } catch (error) { + console.error( + this.loggerLabel, + 'Failed to create worker:', + error + ); + reject(error); + return; + } + + this.workers.set(url, worker); + + worker.on( + 'message', + (message: { + type: string; + data?: EpgData; + error?: string; + url?: string; + }) => { + if (message.type === 'READY') { + console.log( + this.loggerLabel, + 'Worker ready, sending fetch command' + ); + worker.postMessage({ type: 'FETCH_EPG', url }); + } else if (message.type === 'EPG_PARSED') { + console.log( + this.loggerLabel, + 'EPG parsed in worker, merging data...' + ); + if (message.data) { + this.mergeEpgData(message.data); + this.fetchedUrls.add(url); + } + worker.terminate(); + this.workers.delete(url); + resolve(); + } else if (message.type === 'EPG_ERROR') { + console.error( + this.loggerLabel, + 'Worker error:', + message.error + ); + worker.terminate(); + this.workers.delete(url); + reject(new Error(message.error || 'Unknown error')); + } + } + ); + + worker.on('error', (error) => { + console.error(this.loggerLabel, 'Worker error event:', error); + worker.terminate(); + this.workers.delete(url); + reject(error); + }); + + worker.on('exit', (code) => { + if (code !== 0) { + console.error( + this.loggerLabel, + `Worker stopped with exit code ${code}` + ); + } + }); + }); + } + + /** + * Merge EPG data (optimized for large datasets) + */ + private static mergeEpgData(newData: EpgData): void { + console.log( + this.loggerLabel, + `Merging ${newData.channels.length} channels and ${newData.programs.length} programs...` + ); + + // Merge channels (avoid duplicates) + const existingChannelIds = new Set( + this.epgData.channels.map((c) => c.id) + ); + const newChannels = newData.channels.filter( + (c) => !existingChannelIds.has(c.id) + ); + + // Use concat instead of spread for large arrays + this.epgData.channels = this.epgData.channels.concat(newChannels); + + // Merge programs in chunks to avoid stack overflow + const chunkSize = 10000; + for (let i = 0; i < newData.programs.length; i += chunkSize) { + const chunk = newData.programs.slice(i, i + chunkSize); + this.epgData.programs = this.epgData.programs.concat(chunk); + } + + // Rebuild merged data structure + this.rebuildMergedData(); + + console.log( + this.loggerLabel, + `Merged EPG data: ${this.epgData.channels.length} channels, ${this.epgData.programs.length} programs` + ); + } + + /** + * Rebuild merged data structure (channels with programs) + * Optimized for large datasets + */ + private static rebuildMergedData(): void { + console.log(this.loggerLabel, 'Rebuilding merged data structure...'); + this.epgDataMerged.clear(); + + // Create channel lookup map for O(1) access + const channelMap = new Map(this.epgData.channels.map((c) => [c.id, c])); + + // 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, + }); + } + } + + console.log( + this.loggerLabel, + `Rebuild complete. ${this.epgDataMerged.size} channels with programs` + ); + } + + /** + * Get programs for a specific channel + */ + private static handleGetChannelPrograms(channelId: string): EpgProgram[] { + console.log('[EPG Events] Getting programs for channel:', channelId); + console.log( + '[EPG Events] Total merged channels:', + this.epgDataMerged.size + ); + + // First try exact ID match + let channelData = this.epgDataMerged.get(channelId); + + // If not found, try to find by display name + if (!channelData) { + console.log( + '[EPG Events] No exact match, searching by display name...' + ); + for (const [id, channel] of this.epgDataMerged.entries()) { + const displayNames = channel.displayName.map((d) => + d.value.toLowerCase() + ); + if ( + displayNames.some( + (name) => + name === channelId.toLowerCase() || + name.includes(channelId.toLowerCase()) || + channelId.toLowerCase().includes(name) + ) + ) { + console.log( + '[EPG Events] Found match by name:', + id, + '→', + channel.displayName[0]?.value + ); + channelData = channel; + break; + } + } + } + + console.log( + '[EPG Events] Found programs:', + channelData?.programs?.length || 0 + ); + return channelData?.programs || []; + } + + /** + * Get channels by range (for pagination) + */ + private static handleGetChannelsByRange( + skip: number, + limit: number + ): EpgChannelWithPrograms[] { + const channels = Array.from(this.epgDataMerged.values()); + return channels.slice(skip, skip + limit); + } + + /** + * Clear all EPG data + */ + static clearEpgData(): void { + this.epgData = { channels: [], programs: [] }; + this.epgDataMerged.clear(); + this.fetchedUrls.clear(); + + // Terminate all workers + this.workers.forEach((worker) => worker.terminate()); + this.workers.clear(); + + console.log(this.loggerLabel, 'EPG data cleared'); + } } - -ipcMain.handle('GET_CHANNEL_PROGRAMS', async (event, channelId) => { - // fetch EPG data for the given channelId -}); - -ipcMain.handle('FETCH_EPG', async (event, urls: string[]) => { - // fetch EPG data from the given URLs -}); diff --git a/apps/electron-backend/src/app/workers/epg-parser.worker.ts b/apps/electron-backend/src/app/workers/epg-parser.worker.ts new file mode 100644 index 000000000..52a2f33d2 --- /dev/null +++ b/apps/electron-backend/src/app/workers/epg-parser.worker.ts @@ -0,0 +1,157 @@ +import axios from 'axios'; +import { parse as parseEpg } from 'epg-parser'; +import { EpgData, EpgProgram } from 'shared-interfaces'; +import { parentPort } from 'worker_threads'; +import { gunzipSync } from 'zlib'; + +/** + * EPG Parser Worker + * Runs in a separate thread to avoid blocking the main process + * Compatible with epg-parser v0.4.0 + */ + +interface WorkerMessage { + type: 'FETCH_EPG' | 'FORCE_FETCH'; + url: string; +} + +interface WorkerResponse { + type: 'EPG_PARSED' | 'EPG_ERROR' | 'EPG_PROGRESS' | 'READY'; + data?: EpgData; + error?: string; + url?: string; +} + +const loggerLabel = '[EPG Worker]'; + +/** + * Fetches and parses EPG data from URL + */ +async function fetchAndParseEpg(url: string): Promise { + try { + console.log(loggerLabel, 'Fetching EPG from:', url); + + const isGzipped = url.endsWith('.gz'); + const axiosConfig = isGzipped + ? { responseType: 'arraybuffer' as const } + : {}; + + const response = await axios.get(url.trim(), axiosConfig); + console.log( + loggerLabel, + 'EPG data fetched, size:', + response.data.length + ); + + let xmlString: string; + + if (isGzipped) { + console.log(loggerLabel, 'Unzipping...'); + const buffer = Buffer.from(response.data); + const decompressed = gunzipSync(buffer); + xmlString = decompressed.toString('utf-8'); + } else { + xmlString = response.data; + } + + console.log(loggerLabel, 'Parsing EPG XML...'); + const parsed = parseEpg(xmlString); + + // Normalize data to match our interface + const normalized: EpgData = { + channels: parsed.channels.map( + (channel: Record) => ({ + id: channel.id as string, + displayName: + (channel.displayName as { + lang: string; + value: string; + }[]) || [], + icon: + (channel.icon as { + src: string; + width?: number; + height?: number; + }[]) || [], + url: (channel.url as string[]) || [], + }) + ), + programs: parsed.programs.map( + (program: Record) => ({ + ...program, + start: program.start as string, // Already ISO string in v0.4.0 + stop: program.stop as string, // Already ISO string in v0.4.0 + date: + typeof program.date === 'string' + ? program.date + : (program.date as string[])?.[0] || '', + }) + ) as EpgProgram[], + }; + + console.log( + loggerLabel, + `Parsed ${normalized.channels.length} channels and ${normalized.programs.length} programs` + ); + + return normalized; + } catch (error) { + console.error( + loggerLabel, + 'Error fetching EPG:', + error instanceof Error ? error.message : String(error) + ); + throw error; + } +} + +/** + * Worker message handler + */ +if (parentPort) { + console.log(loggerLabel, 'Worker initialized and ready'); + + parentPort.on('message', async (message: WorkerMessage) => { + console.log(loggerLabel, 'Received message:', message.type); + try { + if ( + message.type === 'FETCH_EPG' || + message.type === 'FORCE_FETCH' + ) { + console.log( + loggerLabel, + 'Starting EPG fetch for:', + message.url + ); + const parsedData = await fetchAndParseEpg(message.url); + + const response: WorkerResponse = { + type: 'EPG_PARSED', + data: parsedData, + url: message.url, + }; + + console.log( + loggerLabel, + 'Sending parsed data back to main thread' + ); + parentPort?.postMessage(response); + } + } catch (error) { + console.error(loggerLabel, 'Worker error:', error); + const errorResponse: WorkerResponse = { + type: 'EPG_ERROR', + error: error instanceof Error ? error.message : String(error), + url: message.url, + }; + + parentPort?.postMessage(errorResponse); + } + }); + + // Notify parent that worker is ready + console.log(loggerLabel, 'Sending READY message'); + parentPort.postMessage({ type: 'READY' }); +} else { + console.error(loggerLabel, 'parentPort is not available!'); +} diff --git a/apps/electron-backend/src/main.ts b/apps/electron-backend/src/main.ts index 2ff9eac46..da70560e7 100644 --- a/apps/electron-backend/src/main.ts +++ b/apps/electron-backend/src/main.ts @@ -4,6 +4,7 @@ import App from './app/app'; import { initDatabase } from './app/database/connection'; import DatabaseEvents from './app/events/database.events'; import ElectronEvents from './app/events/electron.events'; +import EpgEvents from './app/events/epg.events'; import PlayerEvents from './app/events/player.events'; import PlaylistEvents from './app/events/playlist.events'; import SettingsEvents from './app/events/setttings.events'; @@ -38,6 +39,7 @@ export default class Main { StalkerEvents.bootstrapStalkerEvents(); XtreamEvents.bootstrapXtreamEvents(); DatabaseEvents.bootstrapDatabaseEvents(); + EpgEvents.bootstrapEpgEvents(); // initialize auto updater service if (!App.isDevelopmentMode()) { diff --git a/global.d.ts b/global.d.ts index c48fc02b5..f3ef4b7bb 100644 --- a/global.d.ts +++ b/global.d.ts @@ -39,8 +39,18 @@ declare global { origin?: string ) => void; autoUpdatePlaylists: (playlists: Playlist[]) => Promise; - fetchEpg: (urls: string[]) => Promise; + fetchEpg: ( + urls: string[] + ) => Promise<{ success: boolean; message?: string }>; getChannelPrograms: (channelId: string) => Promise; + getEpgChannels: () => Promise; + getEpgChannelsByRange: ( + skip: number, + limit: number + ) => Promise; + forceFetchEpg: ( + url: string + ) => Promise<{ success: boolean; message?: string }>; updateSettings: (settings: any) => Promise; setMpvPlayerPath: (mpvPlayerPath: string) => Promise; setVlcPlayerPath: (vlcPlayerPath: string) => Promise; diff --git a/libs/services/src/lib/epg.service.ts b/libs/services/src/lib/epg.service.ts index a00c48c10..61379bdd2 100644 --- a/libs/services/src/lib/epg.service.ts +++ b/libs/services/src/lib/epg.service.ts @@ -2,9 +2,8 @@ import { inject, Injectable } from '@angular/core'; import { MatSnackBar } from '@angular/material/snack-bar'; import { Store } from '@ngrx/store'; import { TranslateService } from '@ngx-translate/core'; -import { invoke } from '@tauri-apps/api/core'; import * as PlaylistActions from 'm3u-state'; -import { BehaviorSubject, from } from 'rxjs'; +import { BehaviorSubject, from, of } from 'rxjs'; import { catchError, map, tap } from 'rxjs/operators'; import { EpgProgram } from 'shared-interfaces'; @@ -37,17 +36,22 @@ export class EpgService { const validUrls = urls.filter((url) => url?.trim()); if (validUrls.length === 0) return; - from(invoke('fetch_epg', { url: validUrls })) + from(window.electron.fetchEpg(validUrls)) .pipe( - tap(() => { - this.epgAvailable.next(true); - this.showSuccessSnackbar(); + tap((result) => { + if (result.success) { + this.epgAvailable.next(true); + this.showSuccessSnackbar(); + } else { + this.epgAvailable.next(false); + this.showErrorSnackbar(result.message); + } }), catchError((err) => { console.error('EPG fetch error:', err); this.epgAvailable.next(false); this.showErrorSnackbar(); - throw err; + return of(null); }) ) .subscribe(); @@ -59,12 +63,10 @@ export class EpgService { getChannelPrograms(channelId: string): void { if (!this.isDesktop) return; console.log('Fetching EPG for channel ID:', channelId); - from(invoke('get_channel_programs', { channelId })) + + from(window.electron.getChannelPrograms(channelId)) .pipe( - tap((programs) => { - console.log('Received programs:', programs); - }), - map((programs) => + map((programs: EpgProgram[]) => programs.map((program) => ({ ...program, start: new Date(program.start).toISOString(), @@ -75,7 +77,7 @@ export class EpgService { console.error('EPG get programs error:', err); this.showErrorSnackbar(); this.currentEpgPrograms.next([]); - throw err; + return of([]); }) ) .subscribe((programs) => { @@ -85,7 +87,6 @@ export class EpgService { }) ); this.currentEpgPrograms.next(programs); - console.log('Updated programs:', programs); // Debug log }); } @@ -116,14 +117,11 @@ export class EpgService { /** * Shows error snackbar */ - private showErrorSnackbar(): void { - this.snackBar.open( - this.translate.instant('EPG.ERROR'), - this.translate.instant('CLOSE'), - { - duration: 2000, - horizontalPosition: 'start', - } - ); + private showErrorSnackbar(message?: string): void { + const errorMessage = message || this.translate.instant('EPG.ERROR'); + this.snackBar.open(errorMessage, this.translate.instant('CLOSE'), { + duration: 3000, + horizontalPosition: 'start', + }); } }