feat(epg): worker-based EPG handlers and fetch logic

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": {

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) =>

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<string, EpgChannelWithPrograms> =
+        new Map();
+    private static fetchedUrls: Set<string> = new Set();
+    private static workers: Map<string, Worker> = 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<void> {
+        // 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<string, EpgProgram[]>();
+
+        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
-});

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<EpgData> {
+    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<string, unknown>) => ({
+                    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<string, unknown>) => ({
+                    ...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!');
+}

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()) {

global.d.ts

@@ -39,8 +39,18 @@ declare global {
                 origin?: string
             ) => void;
             autoUpdatePlaylists: (playlists: Playlist[]) => Promise<Playlist[]>;
-            fetchEpg: (urls: string[]) => Promise<any>;
+            fetchEpg: (
+                urls: string[]
+            ) => Promise<{ success: boolean; message?: string }>;
             getChannelPrograms: (channelId: string) => Promise<any>;
+            getEpgChannels: () => Promise<any>;
+            getEpgChannelsByRange: (
+                skip: number,
+                limit: number
+            ) => Promise<any>;
+            forceFetchEpg: (
+                url: string
+            ) => Promise<{ success: boolean; message?: string }>;
             updateSettings: (settings: any) => Promise<void>;
             setMpvPlayerPath: (mpvPlayerPath: string) => Promise<void>;
             setVlcPlayerPath: (vlcPlayerPath: string) => Promise<void>;

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<EpgProgram[]>('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',
+        });
     }
 }