diff --git a/.changes/epg-double-gzip.md b/.changes/epg-double-gzip.md new file mode 100644 index 000000000..51f35e458 --- /dev/null +++ b/.changes/epg-double-gzip.md @@ -0,0 +1,7 @@ +--- +type: fix +area: epg +issues: [1586] +--- + +EPG guides now load when a server adds gzip compression to an already compressed XMLTV file. Sources using a single gzip layer continue to work. diff --git a/AGENTS.md b/AGENTS.md index 38c8857f7..82d8e4ef7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -210,6 +210,14 @@ Button states use the matching group; "Total selected" counts the whole catalog. Save persists the complete draft, Close discards it, and refresh restores hidden categories by provider ID and type. See `docs/architecture/category-management.md`. +## XMLTV Response Compression + +Electron decodes HTTP compression before the gzip file layer. For `.gz`/gzip +metadata plus HTTP gzip, a streaming signature check unwraps one remaining +file layer while preserving single-layer providers. Errors and cancellation +close the decoding chain. Contract: `docs/architecture/m3u-playlist-module.md` +("XMLTV response compression"). + ## XMLTV Source Removal Saving Settings → EPG reconciles cached XMLTV with committed global URLs and diff --git a/CLAUDE.md b/CLAUDE.md index 5b745a627..5de0694cc 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1553,7 +1553,7 @@ stream_id`); it drops `series_id`/`movie_id`, so the builder pins the **EPG (Electronic Program Guide)**: - XMLTV format support -- Background parsing in worker thread +- Background parsing in worker thread; HTTP/file gzip compatibility follows `docs/architecture/m3u-playlist-module.md` ("XMLTV response compression"). - Stored in database for quick lookup - Global display-time offset (`Settings.epgOffsetMinutes`, Settings → EPG, ±720 min, Electron only): display-only, provider data is never rewritten. Two equivalent forms in `libs/shared/interfaces/src/lib/epg-display-offset.util.ts` — `epgDisplayTimeMs` (shift the programme; `ui/epg` rendering via the `offsetMinutes` input, channel rows, dashboard/recording labels; the programme dialog and the programme guide read the store themselves) and `epgProviderClockMs` (shift "now"; every "currently airing" decision: the `GET_CURRENT_PROGRAMS_BATCH` lookup takes an explicit `nowMs` and `EpgService` tags its cache with the offset, Xtream/Stalker/M3U current-programme selection and previews, the unified collection resolver, dashboard progress, recording overlap). A consumer applies exactly one form per comparison. Contract: `docs/architecture/m3u-playlist-module.md` ("EPG display offset") - Programme guide (Electron, M3U): `app-epg-guide` in `libs/ui/epg` fed by the host-provided `EPG_GUIDE_SOURCE`; the M3U host switches into guide mode (docked player strip, no sidebar/timeline, no remount) from the header action, the palette, the EPG panel's Guide button (timeline or list view) or `G`. Data: `EPG_GET_PROGRAMS_FOR_CHANNELS` / `EPG_GET_PROGRAM_COVERAGE` (keys resolved in main; manual mappings honoured). Contract: `docs/architecture/m3u-playlist-module.md` ("Programme guide"). @@ -1755,6 +1755,14 @@ Database initialization is owned by `libs/shared/database/src/lib/connection.ts` +## XMLTV Response Compression + +Electron decodes HTTP compression before the gzip file layer. For `.gz`/gzip +metadata plus HTTP gzip, a streaming signature check unwraps one remaining +file layer while preserving single-layer providers. Errors and cancellation +close the decoding chain. Contract: `docs/architecture/m3u-playlist-module.md` +("XMLTV response compression"). + ## XMLTV Source Removal Saving Settings → EPG reconciles cached XMLTV with committed global URLs and diff --git a/apps/electron-backend-e2e/src/epg.e2e.ts b/apps/electron-backend-e2e/src/epg.e2e.ts index 83189e1c6..6fa53db09 100644 --- a/apps/electron-backend-e2e/src/epg.e2e.ts +++ b/apps/electron-backend-e2e/src/epg.e2e.ts @@ -1,5 +1,7 @@ import { readFile } from 'node:fs/promises'; +import { createServer } from 'node:http'; import { join } from 'node:path'; +import { gzipSync } from 'node:zlib'; import { Page } from '@playwright/test'; import { buildM3uContent, @@ -69,7 +71,79 @@ function formatXmltvDate(date: Date): string { ].join(''); } +async function createGzipEpgServer(xml: string, doubleGzip: boolean) { + const file = gzipSync(xml); + const body = doubleGzip ? gzipSync(file) : file; + const server = createServer((_request, response) => { + response.writeHead(200, { + 'Content-Type': 'application/gzip', + 'Content-Encoding': 'gzip', + }); + response.end(body); + }); + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, '127.0.0.1', resolve); + }); + const address = server.address(); + if (!address || typeof address === 'string') { + throw new Error('Expected a TCP address for the XMLTV fixture server'); + } + return { + url: `http://127.0.0.1:${address.port}/guide.xml.gz`, + close: () => + new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }), + }; +} + test.describe('Electron EPG', () => { + for (const [name, doubleGzip] of [ + ['double', true], + ['single', false], + ] as const) { + test(`@epg @electron imports ${name} gzip with a .gz URL and HTTP gzip`, async ({ + dataDir, + }) => { + const server = await createGzipEpgServer( + createCurrentXmltvFixture( + 'gzip-news', + 'Gzip News', + 'Gzip Bulletin' + ), + doubleGzip + ); + try { + const app = await launchElectronApp(dataDir); + try { + const result = await app.mainWindow.evaluate( + (url) => window.electron.forceFetchEpg(url), + server.url + ); + expect(result.success).toBe(true); + expect(await getEpgChannelCount(app.mainWindow)).toBe(1); + const metadata = await app.mainWindow.evaluate(() => + window.electron.getEpgChannelMetadata(['gzip-news']) + ); + expect(metadata['gzip-news']).toMatchObject({ + displayName: 'Gzip News', + }); + const programs = await app.mainWindow.evaluate(() => + window.electron.getChannelPrograms('gzip-news') + ); + expect(programs.map((program) => program.title)).toEqual([ + 'Gzip Bulletin', + ]); + } finally { + await closeElectronApp(app); + } + } finally { + await server.close(); + } + }); + } + test('@epg @electron saves unrelated settings when EPG reconciliation is unavailable', async ({ dataDir, }) => { diff --git a/apps/electron-backend/src/app/workers/epg-optional-gunzip.ts b/apps/electron-backend/src/app/workers/epg-optional-gunzip.ts new file mode 100644 index 000000000..e8e77427a --- /dev/null +++ b/apps/electron-backend/src/app/workers/epg-optional-gunzip.ts @@ -0,0 +1,54 @@ +import { Duplex, PassThrough, Readable, pipeline } from 'stream'; +import { createGunzip } from 'zlib'; + +/** Resolve the ambiguous .gz + HTTP gzip case after HTTP decoding. */ +export function createOptionalEpgGunzip(): Duplex { + const input = new PassThrough(); + const output = new PassThrough(); + // Joining both ends makes decoder errors/consumer closure interrupt even + // an outstanding prefix/body read, rather than waiting for the next chunk. + const stage = Duplex.from({ writable: input, readable: output }); + void decodeRemainingFile(input, output).catch((error: Error) => { + stage.destroy(error); + }); + return stage; +} + +async function decodeRemainingFile( + input: Readable, + output: PassThrough +): Promise { + const iterator = input.iterator({ destroyOnReturn: false }); + const prefix = Buffer.alloc(2); + const pending: Buffer[] = []; + let prefixLength = 0; + + // Retain at most two non-empty chunks, never the whole response. + while (prefixLength < prefix.length) { + const next = await iterator.next(); + if (next.done) break; + if (next.value.length === 0) continue; + pending.push(next.value); + prefixLength += next.value.copy(prefix, prefixLength); + } + + const replay = Readable.from( + (async function* () { + try { + yield* pending; + yield* iterator; + } finally { + await iterator.return?.(); + } + })(), + { objectMode: false } + ); + // One optional file layer only. Invalid gzip remains an error. + const decoder = + prefixLength === 2 && prefix[0] === 0x1f && prefix[1] === 0x8b + ? createGunzip() + : new PassThrough(); + pipeline(replay, decoder, output, (error) => { + if (error) output.destroy(error); + }); +} diff --git a/apps/electron-backend/src/app/workers/epg-stream-decoder.spec.ts b/apps/electron-backend/src/app/workers/epg-stream-decoder.spec.ts index 73c6d2882..c34bf628a 100644 --- a/apps/electron-backend/src/app/workers/epg-stream-decoder.spec.ts +++ b/apps/electron-backend/src/app/workers/epg-stream-decoder.spec.ts @@ -1,20 +1,37 @@ -import { PassThrough, Readable } from 'stream'; -import { brotliCompressSync, gzipSync } from 'zlib'; +import { randomBytes } from 'crypto'; +import { PassThrough, Readable, pipeline } from 'stream'; +import { finished } from 'stream/promises'; +import { brotliCompressSync, deflateSync, gzipSync } from 'zlib'; +import { createOptionalEpgGunzip } from './epg-optional-gunzip'; import { createDecodedEpgStream } from './epg-stream-decoder'; import { StreamingEpgParser } from './epg-streaming-parser'; const xmltvFixture = '' + - 'Test Channel'; + 'Test Channel' + + '' + + 'Test Bulletin'; async function collectDecodedText(stream: Readable): Promise { + return (await collectDecodedBytes(stream)).toString('utf-8'); +} + +async function collectDecodedBytes(stream: Readable): Promise { const chunks: Buffer[] = []; for await (const chunk of stream) { chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); } - return Buffer.concat(chunks).toString('utf-8'); + return Buffer.concat(chunks); +} + +function optionalFileStream(source: Readable): Readable { + const output = new PassThrough(); + pipeline(source, createOptionalEpgGunzip(), output, (error) => { + if (error) output.destroy(error); + }); + return output; } function parseChannelCount(xml: string): number { @@ -29,6 +46,100 @@ function parseChannelCount(xml: string): number { } describe('createDecodedEpgStream', () => { + it.each([ + ['gzip file', gzipSync(xmltvFixture), {}, true], + [ + 'HTTP deflate', + deflateSync(xmltvFixture), + { 'content-encoding': 'deflate' }, + false, + ], + [ + 'deflate over gzip file', + deflateSync(gzipSync(xmltvFixture)), + { 'content-encoding': 'deflate' }, + true, + ], + [ + 'gzip then Brotli over gzip file', + brotliCompressSync(gzipSync(gzipSync(xmltvFixture))), + { 'content-encoding': 'gzip, br' }, + true, + ], + [ + 'gzip then Brotli over XML', + brotliCompressSync(gzipSync(xmltvFixture)), + { 'content-encoding': 'gzip, br' }, + true, + ], + [ + 'two declared HTTP gzip layers over XML', + gzipSync(gzipSync(xmltvFixture)), + { 'content-encoding': 'gzip, gzip' }, + true, + ], + ] as const)( + 'preserves %s decoding', + async (_name, payload, headers, fileHint) => { + const decodedText = await collectDecodedText( + createDecodedEpgStream( + Readable.from([payload]), + headers, + fileHint + ) + ); + expect(decodedText).toBe(xmltvFixture); + expect(parseChannelCount(decodedText)).toBe(1); + } + ); + + it('does not unwrap an unadvertised file or recursively unwrap file layers', async () => { + for (const [layers, hint] of [ + [2, false], + [3, true], + ] as const) { + let payload: Buffer = Buffer.from(xmltvFixture); + for (let i = 0; i < layers; i++) payload = gzipSync(payload); + const result = await collectDecodedBytes( + createDecodedEpgStream( + Readable.from([payload]), + { 'content-encoding': 'gzip' }, + hint + ) + ); + expect(result).toEqual(gzipSync(xmltvFixture)); + } + }); + + it('decodes a gzip file inside HTTP gzip before XMLTV parsing (#1586)', async () => { + const decodedText = await collectDecodedText( + createDecodedEpgStream( + Readable.from([gzipSync(gzipSync(xmltvFixture))]), + { 'content-encoding': 'gzip' }, + true + ) + ); + + expect(decodedText).toBe(xmltvFixture); + const channels = jest.fn(); + const programs = jest.fn(); + const parser = new StreamingEpgParser( + channels, + programs, + () => undefined + ); + parser.write(decodedText); + expect(parser.finish()).toEqual({ totalChannels: 1, totalPrograms: 1 }); + expect(channels).toHaveBeenCalledWith([ + expect.objectContaining({ id: 'test' }), + ]); + expect(programs).toHaveBeenCalledWith([ + expect.objectContaining({ + title: [{ lang: '', value: 'Test Bulletin' }], + }), + ]); + }); + it('decodes Brotli transfer-encoded XML before SAX parsing', async () => { const decodedText = await collectDecodedText( createDecodedEpgStream( @@ -127,4 +238,160 @@ describe('createDecodedEpgStream', () => { await expect(readPromise).rejects.toThrow(); expect(source.destroyed).toBe(true); }); + + it.each(['invalid', 'truncated'] as const)( + 'rejects %s inner gzip', + async (kind) => { + const inner = + kind === 'invalid' + ? Buffer.from([0x1f, 0x8b, 0xff, 0, 0, 0, 0, 0, 0, 0]) + : gzipSync(xmltvFixture).subarray(0, -4); + const source = new PassThrough(); + const output = createDecodedEpgStream( + source, + { 'content-encoding': 'gzip' }, + true + ); + const collected = collectDecodedBytes(output); + // A bad header must close even a source that has not ended yet. + if (kind === 'invalid') source.write(gzipSync(inner)); + else source.end(gzipSync(inner)); + await expect(collected).rejects.toThrow(); + expect(source.destroyed).toBe(true); + expect(output.destroyed).toBe(true); + } + ); + + it('stops the entire decoding chain when the consumer closes early', async () => { + const source = new PassThrough(); + const output = createDecodedEpgStream( + source, + { 'content-encoding': 'gzip' }, + true + ); + const sourceClosed = finished(source).catch((error: Error) => error); + const iterator = output[Symbol.asyncIterator](); + source.write(gzipSync(gzipSync(Buffer.alloc(256 * 1024, 0x61)))); + expect((await iterator.next()).done).toBe(false); + await iterator.return?.(); + expect(await sourceClosed).toHaveProperty('message'); + expect(source.destroyed).toBe(true); + expect(output.destroyed).toBe(true); + }); + + it('preserves bytes and bounds read-ahead with a slow consumer', async () => { + const expected = randomBytes(1024 * 1024); + const encoded = gzipSync(gzipSync(expected)); + let bytesRead = 0; + const source = Readable.from( + (function* () { + for (let i = 0; i < encoded.length; i += 1024) { + const chunk = encoded.subarray(i, i + 1024); + bytesRead += chunk.length; + yield chunk; + } + })(), + { objectMode: false, highWaterMark: 1024 } + ); + const output = createDecodedEpgStream( + source, + { 'content-encoding': 'gzip' }, + true + ); + const chunks: Buffer[] = []; + for await (const chunk of output) { + chunks.push(chunk); + await new Promise((resolve) => setImmediate(resolve)); + if (chunks.length === 1) + expect(bytesRead).toBeLessThan(encoded.length); + } + expect(Buffer.concat(chunks)).toEqual(expected); + }); +}); + +describe('optional EPG file layer', () => { + it('waits for the second signature byte delivered in a later turn', async () => { + const payload = gzipSync(xmltvFixture); + const source = Readable.from( + (async function* () { + yield payload.subarray(0, 1); + await new Promise((resolve) => setImmediate(resolve)); + yield payload.subarray(1, 2); + await new Promise((resolve) => setImmediate(resolve)); + yield payload.subarray(2); + })() + ); + expect(await collectDecodedText(optionalFileStream(source))).toBe( + xmltvFixture + ); + }); + + it.each([ + ['gzip', gzipSync(xmltvFixture), Buffer.from(xmltvFixture)], + ['plain XML', Buffer.from(xmltvFixture), Buffer.from(xmltvFixture)], + ['empty', Buffer.alloc(0), Buffer.alloc(0)], + ['one byte', Buffer.from([0x1f]), Buffer.from([0x1f])], + [ + 'non-gzip prefix', + Buffer.from([0x1f, 0x00, 0x80]), + Buffer.from([0x1f, 0x00, 0x80]), + ], + ] as const)( + 'handles split %s prefixes and empty chunks without losing bytes', + async (_name, payload, expected) => { + const source = Readable.from([ + Buffer.alloc(0), + payload.subarray(0, 1), + Buffer.alloc(0), + payload.subarray(1, 2), + Buffer.alloc(0), + payload.subarray(2), + ]); + expect( + await collectDecodedBytes(optionalFileStream(source)) + ).toEqual(expected); + } + ); + + it.each([ + Buffer.alloc(0), + Buffer.from([0x1f]), + gzipSync(xmltvFixture).subarray(0, 10), + ])( + 'propagates source errors while waiting for a prefix or compressed body (%j)', + async (prefix) => { + const source = new PassThrough(); + const output = optionalFileStream(source); + const collected = collectDecodedBytes(output); + source.write(prefix); + await new Promise((resolve) => setImmediate(resolve)); + source.destroy(new Error('upstream failed')); + await expect(collected).rejects.toThrow('upstream failed'); + expect(output.destroyed).toBe(true); + } + ); + + it.each([ + Buffer.alloc(0), + Buffer.from([0x1f]), + gzipSync(xmltvFixture).subarray(0, 10), + ])( + 'closes a pending prefix/body read when the consumer aborts (%j)', + async (prefix) => { + const source = new PassThrough(); + const output = optionalFileStream(source); + const sourceClosed = finished(source).catch( + (error: Error) => error + ); + const outputClosed = finished(output).catch( + (error: Error) => error + ); + source.write(prefix); + await new Promise((resolve) => setImmediate(resolve)); + output.destroy(new Error('consumer stopped')); + expect(await outputClosed).toEqual(new Error('consumer stopped')); + expect(await sourceClosed).toHaveProperty('message'); + expect(source.destroyed).toBe(true); + } + ); }); diff --git a/apps/electron-backend/src/app/workers/epg-stream-decoder.ts b/apps/electron-backend/src/app/workers/epg-stream-decoder.ts index 726294e0a..ce304d17b 100644 --- a/apps/electron-backend/src/app/workers/epg-stream-decoder.ts +++ b/apps/electron-backend/src/app/workers/epg-stream-decoder.ts @@ -1,14 +1,11 @@ -import { PassThrough, Readable, Transform, pipeline } from 'stream'; -import { - createBrotliDecompress, - createGunzip, - createInflate, -} from 'zlib'; +import { Duplex, PassThrough, Readable, Transform, pipeline } from 'stream'; +import { createBrotliDecompress, createGunzip, createInflate } from 'zlib'; import { EpgResponseContentEncoding, getEpgResponseContentEncodings, HeaderReader, } from './epg-response-utils'; +import { createOptionalEpgGunzip } from './epg-optional-gunzip'; function createContentEncodingDecoder( contentEncoding: EpgResponseContentEncoding @@ -29,10 +26,16 @@ export function createDecodedEpgStream( shouldGunzipPayload: boolean ): Readable { const contentEncodings = getEpgResponseContentEncodings(headers); - const transforms = contentEncodings.map(createContentEncodingDecoder); + const transforms: Duplex[] = contentEncodings.map( + createContentEncodingDecoder + ); - if (shouldGunzipPayload && !contentEncodings.includes('gzip')) { - transforms.push(createGunzip()); + if (shouldGunzipPayload) { + transforms.push( + contentEncodings.includes('gzip') + ? createOptionalEpgGunzip() + : createGunzip() + ); } if (transforms.length === 0) { diff --git a/docs/architecture/m3u-playlist-module.md b/docs/architecture/m3u-playlist-module.md index 172ed7944..e3951a2af 100644 --- a/docs/architecture/m3u-playlist-module.md +++ b/docs/architecture/m3u-playlist-module.md @@ -1056,6 +1056,24 @@ These URLs are playlist-scoped by default: ## EPG Integration +### XMLTV response compression + +The Electron EPG worker decodes HTTP `Content-Encoding` layers in reverse +order before parsing XMLTV. The existing URL (including redirects), MIME type +and filename checks identify gzip files independently of HTTP compression. +When both a gzip file hint and an HTTP gzip layer are present, a streaming +stage checks the decoded body's first two bytes (`1F 8B`). A remaining gzip +signature triggers exactly one additional file decompression; otherwise the +body passes through unchanged. This supports both genuinely double-compressed +feeds and providers that describe a single gzip layer with both metadata forms. +Other decoding paths retain their existing behavior. + +The probe preserves bytes across chunk boundaries and keeps bounded buffering +and stream backpressure. Invalid or truncated gzip fails the import; it is not +retried as plain XML. Source errors and consumer cancellation terminate the +whole decoding chain. The decoder does not recursively unpack file layers or +change XML parsing, source reconciliation, or database persistence contracts. + ### EpgService (`@iptvnator/epg/data-access`) ```typescript