feat(epg): support playlist-scoped sources

Add playlist-scoped EPG source support for M3U playlists.

libs/shared/database/src/lib/connection.spec.ts

@@ -82,6 +82,9 @@ describe('database schema statements', () => {
             expect.arrayContaining([
                 'ALTER TABLE categories ADD COLUMN hidden INTEGER DEFAULT 0',
                 'ALTER TABLE playlists ADD COLUMN payload TEXT',
+                'ALTER TABLE playlists ADD COLUMN detected_epg_urls TEXT',
+                'ALTER TABLE playlists ADD COLUMN manual_epg_urls TEXT',
+                'ALTER TABLE playlists ADD COLUMN disabled_epg_urls TEXT',
                 'ALTER TABLE favorites ADD COLUMN position INTEGER DEFAULT 0',
                 'ALTER TABLE content ADD COLUMN backdrop_url TEXT',
             ])
@@ -102,10 +105,26 @@ describe('database schema statements', () => {
                 'CREATE UNIQUE INDEX IF NOT EXISTS categories_playlist_type_xtream_unique ON categories(playlist_id, type, xtream_id)',
                 'CREATE UNIQUE INDEX IF NOT EXISTS content_category_type_xtream_unique ON content(category_id, type, xtream_id)',
                 'CREATE INDEX IF NOT EXISTS favorites_playlist_position_idx ON favorites(playlist_id, position, added_at DESC)',
+                'CREATE INDEX IF NOT EXISTS idx_epg_programs_source ON epg_programs(source_url)',
+                'CREATE INDEX IF NOT EXISTS idx_epg_programs_source_time_range ON epg_programs(source_url, channel_id, start, stop)',
             ])
         );
     });
 
