diff --git a/apps/electron-backend/src/app/events/epg.events.ts b/apps/electron-backend/src/app/events/epg.events.ts index 79d738389..d00313d55 100644 --- a/apps/electron-backend/src/app/events/epg.events.ts +++ b/apps/electron-backend/src/app/events/epg.events.ts @@ -1,6 +1,11 @@ import { app, ipcMain } from 'electron'; import * as path from 'path'; -import { EpgChannelWithPrograms, EpgData, EpgProgram } from 'shared-interfaces'; +import { + EpgChannel, + EpgChannelWithPrograms, + EpgData, + EpgProgram, +} from 'shared-interfaces'; import { Worker } from 'worker_threads'; import { pathToFileURL } from 'url'; @@ -117,7 +122,13 @@ export default class EpgEvents { try { // Worker threads require file:// URLs wrapped in URL object for packaged apps const workerURL = pathToFileURL(workerPath); - worker = new Worker(workerURL); + worker = new Worker(workerURL, { + // Increase memory limits for large EPG files + resourceLimits: { + maxOldGenerationSizeMb: 4096, // 4GB for parsed data + maxYoungGenerationSizeMb: 512, // 512MB for temporary allocations + }, + }); } catch (error) { console.error( this.loggerLabel, @@ -134,29 +145,64 @@ export default class EpgEvents { 'message', (message: { type: string; - data?: EpgData; + channels?: EpgChannel[]; + programs?: EpgProgram[]; error?: string; url?: string; + stats?: { totalChannels: number; totalPrograms: number }; }) => { - if (message.type === 'READY') { - worker.postMessage({ type: 'FETCH_EPG', url }); - } else if (message.type === 'EPG_PARSED') { - if (message.data) { - this.mergeEpgData(message.data); + switch (message.type) { + case 'READY': + worker.postMessage({ type: 'FETCH_EPG', url }); + break; + + case 'EPG_CHANNELS_BATCH': + // Add channels incrementally + if (message.channels) { + this.addChannelsBatch(message.channels); + } + 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) { + console.log( + this.loggerLabel, + `Progress: ${message.stats.totalChannels} channels, ${message.stats.totalPrograms} programs` + ); + } + break; + + case 'EPG_COMPLETE': + console.log( + this.loggerLabel, + `EPG parsing complete for ${url}:`, + message.stats + ); + this.rebuildMergedData(); 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.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; } } ); @@ -180,29 +226,21 @@ export default class EpgEvents { } /** - * Merge EPG data (optimized for large datasets) + * Add a batch of channels (streaming mode) */ - private static mergeEpgData(newData: EpgData): void { - // 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 + 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); + } - // 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(); + /** + * Add a batch of programs (streaming mode) + */ + 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); } /** @@ -247,7 +285,7 @@ export default class EpgEvents { // If not found, try to find by display name if (!channelData) { - for (const [id, channel] of this.epgDataMerged.entries()) { + for (const [, channel] of this.epgDataMerged.entries()) { const displayNames = channel.displayName.map((d) => d.value.toLowerCase() ); 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 cbf56a20e..2943c24b8 100644 --- a/apps/electron-backend/src/app/workers/epg-parser.worker.ts +++ b/apps/electron-backend/src/app/workers/epg-parser.worker.ts @@ -1,12 +1,13 @@ -import { parse as parseEpg } from 'epg-parser'; -import { EpgData, EpgProgram } from 'shared-interfaces'; +import { EpgChannel, EpgProgram } from 'shared-interfaces'; import { parentPort } from 'worker_threads'; -import { gunzipSync } from 'zlib'; +import { createGunzip } from 'zlib'; +import { SaxesParser, SaxesTagPlain } from 'saxes'; +import { Readable } from 'stream'; /** - * EPG Parser Worker - * Runs in a separate thread to avoid blocking the main process - * Compatible with epg-parser v0.4.0 + * Streaming EPG Parser Worker + * Uses SAX parsing to process XML incrementally without loading entire file into memory. + * Supports both regular and gzipped EPG files. */ interface WorkerMessage { @@ -15,84 +16,395 @@ interface WorkerMessage { } interface WorkerResponse { - type: 'EPG_PARSED' | 'EPG_ERROR' | 'EPG_PROGRESS' | 'READY'; - data?: EpgData; + type: + | 'EPG_CHANNELS_BATCH' + | 'EPG_PROGRAMS_BATCH' + | 'EPG_COMPLETE' + | 'EPG_ERROR' + | 'EPG_PROGRESS' + | 'READY'; + channels?: EpgChannel[]; + programs?: EpgProgram[]; error?: string; url?: string; + stats?: { + totalChannels: number; + totalPrograms: number; + }; } const loggerLabel = '[EPG Worker]'; +// Batch size for sending data back to main thread +const CHANNEL_BATCH_SIZE = 100; +const PROGRAM_BATCH_SIZE = 1000; + /** - * Fetches and parses EPG data from URL + * Parse XMLTV datetime format to ISO string + * Format: YYYYMMDDHHmmss +HHMM or YYYYMMDDHHmmss */ -async function fetchAndParseEpg(url: string): Promise { - try { - const isGzipped = url.endsWith('.gz'); +function parseXmltvDate(dateStr: string): string { + if (!dateStr) return ''; - // Use native fetch (available in Node.js 18+) - const response = await fetch(url.trim()); + // Extract date parts: 20231225060000 +0000 + const match = dateStr.match( + /^(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})\s*([+-]\d{4})?$/ + ); - if (!response.ok) { - throw new Error(`HTTP error! status: ${response.status}`); + if (!match) return dateStr; + + 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'; + } + + return isoString; +} + +/** + * Streaming EPG parser using SAX + */ +class StreamingEpgParser { + private parser: SaxesParser; + private channels: EpgChannel[] = []; + private programs: EpgProgram[] = []; + private totalChannels = 0; + private totalPrograms = 0; + + // Current element being parsed + 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) { + this.parser = new SaxesParser(); + this.setupParser(); + } + + private setupParser(): void { + this.parser.on('opentag', (tag: SaxesTagPlain) => { + this.elementStack.push(tag.name); + this.currentTextContent = ''; + + switch (tag.name) { + case 'channel': + this.currentChannel = { + id: (tag.attributes['id'] as string) || '', + displayName: [], + icon: [], + url: [], + }; + break; + + case 'programme': + this.currentProgram = { + start: parseXmltvDate( + (tag.attributes['start'] as string) || '' + ), + stop: parseXmltvDate( + (tag.attributes['stop'] as string) || '' + ), + channel: (tag.attributes['channel'] as string) || '', + title: [], + desc: [], + category: [], + date: '', + episodeNum: [], + previouslyShown: [], + subtitles: [], + icon: [], + rating: [], + credits: [], + audio: [], + }; + break; + + case 'icon': + if (this.currentChannel) { + this.currentChannel.icon!.push({ + src: (tag.attributes['src'] as string) || '', + width: tag.attributes['width'] + ? parseInt(tag.attributes['width'] as string) + : undefined, + height: tag.attributes['height'] + ? parseInt(tag.attributes['height'] as string) + : undefined, + }); + } else if (this.currentProgram) { + this.currentProgram.icon!.push({ + src: (tag.attributes['src'] as string) || '', + width: tag.attributes['width'] + ? parseInt(tag.attributes['width'] as string) + : undefined, + height: tag.attributes['height'] + ? parseInt(tag.attributes['height'] as string) + : undefined, + }); + } + break; + + case 'display-name': + case 'title': + case 'desc': + case 'category': + this.currentLang = (tag.attributes['lang'] as string) || ''; + break; + + 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; + + case 'episode-num': + if (this.currentProgram) { + const system = + (tag.attributes['system'] as string) || ''; + this.currentProgram.episodeNum!.push({ + system, + value: '', + }); + } + break; + } + }); + + this.parser.on('text', (text: string) => { + this.currentTextContent += text; + }); + + this.parser.on('closetag', (tag: SaxesTagPlain) => { + const text = this.currentTextContent.trim(); + + if (this.currentChannel) { + switch (tag.name) { + case 'display-name': + this.currentChannel.displayName!.push({ + lang: this.currentLang, + value: text, + }); + break; + case 'url': + if (text) this.currentChannel.url!.push(text); + break; + case 'channel': + // Channel complete, add to batch + this.channels.push(this.currentChannel as EpgChannel); + this.totalChannels++; + this.currentChannel = null; + + if (this.channels.length >= CHANNEL_BATCH_SIZE) { + this.flushChannels(); + } + break; + } + } + + if (this.currentProgram) { + switch (tag.name) { + case 'title': + this.currentProgram.title!.push({ + lang: this.currentLang, + value: text, + }); + break; + case 'desc': + this.currentProgram.desc!.push({ + lang: this.currentLang, + value: text, + }); + break; + case 'category': + this.currentProgram.category!.push({ + lang: this.currentLang, + value: text, + }); + break; + case 'date': + this.currentProgram.date = text; + break; + case 'value': + // This is for rating value + if ( + this.elementStack.includes('rating') && + this.currentProgram.rating!.length > 0 + ) { + this.currentProgram.rating![ + this.currentProgram.rating!.length - 1 + ].value = text; + } + break; + case 'episode-num': + if (this.currentProgram.episodeNum!.length > 0) { + this.currentProgram.episodeNum![ + this.currentProgram.episodeNum!.length - 1 + ].value = text; + } + break; + case 'programme': + // Program complete, add to batch + this.programs.push(this.currentProgram as EpgProgram); + this.totalPrograms++; + this.currentProgram = null; + + if (this.programs.length >= PROGRAM_BATCH_SIZE) { + this.flushPrograms(); + } + break; + } + } + + this.elementStack.pop(); + this.currentTextContent = ''; + }); + + this.parser.on('error', (err: Error) => { + console.error(loggerLabel, 'Parser error:', err.message); + }); + } + + private flushChannels(): void { + if (this.channels.length > 0) { + this.onChannelsBatch([...this.channels]); + this.channels = []; + this.onProgress(this.totalChannels, this.totalPrograms); } + } - let xmlString: string; + private flushPrograms(): void { + if (this.programs.length > 0) { + this.onProgramsBatch([...this.programs]); + this.programs = []; + this.onProgress(this.totalChannels, this.totalPrograms); + } + } + + write(chunk: string): void { + this.parser.write(chunk); + } + + finish(): { totalChannels: number; totalPrograms: number } { + this.parser.close(); + // Flush any remaining data + this.flushChannels(); + this.flushPrograms(); + return { + totalChannels: this.totalChannels, + totalPrograms: this.totalPrograms, + }; + } +} + +/** + * Fetches and parses EPG data from URL using streaming + */ +async function fetchAndParseEpgStreaming(url: string): Promise { + const isGzipped = url.endsWith('.gz'); + + console.log( + loggerLabel, + `Fetching EPG from ${url} (gzipped: ${isGzipped})` + ); + + const response = await fetch(url.trim()); + + if (!response.ok) { + throw new Error(`HTTP error! status: ${response.status}`); + } + + if (!response.body) { + throw new Error('Response body is null'); + } + + 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) { - // For gzipped files, get as ArrayBuffer - const arrayBuffer = await response.arrayBuffer(); - const buffer = Buffer.from(arrayBuffer); - const decompressed = gunzipSync(buffer); - xmlString = decompressed.toString('utf-8'); - } else { - // For regular files, get as text - xmlString = await response.text(); + const gunzip = createGunzip(); + dataStream = nodeStream.pipe(gunzip); + + gunzip.on('error', (err) => { + console.error(loggerLabel, 'Gunzip error:', err); + reject(err); + }); } - const parsed = parseEpg(xmlString); + dataStream.on('data', (chunk: Buffer) => { + try { + parser.write(chunk.toString('utf-8')); + } catch (err) { + console.error(loggerLabel, 'Parse error:', err); + reject(err); + } + }); - // 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[], - }; + dataStream.on('end', () => { + try { + const stats = parser.finish(); + console.log( + loggerLabel, + `Parsing complete: ${stats.totalChannels} channels, ${stats.totalPrograms} programs` + ); - return normalized; - } catch (error) { - console.error( - loggerLabel, - 'Error fetching EPG:', - error instanceof Error ? error.message : String(error) - ); - throw error; - } + const response: WorkerResponse = { + type: 'EPG_COMPLETE', + url, + stats, + }; + parentPort?.postMessage(response); + resolve(); + } catch (err) { + reject(err); + } + }); + + dataStream.on('error', (err) => { + console.error(loggerLabel, 'Stream error:', err); + reject(err); + }); + }); } /** @@ -105,15 +417,7 @@ if (parentPort) { message.type === 'FETCH_EPG' || message.type === 'FORCE_FETCH' ) { - const parsedData = await fetchAndParseEpg(message.url); - - const response: WorkerResponse = { - type: 'EPG_PARSED', - data: parsedData, - url: message.url, - }; - - parentPort?.postMessage(response); + await fetchAndParseEpgStreaming(message.url); } } catch (error) { console.error(loggerLabel, 'Worker error:', error); @@ -122,7 +426,6 @@ if (parentPort) { error: error instanceof Error ? error.message : String(error), url: message.url, }; - parentPort?.postMessage(errorResponse); } });