fix(epg): decode gzip files wrapped in HTTP gzip (#1589)

.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.

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

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`
 
 <!-- nx configuration end-->
 
+## 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

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<void>((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<void>((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,
     }) => {

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<void> {
+    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);
+    });
+}

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 =
     '<?xml version="1.0" encoding="utf-8" ?>' +
-    '<tv><channel id="test"><display-name>Test Channel</display-name></channel></tv>';
+    '<tv><channel id="test"><display-name>Test Channel</display-name></channel>' +
+    '<programme channel="test" start="20260912090000 +0000" stop="20260912100000 +0000">' +
+    '<title>Test Bulletin</title></programme></tv>';
 
 async function collectDecodedText(stream: Readable): Promise<string> {
+    return (await collectDecodedBytes(stream)).toString('utf-8');
+}
+
+async function collectDecodedBytes(stream: Readable): Promise<Buffer> {
     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<void>((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<void>((resolve) => setImmediate(resolve));
+                yield payload.subarray(1, 2);
+                await new Promise<void>((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<void>((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<void>((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);
+        }
+    );
 });

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

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