fix(epg): remove cached XMLTV data after source deletion (#1548)

* fix(epg): remove cached XMLTV data after source deletion

* fix(epg): close source reconciliation review races

* fix(epg): serialize cleanup with replacement imports

* fix(epg): report retired worker exits as cancellations

* fix(epg): preserve source metadata through cache cleanup

* refactor(epg): separate worker runtime and import lifecycle

* fix(epg): skip cleanup for unchanged source settings

* fix(epg): cancel retired error rows and pending retries

* fix(epg): redact diagnostics and mirror committed settings after cleanup errors

* fix(epg): preserve metadata writer order independently of timestamps

apps/electron-backend/src/app/workers/epg-database.spec.ts

@@ -93,6 +93,10 @@ describe('EpgDatabase', () => {
                   FROM epg_programs
                   WHERE epg_programs.channel_id = epg_channels.id
               )
+              AND NOT EXISTS (
+                  SELECT 1 FROM epg_channel_sources
+                  WHERE epg_channel_sources.channel_id = epg_channels.id
+              )
         `);
 
         expect(preparedSql).toContain(deleteTodayAndFutureSql);
@@ -230,6 +234,7 @@ describe('EpgDatabaseClearOperation', () => {
         expect(exec.mock.calls.map(([statement]) => statement)).toEqual([
             'BEGIN',
             'DELETE FROM epg_programs',
+            'DELETE FROM epg_channel_sources',
             'DELETE FROM epg_channels',
             'COMMIT',
         ]);
@@ -250,6 +255,7 @@ describe('EpgDatabaseClearOperation', () => {
         expect(exec.mock.calls.map(([statement]) => statement)).toEqual([
             'BEGIN',
             'DELETE FROM epg_programs',
+            'DELETE FROM epg_channel_sources',
             'DELETE FROM epg_channels',
             'ROLLBACK',
         ]);
@@ -276,6 +282,10 @@ describe('EpgDatabaseSourceClearOperation', () => {
                   FROM epg_programs
                   WHERE epg_programs.channel_id = epg_channels.id
               )
+              AND NOT EXISTS (
+                  SELECT 1 FROM epg_channel_sources
+                  WHERE epg_channel_sources.channel_id = epg_channels.id
+              )
         `);
 
         expect(preparedSql).toContain(deleteProgramsSql);

apps/electron-backend/src/app/workers/epg-database.ts

@@ -1,6 +1,7 @@
 import type BetterSqlite3 from 'better-sqlite3';
 import { getIptvnatorDatabasePath } from '@iptvnator/shared/database/path-utils';
 import type { ParsedChannel, ParsedProgram } from './epg-streaming-parser';
+import { restoreSurvivingChannelMetadata } from './epg-source-metadata';
 
 /**
  * Database helper for worker-owned EPG operations.
@@ -10,6 +11,8 @@ export class EpgDatabase {
     private readonly db: BetterSqlite3.Database;
     private readonly knownChannelIds = new Set<string>();
     private readonly insertChannelStmt: BetterSqlite3.Statement;
+    private readonly insertChannelSourceStmt: BetterSqlite3.Statement;
+    private readonly deleteChannelSourcesStmt: BetterSqlite3.Statement;
     private readonly insertProgramStmt: BetterSqlite3.Statement;
     private readonly deleteOrphanChannelsForSourceStmt: BetterSqlite3.Statement;
     private readonly deleteTodayAndFutureStmt: BetterSqlite3.Statement;
@@ -30,6 +33,25 @@ export class EpgDatabase {
                 updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
         `);
 
+        // Allocated under the same SQLite write transaction as the channel
+        // upsert: ties and clock rollbacks cannot reorder surviving writers.
+        this.insertChannelSourceStmt = this.db.prepare(`
+            INSERT INTO epg_channel_sources
+                (channel_id, display_name, icon_url, url, source_url, updated_at, write_order)
+            VALUES (?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%SZ', 'now'),
+                (SELECT COALESCE(MAX(write_order), 0) + 1
+                 FROM epg_channel_sources WHERE channel_id = ?))
+            ON CONFLICT(channel_id, source_url) DO UPDATE SET
+                display_name = excluded.display_name,
+                icon_url = excluded.icon_url,
+                url = excluded.url,
+                updated_at = excluded.updated_at,
+                write_order = excluded.write_order
+        `);
+        this.deleteChannelSourcesStmt = this.db.prepare(
+            'DELETE FROM epg_channel_sources WHERE source_url = ?'
+        );
+
         // Guard against duplicate entries when the clearFirst logic misses old
         // rows. The same channel + start + title + source is treated as the
         // same programme — a later import with a corrected stop time simply
@@ -66,6 +88,10 @@ export class EpgDatabase {
                   FROM epg_programs
                   WHERE epg_programs.channel_id = epg_channels.id
               )
+              AND NOT EXISTS (
+                  SELECT 1 FROM epg_channel_sources
+                  WHERE epg_channel_sources.channel_id = epg_channels.id
+              )
         `);
 
         this.deleteTodayAndFutureStmt = this.db.prepare(`
@@ -88,7 +114,9 @@ export class EpgDatabase {
     ): void {
         const insertMany = this.db.transaction((channels: ParsedChannel[]) => {
             if (clearTodayAndFuture) {
+                restoreSurvivingChannelMetadata(this.db, sourceUrl);
                 this.deleteTodayAndFutureStmt.run(sourceUrl);
+                this.deleteChannelSourcesStmt.run(sourceUrl);
                 this.deleteOrphanChannelsForSourceStmt.run(sourceUrl);
                 this.knownChannelIds.clear();
             }
@@ -106,6 +134,14 @@ export class EpgDatabase {
                     url,
                     sourceUrl
                 );
+                this.insertChannelSourceStmt.run(
+                    channel.id,
+                    displayName,
+                    iconUrl,
+                    url,
+                    sourceUrl,
+                    channel.id
+                );
                 this.knownChannelIds.add(channel.id);
             }
         });
@@ -222,6 +258,7 @@ export class EpgDatabaseClearOperation {
         this.db.exec('BEGIN');
         try {
             this.db.exec('DELETE FROM epg_programs');
+            this.db.exec('DELETE FROM epg_channel_sources');
             this.db.exec('DELETE FROM epg_channels');
             this.db.exec('COMMIT');
         } catch (error) {
@@ -256,6 +293,10 @@ export class EpgDatabaseSourceClearOperation {
                   FROM epg_programs
                   WHERE epg_programs.channel_id = epg_channels.id
               )
+              AND NOT EXISTS (
+                  SELECT 1 FROM epg_channel_sources
+                  WHERE epg_channel_sources.channel_id = epg_channels.id
+              )
         `);
     }
 
@@ -266,7 +307,12 @@ export class EpgDatabaseSourceClearOperation {
         }
 
         const clearSource = this.db.transaction((url: string) => {
+            // Capture affected IDs while removed-source provenance still exists.
+            restoreSurvivingChannelMetadata(this.db, url);
             this.deleteProgramsForSourceStmt.run(url);
+            this.db
+                .prepare('DELETE FROM epg_channel_sources WHERE source_url = ?')
+                .run(url);
             this.deleteOrphanChannelsForSourceStmt.run(url);
         });
 

apps/electron-backend/src/app/workers/epg-parser.worker.ts

@@ -1,3 +1,4 @@
+import { epgLogger } from '../util/epg-logger';
 import type BetterSqlite3 from 'better-sqlite3';
 import {
     ELECTRON_BRIDGE_SECURITY_ERROR_CODES,
@@ -102,7 +103,7 @@ async function fetchAndParseEpgStreaming(
     url: string,
     options: ElectronBridgeTrustOptions = {}
 ): Promise<void> {
-    console.log(loggerLabel, `Fetching EPG from ${url}`);
+    epgLogger.log(loggerLabel, 'Fetching EPG source');
 
     // Create database connection in worker
     const epgDb = new EpgDatabase(Database);
@@ -141,18 +142,15 @@ async function fetchAndParseEpgStreaming(
         const contentEncoding = getEpgResponseContentEncoding(response.headers);
 
         if (responseUrl && responseUrl !== url) {
-            console.log(
-                loggerLabel,
-                `Resolved EPG redirect: ${url} -> ${responseUrl}`
-            );
+            epgLogger.log(loggerLabel, 'Resolved EPG redirect');
         }
 
-        console.log(
+        epgLogger.log(
             loggerLabel,
             `EPG response detected as gzipped: ${isGzipped}`
         );
         if (contentEncoding) {
-            console.log(
+            epgLogger.log(
                 loggerLabel,
                 `EPG response content-encoding: ${contentEncoding}`
             );
@@ -200,7 +198,7 @@ async function fetchAndParseEpgStreaming(
                 try {
                     parser.write(chunk.toString('utf-8'));
                 } catch (err) {
-                    console.error(loggerLabel, 'Parse error:', err);
+                    epgLogger.error(loggerLabel, 'Parse error:', err);
                     epgDb.close();
                     reject(err);
                 }
@@ -209,7 +207,7 @@ async function fetchAndParseEpgStreaming(
             dataStream.on('end', () => {
                 try {
                     const stats = parser.finish();
-                    console.log(
+                    epgLogger.log(
                         loggerLabel,
                         `Parsing complete: ${stats.totalChannels} channels, ${stats.totalPrograms} programs`
                     );
@@ -223,7 +221,7 @@ async function fetchAndParseEpgStreaming(
                     // real problem (unreachable feed, SAX parse failure, etc.).
                     if (stats.totalChannels === 0) {
                         const errorMessage = `EPG parse produced 0 channels — feed may be unreachable or unsupported`;
-                        console.error(loggerLabel, `${errorMessage}: ${url}`);
+                        epgLogger.error(loggerLabel, errorMessage);
                         const response: WorkerResponse = {
                             type: 'EPG_ERROR',
                             url,
@@ -251,7 +249,7 @@ async function fetchAndParseEpgStreaming(
             });
 
             dataStream.on('error', (err) => {
-                console.error(loggerLabel, 'Stream error:', err);
+                epgLogger.error(loggerLabel, 'Stream error:', err);
                 epgDb.close();
                 reject(err);
             });
@@ -315,16 +313,16 @@ function clearAllEpgData(): void {
     const clearOperation = new EpgDatabaseClearOperation(Database);
 
     try {
-        console.log(loggerLabel, 'Clearing all EPG data...');
+        epgLogger.log(loggerLabel, 'Clearing all EPG data...');
 
         clearOperation.run();
 
-        console.log(loggerLabel, 'All EPG data cleared');
+        epgLogger.log(loggerLabel, 'All EPG data cleared');
 
         const response: WorkerResponse = { type: 'CLEAR_COMPLETE' };
         parentPort?.postMessage(response);
     } catch (error) {
-        console.error(loggerLabel, 'Error clearing EPG data:', error);
+        epgLogger.error(loggerLabel, 'Error clearing EPG data:', error);
         const errorResponse: WorkerResponse = {
             type: 'EPG_ERROR',
             error: error instanceof Error ? error.message : String(error),
@@ -339,19 +337,16 @@ function clearEpgDataForSource(sourceUrl: string): void {
     const clearOperation = new EpgDatabaseSourceClearOperation(Database);
 
     try {
-        console.log(
-            loggerLabel,
-            `Clearing EPG data for source ${sourceUrl}...`
-        );
+        epgLogger.log(loggerLabel, 'Clearing EPG data for source...');
 
         clearOperation.run(sourceUrl);
 
-        console.log(loggerLabel, `EPG data cleared for source ${sourceUrl}`);
+        epgLogger.log(loggerLabel, 'EPG data cleared for source');
 
         const response: WorkerResponse = { type: 'CLEAR_COMPLETE' };
         parentPort?.postMessage(response);
     } catch (error) {
-        console.error(loggerLabel, 'Error clearing EPG source data:', error);
+        epgLogger.error(loggerLabel, 'Error clearing EPG source data:', error);
         const errorResponse: WorkerResponse = {
             type: 'EPG_ERROR',
             error: error instanceof Error ? error.message : String(error),
@@ -379,7 +374,7 @@ if (parentPort) {
                 clearEpgDataForSource(message.sourceUrl ?? '');
             }
         } catch (error) {
-            console.error(loggerLabel, 'Worker error:', error);
+            epgLogger.error(loggerLabel, 'Worker error:', error);
             const typedError = error as {
                 code?: ElectronBridgeSecurityErrorCode;
                 host?: string;
@@ -398,5 +393,5 @@ if (parentPort) {
     // Notify parent that worker is ready
     parentPort.postMessage({ type: 'READY' });
 } else {
-    console.error(loggerLabel, 'parentPort is not available!');
+    epgLogger.error(loggerLabel, 'parentPort is not available!');
 }

apps/electron-backend/src/app/workers/epg-source-metadata.spec.ts

@@ -0,0 +1,234 @@
+import { execFileSync } from 'node:child_process';
+import { createRequire } from 'node:module';
+import { resolve } from 'node:path';
+import { pathToFileURL } from 'node:url';
+
+/** Exercise worker SQL with Electron's native SQLite binding, including upgrades. */
+function runDatabaseScenario(scenario: string): unknown {
+    const moduleUrl = pathToFileURL(resolve(__dirname, 'epg-database.ts')).href;
+    const connectionUrl = pathToFileURL(
+        resolve(process.cwd(), 'libs/shared/database/src/lib/connection.ts')
+    ).href;
+    const script = `
+        const { default: Database } = await import('better-sqlite3');
+        const { EpgDatabase, EpgDatabaseSourceClearOperation, EpgDatabaseClearOperation }
+            = await import(${JSON.stringify(moduleUrl)});
+        const { __databaseConnectionTestHooks } = await import(${JSON.stringify(connectionUrl)});
+        const sqlite = new Database(':memory:');
+        const statements = __databaseConnectionTestHooks.createTableStatements
+            .filter(sql => /(?:TABLE|INDEX|TRIGGER) IF NOT EXISTS (?:idx_)?epg_/.test(sql));
+        for (const sql of statements) sqlite.exec(sql);
+        class SharedDatabase { constructor() { return sqlite; } }
+        const worker = new EpgDatabase(SharedDatabase);
+        const clear = new EpgDatabaseSourceClearOperation(SharedDatabase);
+        const importSource = (source, programs = true, clearFirst = false) => {
+            worker.insertChannels([{
+                id: 'shared', displayName: [{ value: source + ' News' }],
+                icon: [{ src: source + '.png' }], url: [source + '.website']
+            }], source, clearFirst);
+            if (programs) worker.insertPrograms([{
+                channel: 'shared', start: '2099-01-01T00:00:00Z',
+                stop: '2099-01-01T01:00:00Z', title: [{value: source}]
+            }], source);
+        };
+        const channel = () => sqlite.prepare('SELECT * FROM epg_channels WHERE id = ?').get('shared');
+        ${scenario}
+        sqlite.close();
+    `;
+    return JSON.parse(
+        execFileSync(
+            createRequire(__filename)('electron'),
+            ['--import', 'tsx', '--eval', script],
+            {
+                cwd: process.cwd(),
+                encoding: 'utf8',
+                env: {
+                    ...process.env,
+                    ELECTRON_RUN_AS_NODE: '1',
+                    TSX_TSCONFIG_PATH: resolve(
+                        process.cwd(),
+                        'tsconfig.base.json'
+                    ),
+                },
+            }
+        )
+    );
+}
+
+describe('source-owned XMLTV channel metadata', () => {
+    it.each(['A', 'B'])(
+        'restores the survivor when removing %s, then removes the final owner',
+        (removed) => {
+            const survivor = removed === 'A' ? 'B' : 'A';
+            const result = runDatabaseScenario(`
+            importSource('A'); importSource('B');
+            clear.run('${removed}');
+            clear.run('${removed}');
+            const surviving = channel();
+            const programs = sqlite.prepare('SELECT source_url FROM epg_programs').all();
+            clear.run('${survivor}');
+            process.stdout.write(JSON.stringify({ surviving, programs, remaining: channel() ?? null }));
+        `);
+            expect(result).toEqual({
+                surviving: expect.objectContaining({
+                    display_name: `${survivor} News`,
+                    icon_url: `${survivor}.png`,
+                    url: `${survivor}.website`,
+                    source_url: survivor,
+                }),
+                programs: [{ source_url: survivor }],
+                remaining: null,
+            });
+        }
+    );
+
+    it.each(['same tick', 'clock rollback'])(
+        'restores the latest surviving write across %s and source upserts',
+        (clockMode) => {
+            expect(
+                runDatabaseScenario(`
+            let timestamp = '2026-09-06T10:00:00.000Z';
+            sqlite.function('strftime', { varargs: true }, () => timestamp);
+            importSource('A');
+            if ('${clockMode}' === 'clock rollback') timestamp = '2026-09-06T09:00:00.000Z';
+            importSource('Z');
+            importSource('M');
+            clear.run('M');
+            const first = channel();
+            importSource('A');
+            importSource('M');
+            clear.run('M');
+            const updated = channel();
+            importSource('Z', true, true);
+            importSource('M');
+            clear.run('M');
+            process.stdout.write(JSON.stringify({first, updated, refreshed: channel()}));
+        `)
+            ).toEqual({
+                first: expect.objectContaining({
+                    display_name: 'Z News',
+                    icon_url: 'Z.png',
+                    url: 'Z.website',
+                    source_url: 'Z',
+                }),
+                updated: expect.objectContaining({
+                    display_name: 'A News',
+                    icon_url: 'A.png',
+                    url: 'A.website',
+                    source_url: 'A',
+                }),
+                refreshed: expect.objectContaining({
+                    display_name: 'Z News',
+                    icon_url: 'Z.png',
+                    url: 'Z.website',
+                    source_url: 'Z',
+                }),
+            });
+        }
+    );
+
+    it('retains metadata-only owners during another source refresh and removal', () => {
+        expect(
+            runDatabaseScenario(`
+            importSource('A', false); importSource('B', false);
+            worker.insertChannels([], 'A', true);
+            clear.run('A');
+            const surviving = channel();
+            clear.run('B');
+            process.stdout.write(JSON.stringify({surviving, remaining: channel() ?? null}));
+        `)
+        ).toEqual({
+            surviving: expect.objectContaining({
+                display_name: 'B News',
+                source_url: 'B',
+            }),
+            remaining: null,
+        });
+    });
+
+    it('upgrades an existing ledger without losing snapshots and orders subsequent imports', () => {
+        expect(
+            runDatabaseScenario(`
+            importSource('A'); importSource('Z');
+            sqlite.exec('ALTER TABLE epg_channel_sources DROP COLUMN write_order');
+            const migration = __databaseConnectionTestHooks.columnMigrationStatements
+                .find(sql => sql.startsWith('ALTER TABLE epg_channel_sources'));
+            sqlite.exec(migration);
+            const snapshots = sqlite.prepare('SELECT source_url, write_order FROM epg_channel_sources ORDER BY source_url').all();
+            importSource('Z'); importSource('M');
+            clear.run('M');
+            process.stdout.write(JSON.stringify({ snapshots, surviving: channel() }));
+        `)
+        ).toEqual({
+            snapshots: [
+                { source_url: 'A', write_order: 0 },
+                { source_url: 'Z', write_order: 0 },
+            ],
+            surviving: expect.objectContaining({
+                display_name: 'Z News',
+                source_url: 'Z',
+            }),
+        });
+    });
+
+    it('removes an orphan whose original owner disappeared during refresh', () => {
+        expect(
+            runDatabaseScenario(`
+            importSource('A', false); importSource('B', false);
+            worker.insertChannels([], 'A', true);
+            clear.run('B');
+            process.stdout.write(JSON.stringify(channel() ?? null));
+        `)
+        ).toBeNull();
+    });
+
+    it('restores metadata before a refresh removes the last writer provenance', () => {
+        expect(
+            runDatabaseScenario(`
+            importSource('A'); importSource('B');
+            worker.insertChannels([], 'B', true);
+            clear.run('B');
+            process.stdout.write(JSON.stringify(channel()));
+        `)
+        ).toEqual(
+            expect.objectContaining({
+                display_name: 'A News',
+                icon_url: 'A.png',
+                source_url: 'A',
+            })
+        );
+    });
+
+    it('neutralizes ambiguous legacy metadata while preserving surviving programmes', () => {
+        expect(
+            runDatabaseScenario(`
+            importSource('A'); importSource('B');
+            sqlite.exec('DROP TABLE IF EXISTS epg_channel_sources');
+            for (const sql of statements) sqlite.exec(sql);
+            clear.run('B');
+            process.stdout.write(JSON.stringify(channel()));
+        `)
+        ).toEqual(
+            expect.objectContaining({
+                display_name: 'shared',
+                icon_url: null,
+                url: null,
+                source_url: 'A',
+                updated_at: null,
+            })
+        );
+    });
+
+    it('clears all metadata and permits a clean reimport', () => {
+        expect(
+            runDatabaseScenario(`
+            importSource('A'); importSource('B');
+            new EpgDatabaseClearOperation(SharedDatabase).run();
+            const count = sqlite.prepare('SELECT count(*) AS count FROM epg_channel_sources').get().count;
+            importSource('C');
+            clear.run('C');
+            process.stdout.write(JSON.stringify({ count, remaining: channel() ?? null }));
+        `)
+        ).toEqual({ count: 0, remaining: null });
+    });
+});

apps/electron-backend/src/app/workers/epg-source-metadata.ts

@@ -0,0 +1,32 @@
+import type BetterSqlite3 from 'better-sqlite3';
+
+/** Called inside the source-clear transaction, before deleting its provenance. */
+export function restoreSurvivingChannelMetadata(
+    db: BetterSqlite3.Database,
+    sourceUrl: string
+): void {
+    // Existing global rows cannot be backfilled: their last metadata writer may
+    // differ from source_url. Only imports recorded in the ledger prove origin.
+    const survivingValue = (column: string) => `(
+        SELECT ${column} FROM epg_channel_sources
+        WHERE channel_id = epg_channels.id AND source_url != @sourceUrl
+        ORDER BY write_order DESC, updated_at DESC, source_url LIMIT 1
+    )`;
+    db.prepare(
+        `
+        UPDATE epg_channels SET
+            display_name = COALESCE(${survivingValue('display_name')}, id),
+            icon_url = ${survivingValue('icon_url')},
+            url = ${survivingValue('url')},
+            updated_at = ${survivingValue('updated_at')},
+            source_url = COALESCE(${survivingValue('source_url')}, (
+                SELECT source_url FROM epg_programs
+                WHERE channel_id = epg_channels.id AND source_url != @sourceUrl
+                ORDER BY source_url LIMIT 1
+            ), @sourceUrl)
+        WHERE source_url = @sourceUrl
+            OR id IN (SELECT channel_id FROM epg_channel_sources WHERE source_url = @sourceUrl)
+            OR id IN (SELECT channel_id FROM epg_programs WHERE source_url = @sourceUrl)
+    `
+    ).run({ sourceUrl });
+}