diff --git a/apps/electron-backend/src/app/events/epg-query.service.spec.ts b/apps/electron-backend/src/app/events/epg-query.service.spec.ts index e4603bad0..8db5fbaca 100644 --- a/apps/electron-backend/src/app/events/epg-query.service.spec.ts +++ b/apps/electron-backend/src/app/events/epg-query.service.spec.ts @@ -417,7 +417,7 @@ describe('EpgQueryService', () => { }); joinChain.where = jest.fn((condition: unknown) => { whereCalls.push(condition); - return { limit: joinLimit }; + return { orderBy: jest.fn(() => ({ limit: joinLimit })) }; }); // Program lookup for the resolved (mapped) channel id. diff --git a/apps/electron-backend/src/app/events/epg-query.service.ts b/apps/electron-backend/src/app/events/epg-query.service.ts index 68739d6a4..471721ceb 100644 --- a/apps/electron-backend/src/app/events/epg-query.service.ts +++ b/apps/electron-backend/src/app/events/epg-query.service.ts @@ -947,6 +947,13 @@ export class EpgQueryService { // The join key is built in SQL to mirror // buildXtreamEpgMappingKey(playlistId, xtreamId) → // `xtream:{playlistId}:{xtreamId}`; keep the two in sync. + // + // This layer only receives the provider epg_channel_id, not the + // caller's playlist, so when the same id exists in several + // playlists it cannot scope to the caller's own mapping — the + // renderer resolves the playlist-scoped key directly and this is + // a best-effort fallback. Order deterministically so the chosen + // mapping is at least stable rather than storage-order dependent. const mapped = await db .select({ epgChannelId: schema.epgChannelMappings.epgChannelId, @@ -964,6 +971,10 @@ export class EpgQueryService { ) ) .where(eq(schema.content.epgChannelId, channelKey)) + .orderBy( + schema.categories.playlistId, + schema.content.xtreamId + ) .limit(1); if (mapped.length > 0) { return mapped[0].epgChannelId; diff --git a/libs/portal/xtream/data-access/src/lib/services/epg-queue.service.spec.ts b/libs/portal/xtream/data-access/src/lib/services/epg-queue.service.spec.ts index 3001cc0e0..777ae736b 100644 --- a/libs/portal/xtream/data-access/src/lib/services/epg-queue.service.spec.ts +++ b/libs/portal/xtream/data-access/src/lib/services/epg-queue.service.spec.ts @@ -372,6 +372,30 @@ describe('EpgQueueService', () => { expect(priv().epgChannelByStreamId.has(555)).toBe(false); expect(priv().xmltvPreviewByStreamId.has(555)).toBe(false); }); + + it('discards an in-flight result when the mapping is invalidated mid-request', async () => { + let resolveEpg!: (items: EpgItem[]) => void; + xtreamApi.getShortEpg.mockReturnValue( + new Promise((resolve) => { + resolveEpg = resolve; + }) + ); + const emitted: number[] = []; + const sub = service.epgResult$.subscribe(({ streamId }) => + emitted.push(streamId) + ); + + const fetchPromise = priv().fetchEpg(credentials, 707); + // The user changes the mapping while the request is still running. + service.invalidate(707); + // The provider now returns the pre-change (stale) result. + resolveEpg([makeEpgItem('rtl.de', 'Stale')]); + await fetchPromise; + + expect(service.getCached(707)).toBeNull(); + expect(emitted).not.toContain(707); + sub.unsubscribe(); + }); }); function makeEpgItem(channelId: string, title: string): EpgItem { diff --git a/libs/portal/xtream/data-access/src/lib/services/epg-queue.service.ts b/libs/portal/xtream/data-access/src/lib/services/epg-queue.service.ts index 2cc41a476..21dd782c1 100644 --- a/libs/portal/xtream/data-access/src/lib/services/epg-queue.service.ts +++ b/libs/portal/xtream/data-access/src/lib/services/epg-queue.service.ts @@ -55,6 +55,8 @@ export class EpgQueueService implements OnDestroy { private readonly cache = new Map(); private queue: number[] = []; private readonly inFlight = new Set(); + /** Bumped by invalidate() so a stale in-flight result is discarded. */ + private readonly invalidationEpoch = new Map(); private readonly epgChannelByStreamId = new Map(); private readonly xmltvPreviewByStreamId = new Map(); private visibleSet = new Set(); @@ -84,12 +86,23 @@ export class EpgQueueService implements OnDestroy { * Drop every cached artifact for a stream so the next enqueue refetches * it. Used when a manual EPG mapping for the stream changes, since the * cached preview/resolution was computed for the previous mapping. + * + * Also bumps an invalidation epoch and clears `inFlight`: a request that + * was already running when the mapping changed carries the pre-change + * resolution, so its result is discarded (epoch mismatch in `fetchEpg`) + * and clearing `inFlight` lets the immediate re-enqueue schedule a fresh + * fetch through the mapping-aware `enqueue()` path. */ invalidate(streamId: number): void { this.cache.delete(streamId); this.failureTimestamps.delete(streamId); this.epgChannelByStreamId.delete(streamId); this.xmltvPreviewByStreamId.delete(streamId); + this.inFlight.delete(streamId); + this.invalidationEpoch.set( + streamId, + (this.invalidationEpoch.get(streamId) ?? 0) + 1 + ); } private isFailureCoolingDown(streamId: number): boolean { @@ -317,6 +330,9 @@ export class EpgQueueService implements OnDestroy { credentials: XtreamCredentials, streamId: number ): Promise { + const startEpoch = this.invalidationEpoch.get(streamId) ?? 0; + const isStale = (): boolean => + (this.invalidationEpoch.get(streamId) ?? 0) !== startEpoch; try { const apiItems = await this.apiService.getShortEpg( credentials, @@ -325,6 +341,11 @@ export class EpgQueueService implements OnDestroy { { suppressErrorLog: true } ); + // A mapping change during the request invalidated this result. + if (isStale()) { + return; + } + if (apiItems.length > 0) { this.recordSuccess(streamId, apiItems); return; @@ -333,6 +354,9 @@ export class EpgQueueService implements OnDestroy { const xmltv = this.xmltvPreviewByStreamId.get(streamId); this.recordSuccess(streamId, xmltv ? [xmltv] : []); } catch (error) { + if (isStale()) { + return; + } this.failureTimestamps.set(streamId, Date.now()); this.logger.error( `Failed to load EPG for stream ${streamId}`, diff --git a/libs/portal/xtream/feature/src/lib/portal-channels-list/portal-channels-list.component.ts b/libs/portal/xtream/feature/src/lib/portal-channels-list/portal-channels-list.component.ts index 3443689a8..8a2076a32 100644 --- a/libs/portal/xtream/feature/src/lib/portal-channels-list/portal-channels-list.component.ts +++ b/libs/portal/xtream/feature/src/lib/portal-channels-list/portal-channels-list.component.ts @@ -514,20 +514,26 @@ export class PortalChannelsListComponent implements AfterViewInit, OnDestroy { } const channelKey = buildXtreamEpgMappingKey(playlistId, xtreamId); - void this.openEpgMappingDialog(channelKey, channel, xtreamId); + void this.openEpgMappingDialog( + channelKey, + channel, + xtreamId, + playlistId + ); } private async openEpgMappingDialog( channelKey: string, channel: XtreamChannelListItem, - streamId: number + streamId: number, + playlistId: string ): Promise { const mappingBefore = await this.readEpgMapping(channelKey); EpgMappingDialogComponent.open(this.dialog, { channelKey, channelName: channel.title ?? channel.name ?? String(streamId), - playlistId: this.xtreamStore.currentPlaylist()?.id, + playlistId, }) .afterClosed() .subscribe(async () => {