+    it('creates indexes for migrated EPG program columns only after column migrations run', () => {
+        const createSchemaSql = createTableStatements.map(compactSql);
+
+        expect(createSchemaSql).not.toContain(
+            'CREATE INDEX IF NOT EXISTS idx_epg_programs_source ON epg_programs(source_url)'
+        );
+        expect(createSchemaSql).not.toContain(
+            'CREATE INDEX IF NOT EXISTS idx_epg_programs_source_time_range ON epg_programs(source_url, channel_id, start, stop)'
+        );
+        expect(columnMigrationStatements.map(compactSql)).toContain(
+            'ALTER TABLE epg_programs ADD COLUMN source_url TEXT'
+        );
+    });
+
     it('does not define duplicate fresh-install schema object names', () => {
         const objectNames = [
             ...createdObjectNames(
@@ -307,4 +326,102 @@ describe('database schema statements', () => {
             expect.stringContaining('UPDATE content')
         );
     });
+
+    it('backfills migrated EPG program source URLs after creating scoped EPG indexes', () => {
+        const callOrder: string[] = [];
+        const runMigrations = (
+            __databaseConnectionTestHooks as unknown as {
+                runMigrations: (sqlite: {
+                    exec: (statement: string) => void;
+                    prepare: (statement: string) => {
+                        all?: () => unknown[];
+                        get?: (...args: unknown[]) => unknown;
+                        run?: (...args: unknown[]) => unknown;
+                    };
+                    transaction: (callback: () => void) => () => void;
+                }) => void;
+            }
+        ).runMigrations;
+        const sqlite = {
+            exec: jest.fn((statement: string) => {
+                if (statement.includes('idx_epg_programs_source')) {
+                    callOrder.push('epg-source-index');
+                }
+            }),
+            prepare: jest.fn((statement: string) => {
+                if (statement.includes('SELECT value FROM app_state')) {
+                    return {
+                        get: (): undefined => undefined,
+                    };
+                }
+                if (statement.includes('UPDATE epg_programs')) {
+                    return {
+                        run: () => {
+                            callOrder.push('epg-source-backfill');
+                        },
+                    };
+                }
+                if (statement.includes('INSERT INTO app_state')) {
+                    return { run: jest.fn() };
+                }
+
+                return {
+                    all: () => [],
+                    get: (): undefined => undefined,
+                    run: jest.fn(),
+                };
+            }),
+            transaction: jest.fn((callback: () => void) => callback),
+        };
+
+        runMigrations(sqlite);
+
+        expect(callOrder).toContain('epg-source-backfill');
+        expect(callOrder).toContain('epg-source-index');
+        expect(callOrder.indexOf('epg-source-index')).toBeLessThan(
+            callOrder.indexOf('epg-source-backfill')
+        );
+    });
+
+    it('backfills migrated EPG program source URLs in bounded batches', () => {
+        const { backfillEpgProgramSourceUrls } =
+            __databaseConnectionTestHooks;
+        let updateStatement = '';
+        const backfillRun = jest
+            .fn()
+            .mockReturnValueOnce({ changes: 50_000 })
+            .mockReturnValueOnce({ changes: 12 });
+        const stateRun = jest.fn();
+        const prepare = jest.fn((statement: string) => {
+            if (statement.includes('SELECT value FROM app_state')) {
+                return {
+                    get: (): undefined => undefined,
+                };
+            }
+            if (statement.includes('UPDATE epg_programs')) {
+                updateStatement = compactSql(statement);
+                return {
+                    run: backfillRun,
+                };
+            }
+            if (statement.includes('INSERT INTO app_state')) {
+                return { run: stateRun };
+            }
+
+            throw new Error(`Unexpected statement: ${compactSql(statement)}`);
+        });
+        const transaction = jest.fn((callback: () => void) => callback);
+        const sqlite = {
+            prepare,
+            transaction,
+        } as unknown as Parameters<typeof backfillEpgProgramSourceUrls>[0];
+
+        backfillEpgProgramSourceUrls(sqlite);
+
+        expect(updateStatement).toContain('LIMIT 50000');
+        expect(backfillRun).toHaveBeenCalledTimes(2);
+        expect(stateRun).toHaveBeenCalledWith(
+            'migration:epg-program-source-url-backfill:v1'
+        );
+    });
 });

libs/shared/database/src/lib/connection.ts

@@ -28,6 +28,9 @@ const XTREAM_ADDED_EPOCH_SECONDS_MIGRATION_KEY =
     'migration:xtream-content-added-epoch-seconds:v1';
 const CONTENT_TITLE_FTS_MIGRATION_KEY =
     'migration:content-title-fts-trigram:v1';
+const EPG_PROGRAM_SOURCE_URL_BACKFILL_MIGRATION_KEY =
+    'migration:epg-program-source-url-backfill:v1';
+const EPG_PROGRAM_SOURCE_URL_BACKFILL_BATCH_SIZE = 50_000;
 
 function readTraceFlag(name: string): boolean {
     const value = process.env[name]?.trim().toLowerCase();
@@ -84,6 +87,10 @@ const CREATE_TABLE_STATEMENTS = [
       origin TEXT,
       referrer TEXT,
       filePath TEXT,
+      epg_urls TEXT,
+      detected_epg_urls TEXT,
+      manual_epg_urls TEXT,
+      disabled_epg_urls TEXT,
       autoRefresh INTEGER DEFAULT 0,
       macAddress TEXT,
       url TEXT,
@@ -210,6 +217,7 @@ const CREATE_TABLE_STATEMENTS = [
       icon_url TEXT,
       rating TEXT,
       episode_num TEXT,
+      source_url TEXT,
       FOREIGN KEY (channel_id) REFERENCES epg_channels(id) ON DELETE CASCADE
   )`,
     // EPG indexes
@@ -304,6 +312,11 @@ const COLUMN_MIGRATION_STATEMENTS = [
     `ALTER TABLE playlists ADD COLUMN favorites TEXT`,
     `ALTER TABLE playlists ADD COLUMN recently_viewed TEXT`,
     `ALTER TABLE playlists ADD COLUMN payload TEXT`,
+    // v1.2.1: Keep M3U-detected EPG URLs available in lightweight playlist metadata
+    `ALTER TABLE playlists ADD COLUMN epg_urls TEXT`,
+    `ALTER TABLE playlists ADD COLUMN detected_epg_urls TEXT`,
+    `ALTER TABLE playlists ADD COLUMN manual_epg_urls TEXT`,
+    `ALTER TABLE playlists ADD COLUMN disabled_epg_urls TEXT`,
     // v1.2.0 -> v1.3.0: Add position column to favorites for global favorites ordering
     `ALTER TABLE favorites ADD COLUMN position INTEGER DEFAULT 0`,
     // v1.4.0 -> v1.5.0: Preserve Xtream live metadata required for EPG/catch-up
@@ -313,6 +326,8 @@ const COLUMN_MIGRATION_STATEMENTS = [
     `ALTER TABLE content ADD COLUMN direct_source TEXT`,
     // v1.5.0 -> v1.6.0: Cinematic backdrop persisted on first detail fetch
     `ALTER TABLE content ADD COLUMN backdrop_url TEXT`,
+    // v1.7.1: Scope XMLTV programs to their source URL for playlist-local EPG lookup
+    `ALTER TABLE epg_programs ADD COLUMN source_url TEXT`,
 ];
 
 const INDEX_MIGRATION_STATEMENTS = [
@@ -321,6 +336,9 @@ const INDEX_MIGRATION_STATEMENTS = [
     `CREATE UNIQUE INDEX IF NOT EXISTS content_category_type_xtream_unique ON content(category_id, type, xtream_id)`,
     // v1.6.0 -> v1.7.0: Query global favorites in stable display order
     `CREATE INDEX IF NOT EXISTS favorites_playlist_position_idx ON favorites(playlist_id, position, added_at DESC)`,
+    // v1.7.1 -> v1.7.2: Query playlist-scoped EPG by source URL and channel/time
+    `CREATE INDEX IF NOT EXISTS idx_epg_programs_source ON epg_programs(source_url)`,
+    `CREATE INDEX IF NOT EXISTS idx_epg_programs_source_time_range ON epg_programs(source_url, channel_id, start, stop)`,
 ];
 
 export const __databaseConnectionTestHooks = {
@@ -329,6 +347,7 @@ export const __databaseConnectionTestHooks = {
     indexMigrationStatements: INDEX_MIGRATION_STATEMENTS,
     normalizeXtreamContentAddedEpochs,
     ensureContentTitleFts,
+    backfillEpgProgramSourceUrls,
     runMigrations,
 } as const;
 
@@ -594,6 +613,78 @@ function ensureContentTitleFts(sqliteDb: Database.Database): void {
     }
 }
 
+function backfillEpgProgramSourceUrls(sqliteDb: Database.Database): void {
+    try {
+        const migrationState = sqliteDb
+            .prepare(`SELECT value FROM app_state WHERE key = ?`)
+            .get(EPG_PROGRAM_SOURCE_URL_BACKFILL_MIGRATION_KEY) as
+            | { value?: unknown }
+            | undefined;
+
+        if (migrationState?.value === 'done') {
+            return;
+        }
+
+        const backfillStatement = sqliteDb.prepare(
+            `UPDATE epg_programs
+             SET source_url = (
+                 SELECT epg_channels.source_url
+                 FROM epg_channels
+                 WHERE epg_channels.id = epg_programs.channel_id
+                 LIMIT 1
+             )
+             WHERE id IN (
+                 SELECT pending_programs.id
+                 FROM epg_programs AS pending_programs
+                 JOIN epg_channels
+                   ON epg_channels.id = pending_programs.channel_id
+                 WHERE pending_programs.source_url IS NULL
+                   AND epg_channels.source_url IS NOT NULL
+                   AND epg_channels.source_url <> ''
+                 LIMIT ${EPG_PROGRAM_SOURCE_URL_BACKFILL_BATCH_SIZE}
+             )`
+        );
+        const markMigrationDoneStatement = sqliteDb.prepare(
+            `INSERT INTO app_state (key, value, updated_at)
+             VALUES (?, 'done', datetime('now'))
+             ON CONFLICT(key) DO UPDATE SET
+                value = excluded.value,
+                updated_at = excluded.updated_at`
+        );
+
+        const executeBackfillBatch = sqliteDb.transaction((): number => {
+            const result = backfillStatement.run();
+            return typeof result === 'object' &&
+                result !== null &&
+                'changes' in result &&
+                typeof result.changes === 'number'
+                ? result.changes
+                : 0;
+        });
+        const markMigrationDone = sqliteDb.transaction(() => {
+            markMigrationDoneStatement.run(
+                EPG_PROGRAM_SOURCE_URL_BACKFILL_MIGRATION_KEY
+            );
+        });
+
+        let updatedRows = 0;
+        do {
+            updatedRows = executeBackfillBatch();
+        } while (updatedRows === EPG_PROGRAM_SOURCE_URL_BACKFILL_BATCH_SIZE);
+
+        markMigrationDone();
+    } catch (error) {
+        const message =
+            typeof error === 'object' && error !== null && 'message' in error
+                ? String((error as { message?: unknown }).message ?? error)
+                : String(error);
+
+        console.warn(
+            `EPG program source URL backfill failed (continuing): ${message}`
+        );
+    }
+}
+
 function runMigrationStatements(
     sqliteDb: Database.Database,
     statements: string[]
@@ -631,6 +722,7 @@ function runMigrations(sqliteDb: Database.Database): void {
     deduplicateXtreamCache(sqliteDb);
     normalizeXtreamContentAddedEpochs(sqliteDb);
     runMigrationStatements(sqliteDb, INDEX_MIGRATION_STATEMENTS);
+    backfillEpgProgramSourceUrls(sqliteDb);
 }
 
 export interface DatabaseOptions {

libs/shared/database/src/lib/schema.ts

@@ -32,6 +32,10 @@ export const playlists = sqliteTable('playlists', {
     origin: text('origin'),
     referrer: text('referrer'),
     filePath: text('filePath'),
+    epgUrls: text('epg_urls'),
+    detectedEpgUrls: text('detected_epg_urls'),
+    manualEpgUrls: text('manual_epg_urls'),
+    disabledEpgUrls: text('disabled_epg_urls'),
     autoRefresh: integer('autoRefresh', { mode: 'boolean' }).default(false),
     macAddress: text('macAddress'),
     url: text('url'),
@@ -204,9 +208,11 @@ export const epgPrograms = sqliteTable(
         iconUrl: text('icon_url'),
         rating: text('rating'),
         episodeNum: text('episode_num'),
+        sourceUrl: text('source_url'),
     },
     (table) => ({
         channelIdx: index('idx_epg_programs_channel').on(table.channelId),
+        sourceIdx: index('idx_epg_programs_source').on(table.sourceUrl),
         startIdx: index('idx_epg_programs_start').on(table.start),
         stopIdx: index('idx_epg_programs_stop').on(table.stop),
         timeRangeIdx: index('idx_epg_programs_time_range').on(
@@ -214,6 +220,12 @@ export const epgPrograms = sqliteTable(
             table.start,
             table.stop
         ),
+        sourceTimeRangeIdx: index('idx_epg_programs_source_time_range').on(
+            table.sourceUrl,
+            table.channelId,
+            table.start,
+            table.stop
+        ),
     })
 );
 
@@ -256,9 +268,10 @@ export const playbackPositions = sqliteTable(
             table.seriesXtreamId
         ),
         updatedIdx: index('playback_positions_updated_idx').on(table.updatedAt),
-        playlistUpdatedIdx: index(
-            'playback_positions_playlist_updated_idx'
-        ).on(table.playlistId, sql`${table.updatedAt} DESC`),
+        playlistUpdatedIdx: index('playback_positions_playlist_updated_idx').on(
+            table.playlistId,
+            sql`${table.updatedAt} DESC`
+        ),
     })
 );