From 4ed22af9e55f04db3ae92a1331c2dd762db184b7 Mon Sep 17 00:00:00 2001 From: Technomancer702 Date: Mon, 12 Oct 2026 16:16:48 -0700 Subject: [PATCH] Preserve the guide until a complete EPG refresh can activate Adapted from MikaelKW/nodecast-tv-plus commits c267d98d734753a09b08d267c8ec2c3fa11c33f5, b761690bac7626a3cc5ddf754d96b14a02f506fe, and 5524fdaffed5c6a44ae69474e7eaa2f18d428c75. Preserve every parser batch, decode gzip once, and retain live channel metadata. Co-authored-by: Mikael --- server/routes/sources.js | 1 + server/services/epgParser.js | 595 +++++++++------------------------ server/services/syncService.js | 287 +++++++++++----- server/services/xmltvDate.js | 94 ++++++ test/epg-refresh.test.js | 241 +++++++++++++ 5 files changed, 701 insertions(+), 517 deletions(-) create mode 100644 server/services/xmltvDate.js create mode 100644 test/epg-refresh.test.js diff --git a/server/routes/sources.js b/server/routes/sources.js index 0b152f2..a0b95bc 100644 --- a/server/routes/sources.js +++ b/server/routes/sources.js @@ -120,6 +120,7 @@ router.delete('/:id', async (req, res) => { } // Cascade delete: Clean up SQLite data for this source + syncService.invalidateEpgSource(sourceId); const db = getDb(); const deleteCategories = db.prepare('DELETE FROM categories WHERE source_id = ?'); const deleteItems = db.prepare('DELETE FROM playlist_items WHERE source_id = ?'); diff --git a/server/services/epgParser.js b/server/services/epgParser.js index 9b75fff..7ce3f80 100644 --- a/server/services/epgParser.js +++ b/server/services/epgParser.js @@ -1,448 +1,179 @@ -/** - * EPG (XMLTV) Parser (Streaming) - * Parses XMLTV format EPG data and extracts channel/programme information using streaming XML parser - */ - +/** Streaming XMLTV parsing with bounded batches and complete-document validation. */ const sax = require('sax'); -const zlib = require('zlib'); -const { Readable } = require('stream'); +const zlib = require('node:zlib'); +const { Readable } = require('node:stream'); +const { pipeline } = require('node:stream/promises'); +const { StringDecoder } = require('node:string_decoder'); +const { parseXmltvDate } = require('./xmltvDate'); -/** - * Parse XMLTV date format (YYYYMMDDHHmmss +ZZZZ) - * @param {string} dateStr - XMLTV format date string - * @returns {Date} - */ -function parseXmltvDate(dateStr) { - if (!dateStr) return null; +async function* parseStreaming(input, batchSize = 1000) { + if (!Number.isInteger(batchSize) || batchSize < 1) throw new Error('Invalid EPG batch size'); + const stream = typeof input === 'string' ? Readable.from([input]) : input; + const parser = sax.parser(true, { trim: true, normalize: true }); + const decoder = new StringDecoder('utf8'); + let channels = [], programmes = [], skippedProgrammes = 0; + const ready = []; + let currentObject = null, objectTag = null, textBuffer = ''; + let validTimestamps = true, error = null, depth = 0; + let sawTvRoot = false, closedTvRoot = false; - // Format: 20231225120000 +0000 - const match = dateStr.match(/^(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})\s*([+-]\d{4})?$/); - if (!match) { - // Try ISO format fallback - return new Date(dateStr); - } - - const [, year, month, day, hour, minute, second, tz] = match; - let isoStr = `${year}-${month}-${day}T${hour}:${minute}:${second}`; - - if (tz) { - const tzHours = tz.substring(0, 3); - const tzMins = tz.substring(3); - isoStr += `${tzHours}:${tzMins}`; - } else { - isoStr += 'Z'; - } - - return new Date(isoStr); -} - -/** - * Parse XMLTV content (Stream or String) - * @param {Readable|string} input - XMLTV content as Stream or String - * @returns {Promise<{ channels: Array, programmes: Array }>} - */ -function parse(input) { - return new Promise((resolve, reject) => { - const channels = []; - const programmes = []; - - const saxStream = sax.createStream(true, { trim: true, normalize: true }); // strict mode - - let currentTag = null; - let currentObject = null; - let textBuffer = ''; - - saxStream.on('error', function (e) { - // clear the error - this._parser.error = null; - this._parser.resume(); - console.warn('XML Parse Warning:', e.message); - }); - - saxStream.on('opentag', function (node) { - currentTag = node.name; - const attr = node.attributes; - - if (currentTag === 'channel') { - currentObject = { - id: attr.id, - name: null, // Will be populated by display-name tag - icon: null, - url: null - }; - } else if (currentTag === 'programme') { - currentObject = { - channelId: attr.channel, - start: parseXmltvDate(attr.start), - stop: parseXmltvDate(attr.stop), - title: null, - subtitle: null, - description: null, - category: [], - icon: null, - date: null, - episodeNum: null - }; - } else if (currentTag === 'icon') { - if (currentObject) { - currentObject.icon = attr.src; - } - } - textBuffer = ''; - }); - - saxStream.on('text', function (text) { - textBuffer += text; - }); - - saxStream.on('cdata', function (text) { - textBuffer += text; - }); - - saxStream.on('closetag', function (tagName) { - if (tagName === 'channel') { - if (currentObject) channels.push(currentObject); - currentObject = null; - } else if (tagName === 'programme') { - if (currentObject) programmes.push(currentObject); - currentObject = null; - } else if (currentObject) { - // Handle properties within objects - switch (tagName) { - case 'display-name': // channel name - if (!currentObject.name) currentObject.name = textBuffer; - break; - case 'url': // channel url - currentObject.url = textBuffer; - break; - case 'title': - currentObject.title = textBuffer; - break; - case 'sub-title': - currentObject.subtitle = textBuffer; - break; - case 'desc': - currentObject.description = textBuffer; - break; - case 'category': - if (textBuffer && currentObject.category) currentObject.category.push(textBuffer); - break; - case 'date': - currentObject.date = textBuffer; - break; - case 'episode-num': - // Prefer system "xmltv_ns" or just take text - // Complex episode parsing logic can go here if needed - currentObject.episodeNum = textBuffer; - break; - } - } - }); - - saxStream.on('end', function () { - resolve({ channels, programmes }); - }); - - // Handle input type - if (typeof input === 'string') { - const inputStream = Readable.from([input]); - inputStream.pipe(saxStream); - } else { - input.pipe(saxStream); + const queueBatch = isLast => { + ready.push({ channels, programmes, skippedProgrammes, isLast }); + channels = []; + programmes = []; + skippedProgrammes = 0; + }; + parser.onerror = e => { + error ||= new Error(`Invalid XMLTV document: ${e.message}`); + parser.resume(); + }; + parser.onopentag = node => { + if (depth === 0) { + if (node.name !== 'tv' || sawTvRoot) error ||= new Error('Invalid XMLTV document: expected one root'); + sawTvRoot = node.name === 'tv'; } - }); + depth++; + const attr = node.attributes; + if (depth === 2 && node.name === 'channel') { + objectTag = 'channel'; + currentObject = { id: attr.id, name: null, icon: null, url: null }; + } else if (depth === 2 && node.name === 'programme') { + objectTag = 'programme'; + const start = parseXmltvDate(attr.start), stop = parseXmltvDate(attr.stop); + validTimestamps = Boolean(attr.channel && start) && (!attr.stop || Boolean(stop)); + currentObject = { channelId: attr.channel, start, stop, title: null, + subtitle: null, description: null, category: [], icon: null, date: null, episodeNum: null }; + } else if (node.name === 'icon' && currentObject) { + currentObject.icon = attr.src; + } + textBuffer = ''; + }; + parser.ontext = text => { textBuffer += text; }; + parser.oncdata = parser.ontext; + parser.onclosetag = tag => { + if (depth === 1 && tag === 'tv') closedTvRoot = true; + if (depth === 2 && tag === objectTag) { + if (objectTag === 'channel' && currentObject.id) { + currentObject.name ||= currentObject.id; + channels.push(currentObject); + } else if (objectTag === 'programme') { + if (validTimestamps) programmes.push(currentObject); + else skippedProgrammes++; + } + currentObject = null; + objectTag = null; + if (programmes.length >= batchSize || channels.length >= batchSize) queueBatch(false); + } else if (currentObject) { + switch (tag) { + case 'display-name': currentObject.name ||= textBuffer; break; + case 'url': currentObject.url = textBuffer; break; + case 'title': currentObject.title = textBuffer; break; + case 'sub-title': currentObject.subtitle = textBuffer; break; + case 'desc': currentObject.description = textBuffer; break; + case 'category': if (textBuffer && currentObject.category) currentObject.category.push(textBuffer); break; + case 'date': currentObject.date = textBuffer; break; + case 'episode-num': currentObject.episodeNum = textBuffer; break; + } + } + depth--; + }; + + try { + for await (const chunk of stream) { + const bytes = typeof chunk === 'string' ? Buffer.from(chunk) : chunk; + // A network chunk can contain many batches. Keep each one and yield + // between bounded SAX writes instead of overwriting a pending batch. + for (let offset = 0; offset < bytes.length; offset += 16384) { + parser.write(decoder.write(bytes.subarray(offset, offset + 16384))); + if (error) throw error; + while (ready.length) yield ready.shift(); + } + } + parser.write(decoder.end()).close(); + if (error) throw error; + if (!sawTvRoot || !closedTvRoot) throw new Error('Invalid XMLTV document: missing complete root'); + queueBatch(true); + while (ready.length) yield ready.shift(); + } finally { + stream.destroy?.(); + } +} + +async function parse(input) { + const channels = [], programmes = []; + let skippedProgrammes = 0; + for await (const batch of parseStreaming(input)) { + channels.push(...batch.channels); + programmes.push(...batch.programmes); + skippedProgrammes += batch.skippedProgrammes; + } + return { channels, programmes, skippedProgrammes }; +} + +async function* fetchAndParseStreaming(url, batchSize = 1000) { + const response = await fetch(url); + if (!response.ok) { + await response.body?.cancel(); + throw new Error(`Failed to fetch EPG: ${response.status} ${response.statusText}`); + } + const body = response.body && typeof response.body.pipe === 'function' + ? response.body : response.body ? Readable.fromWeb(response.body) : Readable.from([]); + let source, decoded, completion; + try { + // Fetch transparently decodes Content-Encoding. Inspect actual bytes + // so .gz downloads work without double-decompressing HTTP gzip. + const iterator = body[Symbol.asyncIterator](); + const prefix = []; + let prefixLength = 0; + while (prefixLength < 2) { + const next = await iterator.next(); + if (next.done) break; + const chunk = Buffer.from(next.value); + prefix.push(chunk); + prefixLength += chunk.length; + } + const first = Buffer.concat(prefix); + source = Readable.from((async function* () { + if (first.length) yield first; + for await (const chunk of { [Symbol.asyncIterator]: () => iterator }) yield chunk; + })(), { objectMode: false }); + decoded = source; + if (first[0] === 0x1f && first[1] === 0x8b) { + decoded = zlib.createGunzip(); + completion = pipeline(source, decoded); + completion.catch(() => {}); // The decoder's async iterator reports failures. + } + yield* parseStreaming(decoded, batchSize); + } finally { + decoded?.destroy(); + source?.destroy(); + body.destroy(); + if (completion) await completion.catch(() => {}); + } +} + +async function fetchAndParse(url) { + const channels = [], programmes = []; + let skippedProgrammes = 0; + for await (const batch of fetchAndParseStreaming(url)) { + channels.push(...batch.channels); + programmes.push(...batch.programmes); + skippedProgrammes += batch.skippedProgrammes; + } + return { channels, programmes, skippedProgrammes }; } -/** - * Get programmes for a specific channel - */ function getProgrammesForChannel(programmes, channelId) { return programmes.filter(p => p.channelId === channelId); } -/** - * Get current and upcoming programmes for a channel - */ function getCurrentAndUpcoming(programmes, channelId, count = 5) { const now = new Date(); const channelProgrammes = getProgrammesForChannel(programmes, channelId); - - // Sort by start time channelProgrammes.sort((a, b) => a.start - b.start); - - // Find current and upcoming - const current = channelProgrammes.find(p => p.start <= now && p.stop > now); - const upcoming = channelProgrammes - .filter(p => p.start > now) - .slice(0, count); - - return { current, upcoming }; + return { current: channelProgrammes.find(p => p.start <= now && p.stop > now), + upcoming: channelProgrammes.filter(p => p.start > now).slice(0, count) }; } -/** - * Fetch and parse XMLTV from URL - */ -async function fetchAndParse(url) { - const response = await fetch(url); - if (!response.ok) { - throw new Error(`Failed to fetch EPG: ${response.status} ${response.statusText}`); - } - - let stream; - if (response.body && typeof response.body.pipe === 'function') { - stream = response.body; - } else if (response.body) { - stream = Readable.fromWeb(response.body); - } else { - stream = Readable.from([]); - } - - // Check for GZIP - // Note: We can't easily check for magic bytes on a stream without buffering. - // We'll rely on response headers or file extension mostly, or try to peek. - // For now, let's assume if content-encoding is gzip OR url ends in .gz - - // However, undici/fetch usually handles 'Content-Encoding: gzip' automatically transparently. - // We only need to manually gunzip if the server serves it as application/octet-stream but it's actually gzipped, - // or if it's a .gz file download. - - // A robust way for streams is checking magic bytes, but that requires peeking. - // Simplified approach: try to pipe through gunzip if the URL indicates it. - - const isGzipped = url.endsWith('.gz') || (response.headers.get('content-type') || '').includes('gzip'); - - if (isGzipped) { - const gunzip = zlib.createGunzip(); - stream.pipe(gunzip); - return parse(gunzip); - } - - // In the previous version we read magic bytes. - // To support that with streams we'd need a peek stream. - // For now let's trust the transparent decompression of fetch or the URL. - - return parse(stream); -} - -/** - * Streaming EPG parser that yields batches of programmes (memory-efficient) - * Channels are collected and returned with the first batch, then programmes are yielded in batches. - * - * @param {string} url - XMLTV URL - * @param {number} batchSize - Number of programmes per batch (default: 1000) - * @yields {{ channels: Array|null, programmes: Array, isLast: boolean }} - */ -async function* fetchAndParseStreaming(url, batchSize = 1000) { - const response = await fetch(url); - if (!response.ok) { - throw new Error(`Failed to fetch EPG: ${response.status} ${response.statusText}`); - } - - let stream; - if (response.body && typeof response.body.pipe === 'function') { - stream = response.body; - } else if (response.body) { - stream = Readable.fromWeb(response.body); - } else { - stream = Readable.from([]); - } - - const isGzipped = url.endsWith('.gz') || (response.headers.get('content-type') || '').includes('gzip'); - - if (isGzipped) { - const gunzip = zlib.createGunzip(); - stream.pipe(gunzip); - stream = gunzip; - } - - // Use async iterator pattern with SAX - yield* parseStreaming(stream, batchSize); -} - -/** - * Parse XMLTV as streaming async generator - * @param {Readable} input - XMLTV stream - * @param {number} batchSize - Number of programmes per batch - * @yields {{ channels: Array|null, programmes: Array, isLast: boolean }} - */ -async function* parseStreaming(input, batchSize = 1000) { - const channels = []; - let programmeBatch = []; - let channelsYielded = false; - - // We need to convert SAX events to an async iterator - // This requires collecting events and yielding when batch is full - - const saxStream = sax.createStream(true, { trim: true, normalize: true }); - - let currentTag = null; - let currentObject = null; - let textBuffer = ''; - let resolveNext = null; - let pendingBatch = null; - let ended = false; - let error = null; - - saxStream.on('error', function (e) { - this._parser.error = null; - this._parser.resume(); - console.warn('XML Parse Warning:', e.message); - }); - - saxStream.on('opentag', function (node) { - currentTag = node.name; - const attr = node.attributes; - - if (currentTag === 'channel') { - currentObject = { - id: attr.id, - name: null, - icon: null, - url: null - }; - } else if (currentTag === 'programme') { - currentObject = { - channelId: attr.channel, - start: parseXmltvDate(attr.start), - stop: parseXmltvDate(attr.stop), - title: null, - subtitle: null, - description: null, - category: [], - icon: null, - date: null, - episodeNum: null - }; - } else if (currentTag === 'icon') { - if (currentObject) { - currentObject.icon = attr.src; - } - } - textBuffer = ''; - }); - - saxStream.on('text', function (text) { - textBuffer += text; - }); - - saxStream.on('cdata', function (text) { - textBuffer += text; - }); - - saxStream.on('closetag', function (tagName) { - if (tagName === 'channel') { - if (currentObject) channels.push(currentObject); - currentObject = null; - } else if (tagName === 'programme') { - if (currentObject) { - programmeBatch.push(currentObject); - - // Check if we should yield a batch - if (programmeBatch.length >= batchSize) { - const batch = { - channels: !channelsYielded ? channels : null, - programmes: programmeBatch, - isLast: false - }; - channelsYielded = true; - programmeBatch = []; - - if (resolveNext) { - resolveNext(batch); - resolveNext = null; - } else { - pendingBatch = batch; - } - } - } - currentObject = null; - } else if (currentObject) { - switch (tagName) { - case 'display-name': - if (!currentObject.name) currentObject.name = textBuffer; - break; - case 'url': - currentObject.url = textBuffer; - break; - case 'title': - currentObject.title = textBuffer; - break; - case 'sub-title': - currentObject.subtitle = textBuffer; - break; - case 'desc': - currentObject.description = textBuffer; - break; - case 'category': - if (textBuffer && currentObject.category) currentObject.category.push(textBuffer); - break; - case 'date': - currentObject.date = textBuffer; - break; - case 'episode-num': - currentObject.episodeNum = textBuffer; - break; - } - } - }); - - saxStream.on('end', function () { - ended = true; - // Yield final batch - const batch = { - channels: !channelsYielded ? channels : null, - programmes: programmeBatch, - isLast: true - }; - if (resolveNext) { - resolveNext(batch); - resolveNext = null; - } else { - pendingBatch = batch; - } - }); - - saxStream.on('error', function (e) { - error = e; - if (resolveNext) { - resolveNext(null); - } - }); - - // Start piping - input.pipe(saxStream); - - // Yield batches as they become available - while (!ended || pendingBatch) { - if (pendingBatch) { - const batch = pendingBatch; - pendingBatch = null; - yield batch; - if (batch.isLast) break; - } else if (!ended) { - // Wait for next batch - const batch = await new Promise(resolve => { - resolveNext = resolve; - }); - if (batch) { - yield batch; - if (batch.isLast) break; - } - } - } - - if (error) { - throw error; - } -} - -module.exports = { - parse, - parseXmltvDate, - fetchAndParse, - fetchAndParseStreaming, - parseStreaming, - getProgrammesForChannel, - getCurrentAndUpcoming -}; - +module.exports = { parse, parseXmltvDate, fetchAndParse, fetchAndParseStreaming, + parseStreaming, getProgrammesForChannel, getCurrentAndUpcoming }; diff --git a/server/services/syncService.js b/server/services/syncService.js index 8ac35f5..a718ad9 100644 --- a/server/services/syncService.js +++ b/server/services/syncService.js @@ -3,9 +3,12 @@ const { sources, settings } = require('../db'); // For source config and setting const xtreamApi = require('./xtreamApi'); const m3uParser = require('./m3uParser'); const epgParser = require('./epgParser'); +const { randomUUID } = require('node:crypto'); // Sync tracking const activeSyncs = new Set(); // sourceId +const activeEpgSyncs = new Set(); +const epgSourceGenerations = new Map(); class SyncService { constructor() { @@ -14,6 +17,11 @@ class SyncService { this._currentInterval = null; } + invalidateEpgSource(sourceId) { + const key = String(sourceId); + epgSourceGenerations.set(key, (epgSourceGenerations.get(key) || 0) + 1); + } + /** * Get when the last global sync completed */ @@ -389,107 +397,216 @@ class SyncService { * Processes EPG files in batches to avoid OOM on large EPG data */ async syncEpgFromUrl(sourceId, url) { - console.log(`[Sync] Fetching EPG from: ${url.substring(0, 60)}...`); + console.log(`[Sync] Fetching EPG from: ${new URL(url).origin}`); - // Temporary memory logging for verification - const logMemory = () => { - const used = process.memoryUsage(); - console.log(`[Sync] Memory: ${Math.round(used.heapUsed / 1024 / 1024)}MB heap`); - }; - - logMemory(); - - const db = getDb(); - let allChannels = []; - let totalProgrammes = 0; - let batchCount = 0; - - // Clear old programmes first - db.prepare('DELETE FROM epg_programs WHERE source_id = ?').run(sourceId); - - const programmeStmt = db.prepare(` - INSERT INTO epg_programs (channel_id, source_id, start_time, end_time, title, description, data) - VALUES (?, ?, ?, ?, ?, ?, ?) - `); - - const insertProgrammes = db.transaction((progs) => { - for (const p of progs) { - programmeStmt.run( - p.channelId, - sourceId, - p.start ? p.start.getTime() : 0, - p.stop ? p.stop.getTime() : 0, - p.title, - p.description || p.desc, - JSON.stringify(p) - ); - } - }); - - // Stream and process in batches (default 1000 programmes per batch) - for await (const batch of epgParser.fetchAndParseStreaming(url)) { - batchCount++; - - // Collect channels from first batch - if (batch.channels) { - allChannels = batch.channels; - } - - // Save this batch of programmes immediately - if (batch.programmes.length > 0) { - insertProgrammes(batch.programmes); - totalProgrammes += batch.programmes.length; - } - - // Log progress every 10 batches - if (batchCount % 10 === 0) { - console.log(`[Sync] Processed ${totalProgrammes} programmes so far...`); - logMemory(); - } - - // Yield to event loop - await new Promise(resolve => setImmediate(resolve)); + const sourceKey = String(sourceId); + if (activeEpgSyncs.has(sourceKey)) { + throw new Error(`EPG refresh already in progress for source ${sourceId}`); } + activeEpgSyncs.add(sourceKey); + const sourceGeneration = epgSourceGenerations.get(sourceKey) || 0; + let cleanupRun = null; + let refreshError = null; - console.log(`[Sync] EPG Parsed: ${allChannels.length} channels, ${totalProgrammes} programmes`); - logMemory(); + try { - // Save EPG Channels - if (allChannels.length > 0) { + // Temporary memory logging for verification + const logMemory = () => { + const used = process.memoryUsage(); + console.log(`[Sync] Memory: ${Math.round(used.heapUsed / 1024 / 1024)}MB heap`); + }; + + logMemory(); + + const db = getDb(); + const runId = randomUUID(); + let totalChannels = 0; + let totalProgrammes = 0; + let skippedProgrammes = 0; + let batchCount = 0; + let sawFinalBatch = false; + + db.exec(` + CREATE TEMP TABLE IF NOT EXISTS epg_programs_staging ( + run_id TEXT NOT NULL, + source_id INTEGER NOT NULL, + channel_id TEXT NOT NULL, + start_time INTEGER NOT NULL, + end_time INTEGER NOT NULL, + title TEXT, + description TEXT, + data JSON + ); + CREATE INDEX IF NOT EXISTS temp.idx_epg_programs_staging_run + ON epg_programs_staging(run_id, source_id); + + CREATE TEMP TABLE IF NOT EXISTS epg_channels_staging ( + run_id TEXT NOT NULL, + source_id INTEGER NOT NULL, + id TEXT NOT NULL, + item_id TEXT NOT NULL, + name TEXT, + stream_icon TEXT, + data JSON, + PRIMARY KEY (run_id, id) + ); + `); + + const deleteStagedProgrammes = db.prepare( + 'DELETE FROM temp.epg_programs_staging WHERE run_id = ?' + ); + const deleteStagedChannels = db.prepare( + 'DELETE FROM temp.epg_channels_staging WHERE run_id = ?' + ); + cleanupRun = () => { + deleteStagedProgrammes.run(runId); + deleteStagedChannels.run(runId); + }; + + // A prior interrupted attempt in this process must never be reused. + db.prepare('DELETE FROM temp.epg_programs_staging WHERE source_id = ?').run(sourceId); + db.prepare('DELETE FROM temp.epg_channels_staging WHERE source_id = ?').run(sourceId); + + const programmeStmt = db.prepare(` + INSERT INTO temp.epg_programs_staging ( + run_id, source_id, channel_id, start_time, end_time, title, description, data + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `); const channelStmt = db.prepare(` - INSERT INTO playlist_items ( - id, source_id, item_id, type, name, stream_icon, - stream_url, category_id, data - ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(id) DO UPDATE SET + INSERT INTO temp.epg_channels_staging ( + run_id, source_id, id, item_id, name, stream_icon, data + ) VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(run_id, id) DO UPDATE SET + item_id = excluded.item_id, name = excluded.name, stream_icon = excluded.stream_icon, data = excluded.data `); - const insertChannels = db.transaction((chanList) => { - for (const ch of chanList) { - const id = `${sourceId}:${ch.id}`; - channelStmt.run( - id, + const stageProgrammes = db.transaction((programmes) => { + for (const programme of programmes) { + programmeStmt.run( + runId, sourceId, - ch.id, - 'epg_channel', - ch.name, - ch.icon || null, - null, - null, - JSON.stringify(ch) + programme.channelId, + programme.start ? programme.start.getTime() : 0, + programme.stop ? programme.stop.getTime() : 0, + programme.title, + programme.description || programme.desc, + JSON.stringify(programme) + ); + } + }); + const stageChannels = db.transaction((channels) => { + for (const channel of channels) { + // EPG channel IDs often overlap Xtream stream IDs. Keep the + // staged row in the same dedicated namespace as the active row. + const id = `${sourceId}:epg_channel:${channel.id}`; + channelStmt.run( + runId, + sourceId, + id, + channel.id, + channel.name, + channel.icon || null, + JSON.stringify(channel) ); } }); - insertChannels(allChannels); - console.log(`[Sync] Saved ${allChannels.length} EPG channels`); - } + const activateStagedGuide = db.transaction(() => { + db.prepare('DELETE FROM epg_programs WHERE source_id = ?').run(sourceId); + db.prepare(` + INSERT INTO epg_programs ( + channel_id, source_id, start_time, end_time, title, description, data + ) + SELECT channel_id, source_id, start_time, end_time, title, description, data + FROM temp.epg_programs_staging + WHERE run_id = ? AND source_id = ? + `).run(runId, sourceId); - console.log(`[Sync] Saved ${totalProgrammes} programmes`); + db.prepare(` + DELETE FROM playlist_items + WHERE source_id = ? AND type = 'epg_channel' + `).run(sourceId); + db.prepare(` + INSERT INTO playlist_items ( + id, source_id, item_id, type, name, stream_icon, + stream_url, category_id, data + ) + SELECT id, source_id, item_id, 'epg_channel', name, stream_icon, + NULL, NULL, data + FROM temp.epg_channels_staging + WHERE run_id = ? AND source_id = ? + `).run(runId, sourceId); + + cleanupRun(); + }); + + const sourceAtStart = await sources.getById(sourceId); + + // Stream into isolated staging tables. The active guide remains + // readable throughout download and parsing. + for await (const batch of epgParser.fetchAndParseStreaming(url)) { + batchCount++; + skippedProgrammes += batch.skippedProgrammes || 0; + + if (batch.channels !== null && batch.channels !== undefined) { + stageChannels(batch.channels); + totalChannels += batch.channels.length; + } + + if (batch.programmes.length > 0) { + stageProgrammes(batch.programmes); + totalProgrammes += batch.programmes.length; + } + + if (batch.isLast) sawFinalBatch = true; + + if (batchCount % 10 === 0) { + console.log(`[Sync] Staged ${totalProgrammes} programmes so far...`); + logMemory(); + } + + // Keep requests responsive between SQLite batches. + await new Promise(resolve => setImmediate(resolve)); + } + + if (!sawFinalBatch) { + throw new Error('EPG refresh ended before the XMLTV document completed'); + } + + // If a real configured source disappeared while its download was + // running, do not recreate any of its guide data. + const sourceStillExists = !sourceAtStart || await sources.getById(sourceId); + const sourceWasInvalidated = (epgSourceGenerations.get(sourceKey) || 0) !== sourceGeneration; + if (sourceWasInvalidated || !sourceStillExists) { + throw new Error(`EPG source ${sourceId} was removed during refresh`); + } + + console.log(`[Sync] EPG Parsed: ${totalChannels} channels, ${totalProgrammes} programmes`); + if (skippedProgrammes > 0) { + console.warn(`[Sync] Skipped ${skippedProgrammes} programme entries with invalid XMLTV timestamps`); + } + logMemory(); + + // One short transaction makes programmes and channel mapping + // visible together. A valid empty feed intentionally clears both. + activateStagedGuide(); + console.log(`[Sync] Activated ${totalChannels} EPG channels and ${totalProgrammes} programmes`); + } catch (error) { + refreshError = error; + throw error; + } finally { + try { + if (cleanupRun) cleanupRun(); + } catch (cleanupError) { + console.warn('[Sync] Failed to clean EPG staging data:', cleanupError.message); + if (!refreshError) throw cleanupError; + } finally { + activeEpgSyncs.delete(sourceKey); + } + } } /** diff --git a/server/services/xmltvDate.js b/server/services/xmltvDate.js new file mode 100644 index 0000000..7b7215f --- /dev/null +++ b/server/services/xmltvDate.js @@ -0,0 +1,94 @@ +function createValidatedCalendarDate(year, month, day, hour, minute, second) { + if ( + month < 1 || month > 12 || + day < 1 || day > 31 || + hour < 0 || hour > 23 || + minute < 0 || minute > 59 || + second < 0 || second > 59 + ) { + return null; + } + + // setUTCFullYear avoids Date.UTC's special handling of years 0-99. + const calendarDate = new Date(0); + calendarDate.setUTCFullYear(year, month - 1, day); + calendarDate.setUTCHours(hour, minute, second, 0); + + // Date normalizes impossible calendar values (for example February 30), + // so compare every component before accepting it. + if ( + calendarDate.getUTCFullYear() !== year || + calendarDate.getUTCMonth() !== month - 1 || + calendarDate.getUTCDate() !== day || + calendarDate.getUTCHours() !== hour || + calendarDate.getUTCMinutes() !== minute || + calendarDate.getUTCSeconds() !== second + ) { + return null; + } + + return calendarDate; +} + +function parseTimezoneOffset(timezone) { + if (!timezone || timezone.toUpperCase() === 'Z') return 0; + + const normalized = timezone.replace(':', ''); + const offsetHours = Number(normalized.slice(1, 3)); + const offsetRemainder = Number(normalized.slice(3, 5)); + if (offsetHours > 23 || offsetRemainder > 59) return null; + + const total = (offsetHours * 60) + offsetRemainder; + return normalized[0] === '-' ? -total : total; +} + +/** + * Parse an XMLTV date/time. XMLTV permits YYYYMMDDHHmmss or a shorter + * initial precision (for example YYYYMMDDHHmm, YYYYMMDDHH, or YYYYMMDD). + * Missing components use the earliest value in that precision and UTC is + * assumed when no numeric offset is provided. + * @param {string} dateStr - XMLTV format date string + * @returns {Date|null} + */ +function parseXmltvDate(dateStr) { + if (!dateStr) return null; + + const value = String(dateStr).trim(); + const match = value.match(/^(\d{4}(?:\d{2}){0,5})\s*(Z|[+-]\d{4})?$/i); + if (!match) { + // Preserve the existing compatibility fallback for explicit ISO 8601 + // timestamps without accepting arbitrary host-dependent date strings. + const isoMatch = value.match(/^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2})(?::(\d{2})(?:\.\d{1,3})?)?(Z|[+-]\d{2}:?\d{2})$/i); + if (!isoMatch) return null; + + const [, year, month, day, hour, minute, second = '0', timezone] = isoMatch; + if (!createValidatedCalendarDate( + Number(year), + Number(month), + Number(day), + Number(hour), + Number(minute), + Number(second) + ) || parseTimezoneOffset(timezone) === null) return null; + + const isoDate = new Date(value); + return Number.isNaN(isoDate.getTime()) ? null : isoDate; + } + + const digits = match[1]; + const timezone = match[2]?.toUpperCase(); + const year = Number(digits.slice(0, 4)); + const month = digits.length >= 6 ? Number(digits.slice(4, 6)) : 1; + const day = digits.length >= 8 ? Number(digits.slice(6, 8)) : 1; + const hour = digits.length >= 10 ? Number(digits.slice(8, 10)) : 0; + const minute = digits.length >= 12 ? Number(digits.slice(10, 12)) : 0; + const second = digits.length >= 14 ? Number(digits.slice(12, 14)) : 0; + + const calendarDate = createValidatedCalendarDate(year, month, day, hour, minute, second); + const offsetMinutes = parseTimezoneOffset(timezone); + if (!calendarDate || offsetMinutes === null) return null; + + return new Date(calendarDate.getTime() - (offsetMinutes * 60 * 1000)); +} + +module.exports = { parseXmltvDate }; diff --git a/test/epg-refresh.test.js b/test/epg-refresh.test.js new file mode 100644 index 0000000..d2f675a --- /dev/null +++ b/test/epg-refresh.test.js @@ -0,0 +1,241 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); +const vm = require('node:vm'); +const http = require('node:http'); +const zlib = require('node:zlib'); +const { Readable } = require('node:stream'); +const { createRequire } = require('node:module'); +const parser = require('../server/services/epgParser'); + +const programme = (title, channel = '1') => `${title}`; +const channel = (id = '1', name = 'Guide channel') => `${name}`; +const guide = (title = 'Old', id = '1') => `${channel(id)}${programme(title, id)}`; + +async function collect(input, batchSize = 7) { + const result = { channels: [], programmes: [], skipped: 0, final: 0 }; + for await (const batch of parser.parseStreaming(input, batchSize)) { + assert.ok(batch.programmes.length <= batchSize); + assert.ok(batch.channels.length <= batchSize); + result.channels.push(...batch.channels); + result.programmes.push(...batch.programmes); + result.skipped += batch.skippedProgrammes; + result.final += Number(batch.isLast); + } + return result; +} + +test('XMLTV accepts reduced precision and rejects impossible timestamps', async () => { + for (const [value, expected] of [ + ['20260714061234 +0000', '2026-07-14T06:12:34.000Z'], + ['202607140612 +0000', '2026-07-14T06:12:00.000Z'], + ['2026071406 -0500', '2026-07-14T11:00:00.000Z'], + ['20260714 +0230', '2026-07-13T21:30:00.000Z'], + ['202607', '2026-07-01T00:00:00.000Z'], ['2026', '2026-01-01T00:00:00.000Z'], + ['20260714061234Z', '2026-07-14T06:12:34.000Z'], + ['2026-07-14T06:12:34+02:00', '2026-07-14T04:12:34.000Z'] + ]) assert.equal(parser.parseXmltvDate(value).toISOString(), expected, value); + for (const value of ['', 'garbage', '20260230060000 +0000', '20261301060000', + '20260714240000', '20260714066000', '20260714060060', '20260714060000 +2460', + '20260714060 +0000', '2026-02-30T06:00:00Z']) assert.equal(parser.parseXmltvDate(value), null, value); + const xml = `${channel()}${programme('Valid')}${programme('Invalid').replace('202607140600', '202602300600')}`; + assert.equal((await parser.parse(xml)).programmes.length, 1); + const streamed = await collect(xml); + assert.equal(streamed.programmes.length, 1); + assert.equal(streamed.skipped, 1); +}); + +test('streaming XMLTV retains every batch in one chunk and channels arriving after programmes', async () => { + const titles = Array.from({ length: 2501 }, (_, i) => `Programme ${i}`); + const xml = `${channel()}${titles.map(title => programme(title)).join('')}${channel('10', 'Late')}`; + const result = await collect(Readable.from([Buffer.from(xml)])); + assert.deepEqual(result.programmes.map(p => p.title), titles); + assert.deepEqual(result.channels.map(c => c.id), ['1', '10']); + assert.equal(result.final, 1); + const unicode = Buffer.from(guide('Café 東京')); + const split = await collect(Readable.from(Array.from(unicode, byte => Buffer.from([byte])))); + assert.equal(split.programmes[0].title, 'Café 東京'); +}); + +test('both XMLTV parsers reject malformed or incomplete documents and propagate stream errors', async () => { + for (const xml of ['', 'Failure', '', `${programme('Partial')}`, + '', '']) { + await assert.rejects(parser.parse(xml), /XMLTV/); + await assert.rejects(collect(xml), /XMLTV/); + } + const broken = () => Readable.from((async function* () { yield ''; throw new Error('Network interrupted'); })()); + await assert.rejects(parser.parse(broken()), /Network interrupted/); + await assert.rejects(collect(broken()), /Network interrupted/); +}); + +function fixture(t, epg = parser) { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'nodecast-epg-test-')); + const filename = path.join(__dirname, '..', 'server/db/sqlite.js'); + const sqliteModule = { exports: {} }; + vm.runInNewContext(fs.readFileSync(filename, 'utf8'), { module: sqliteModule, + require: createRequire(filename), __dirname: path.dirname(filename), + process: { env: { NODECAST_DATA_DIR: dir } }, console: { log() {} } }, { filename }); + const db = sqliteModule.exports.getDb(); + t.after(() => { + db.close(); + assert.equal(path.dirname(path.resolve(dir)), path.resolve(os.tmpdir())); + fs.rmSync(dir, { recursive: true, force: true }); + }); + const configured = new Map([[1, { id: 1 }], [2, { id: 2 }]]); + const sources = { getById: async id => configured.get(Number(id)), delete: async id => configured.delete(Number(id)) }; + const syncFile = path.join(__dirname, '..', 'server/services/syncService.js'); + const syncModule = { exports: {} }, realRequire = createRequire(syncFile); + vm.runInNewContext(fs.readFileSync(syncFile, 'utf8'), { module: syncModule, URL, process, setImmediate, + console: { log() {}, warn() {} }, require(id) { + if (id === '../db/sqlite') return { getDb: () => db }; + if (id === '../db') return { sources, settings: {} }; + if (id === './epgParser') return epg; + if (id === './xtreamApi' || id === './m3uParser') return {}; + return realRequire(id); + } }, { filename: syncFile }); + const sync = syncModule.exports; + const snapshot = sourceId => ({ + programmes: db.prepare('SELECT channel_id,title,start_time,end_time FROM epg_programs WHERE source_id=? ORDER BY title').all(sourceId), + channels: db.prepare("SELECT id,item_id,name FROM playlist_items WHERE source_id=? AND type='epg_channel' ORDER BY id").all(sourceId) + }); + const stagedCount = () => db.prepare('SELECT COUNT(*) AS count FROM temp.epg_programs_staging').get().count + + db.prepare('SELECT COUNT(*) AS count FROM temp.epg_channels_staging').get().count; + return { db, sync, sources, configured, snapshot, stagedCount }; +} + +async function origin(t, handler) { + const server = http.createServer(handler); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + t.after(() => new Promise(resolve => { server.close(resolve); server.closeAllConnections(); })); + return `http://127.0.0.1:${server.address().port}`; +} + +test('EPG replacement preserves live media and other sources while clearing stale guide entries', async t => { + const { db, sync, snapshot, stagedCount } = fixture(t); + let xml = guide(); + const base = await origin(t, (req, res) => res.end(xml)); + const insert = db.prepare('INSERT INTO playlist_items (id,source_id,item_id,type,name,stream_url,data) VALUES (?,?,?,?,?,?,?)'); + for (const id of ['1', '10']) insert.run(`1:${id}`, 1, id, 'live', `Live ${id}`, `http://provider.test/${id}`, '{"retained":true}'); + insert.run('1:legacy', 1, 'legacy', 'epg_channel', 'Legacy guide', null, '{}'); + const liveBefore = db.prepare("SELECT * FROM playlist_items WHERE type='live' ORDER BY id").all(); + await sync.syncEpgFromUrl(2, base); + const otherBefore = snapshot(2); + xml = `${channel('1')}${channel('10')}${programme('Fresh', '10')}`; + await sync.syncEpgFromUrl(1, base); + assert.deepEqual(db.prepare("SELECT * FROM playlist_items WHERE type='live' ORDER BY id").all(), liveBefore); + assert.deepEqual(snapshot(1).channels.map(c => c.id), ['1:epg_channel:1', '1:epg_channel:10']); + assert.equal(snapshot(1).programmes[0].title, 'Fresh'); + assert.deepEqual(snapshot(2), otherBefore); + xml = ''; + await sync.syncEpgFromUrl(1, base); + assert.deepEqual(snapshot(1), { programmes: [], channels: [] }); + assert.deepEqual(db.prepare("SELECT * FROM playlist_items WHERE type='live' ORDER BY id").all(), liveBefore); + assert.equal(stagedCount(), 0); +}); + +test('failed XML, gzip, HTTP and interrupted downloads preserve the complete previous guide', async t => { + const { sync, snapshot, stagedCount } = fixture(t); + const prefix = `${channel()}${Array.from({ length: 1001 }, (_, i) => programme(`New ${i}`)).join('')}`; + const base = await origin(t, (req, res) => { + if (req.url === '/good') res.end(guide()); + else if (req.url === '/truncated') res.end(prefix); + else if (req.url === '/bad.gz') res.end(Buffer.from([0x1f, 0x8b, 0, 0, 0])); + else if (req.url === '/http') { res.writeHead(503); res.end('Unavailable'); } + else { res.writeHead(200, { 'Content-Length': Buffer.byteLength(prefix) + 100 }); res.write(prefix); setImmediate(() => res.destroy()); } + }); + await sync.syncEpgFromUrl(1, `${base}/good`); + const old = snapshot(1); + for (const suffix of ['truncated', 'bad.gz', 'http', 'interrupted']) { + await assert.rejects(sync.syncEpgFromUrl(1, `${base}/${suffix}`)); + assert.deepEqual(snapshot(1), old, suffix); + assert.equal(stagedCount(), 0, suffix); + } + await sync.syncEpgFromUrl(1, `${base}/good`); + assert.deepEqual(snapshot(1), old); +}); + +test('gzip downloads and transparent HTTP gzip decode once, including split magic bytes', async t => { + const { sync, snapshot } = fixture(t); + const compressed = zlib.gzipSync(guide('Compressed')); + const base = await origin(t, (req, res) => { + if (req.url === '/encoded.gz') res.setHeader('Content-Encoding', 'gzip'); + res.write(compressed.subarray(0, 1)); + setImmediate(() => res.end(compressed.subarray(1))); + }); + for (const suffix of ['download.gz?token=test', 'encoded.gz']) { + await sync.syncEpgFromUrl(1, `${base}/${suffix}`); + assert.equal(snapshot(1).programmes[0].title, 'Compressed'); + } + assert.equal((await parser.fetchAndParse(`${base}/encoded.gz`)).programmes[0].title, 'Compressed'); +}); + +test('different sources can refresh concurrently without mixing staged data', async t => { + const { sync, snapshot, stagedCount } = fixture(t); + const base = await origin(t, (req, res) => res.end(`${channel()}${Array.from({ length: 1101 }, (_, i) => programme(`${req.url}-${i}`)).join('')}`)); + await Promise.all([sync.syncEpgFromUrl(1, `${base}/one`), sync.syncEpgFromUrl(2, `${base}/two`)]); + for (const [sourceId, prefix] of [[1, '/one-'], [2, '/two-']]) { + const data = snapshot(sourceId); + assert.equal(data.programmes.length, 1101); + assert.ok(data.programmes.every(p => p.title.startsWith(prefix))); + assert.equal(data.channels.length, 1); + } + assert.equal(stagedCount(), 0); +}); + +test('activation is atomic, rolls back SQL failures, and serializes same-source refreshes', async t => { + let streamFactory = async function* () { yield* parser.parseStreaming(guide()); }; + const epg = { fetchAndParseStreaming: (...args) => streamFactory(...args) }; + const { db, sync, snapshot, stagedCount } = fixture(t, epg); + const url = 'http://provider.test/guide'; + await sync.syncEpgFromUrl(1, url); + const old = snapshot(1); + streamFactory = async function* () { + for await (const batch of parser.parseStreaming(guide('New', '10'), 1)) { + yield batch; + assert.deepEqual(snapshot(1), old); + await assert.rejects(sync.syncEpgFromUrl(1, url), /already in progress/); + } + }; + db.exec("CREATE TRIGGER fail_guide BEFORE INSERT ON playlist_items WHEN NEW.type='epg_channel' BEGIN SELECT RAISE(ABORT,'activation failed'); END;"); + await assert.rejects(sync.syncEpgFromUrl(1, url), /activation failed/); + assert.deepEqual(snapshot(1), old); + assert.equal(stagedCount(), 0); + db.exec('DROP TRIGGER fail_guide'); + await sync.syncEpgFromUrl(1, url); + assert.equal(snapshot(1).programmes[0].title, 'New'); +}); + +test('deleting a source through its route invalidates an in-flight guide refresh', async t => { + let release, staged; + const paused = new Promise(resolve => { staged = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const epg = { async *fetchAndParseStreaming() { + for await (const batch of parser.parseStreaming(guide('Deleted'), 1)) { + yield batch; + if (!batch.isLast) { staged(); await gate; } + } + } }; + const { db, sync, sources, snapshot, stagedCount } = fixture(t, epg); + const filename = path.join(__dirname, '..', 'server/routes/sources.js'); + let deleteRoute; + const router = { get() {}, post() {}, put() {}, delete(route, handler) { deleteRoute = handler; } }; + vm.runInNewContext(fs.readFileSync(filename, 'utf8'), { module: { exports: {} }, console: { log() {}, error() {} }, require(id) { + if (id === 'express') return { Router: () => router }; + if (id === '../db') return { sources }; + if (id === '../db/sqlite') return { getDb: () => db }; + if (id === '../services/syncService') return sync; + return {}; + } }, { filename }); + const refreshing = sync.syncEpgFromUrl(1, 'http://provider.test/guide'); + const rejected = assert.rejects(refreshing, /removed during refresh/); + await paused; + const res = { status(code) { this.code = code; return this; }, json(value) { this.value = value; } }; + await deleteRoute({ params: { id: '1' } }, res); + assert.equal(res.value.success, true); + release(); + await rejected; + assert.deepEqual(snapshot(1), { programmes: [], channels: [] }); + assert.equal(stagedCount(), 0); +});