diff --git a/.changes/stalker-itv-epg-current-program.md b/.changes/stalker-itv-epg-current-program.md new file mode 100644 index 000000000..ae1349cf9 --- /dev/null +++ b/.changes/stalker-itv-epg-current-program.md @@ -0,0 +1,9 @@ +--- +type: fix +area: stalker +--- + +Stalker Live TV now shows the currently airing programme even on portals whose +bulk EPG only lists upcoming shows: channel rows fall back to a per-channel +short-EPG lookup, and the EPG panel merges "what's on now" into the schedule +instead of showing future programmes only. diff --git a/.codex/skills/stalker-portal/SKILL.md b/.codex/skills/stalker-portal/SKILL.md index 0ebfffadb..b80f6b7c4 100644 --- a/.codex/skills/stalker-portal/SKILL.md +++ b/.codex/skills/stalker-portal/SKILL.md @@ -67,8 +67,9 @@ season/episode numbers. Keep them on subsequent position writes. ## Live Contract -- Start bulk ITV EPG eagerly once channel rows exist. Rows read the bulk cache; - only the active channel may fall back to `get_short_epg`. +- Start bulk ITV EPG eagerly once channel rows exist. Rows read the bulk + cache, falling back to throttled `get_short_epg` previews when it lacks + "now". - Radio skips EPG and external players, preserves live collection identity with `radio: 'true'`, and uses the shared inline audio player. diff --git a/docs/architecture/stalker-epg.md b/docs/architecture/stalker-epg.md index 95833c2e5..cec340428 100644 --- a/docs/architecture/stalker-epg.md +++ b/docs/architecture/stalker-epg.md @@ -15,12 +15,15 @@ Stalker now uses two EPG paths with different purposes: - The active channel EPG panel uses `get_epg_info` as a bulk endpoint, fetches a 7-day window once per playlist session, caches programs by channel id, and renders the selected channel through the shared `app-epg-timeline` component. -- Channel rows never send per-row EPG requests. The bulk EPG load is triggered +- Channel rows read the bulk cache first. The bulk EPG load is triggered **eagerly when a category's channels first render** (a constructor effect in `StalkerLiveStreamLayoutComponent` calls `ensureBulkItvEpg(168)` once ITV channels are present) — not only after the first channel is played — so the row "now playing" previews and the EPG panel populate immediately. Rows derive - their current program and progress bar from the cached bulk map. + their current program and progress bar from the cached bulk map; rows the + settled bulk guide cannot answer fall back to throttled per-channel + `get_short_epg` through `StalkerEpgPreviewQueue` (see "Channel row preview + flow"). - Effect ordering matters: the eager-EPG effect is registered **after** the playlist-change effect that calls `clearBulkItvEpgCache()`. On a portal switch the cache is cleared first and then refilled; if the order is @@ -72,7 +75,7 @@ date-navigator UI used in the M3U/Xtream flows. ## Stalker EPG API -### `get_short_epg` (active-panel fallback) +### `get_short_epg` (active-panel and row-preview fallback) **Request** @@ -83,6 +86,7 @@ GET load.php?type=itv&action=get_short_epg&ch_id={channel_id}&size={n}&JsHttpReq **Current usage** - Active panel fallback path: `size=10` +- Row-preview fallback queue: `size=3` (`EPG_PREVIEW_FETCH_SIZE`) **Response** @@ -109,8 +113,9 @@ GET load.php?type=itv&action=get_short_epg&ch_id={channel_id}&size={n}&JsHttpReq **Notes** - The response is normalized into shared `EpgItem[]` -- Only the active-panel fallback uses this path and maps the result into - controlled `EpgProgram[]` +- Two fallback consumers use this path and map the result into controlled + `EpgProgram[]`: the active-panel fallback and the throttled row-preview + queue (both only when the bulk guide cannot answer "what's on now") ### `get_epg_info` (bulk row-preview and active-panel source) @@ -159,7 +164,8 @@ GET load.php?type=itv&action=get_epg_info&period={hours}&JsHttpRequest=1-xml ### Fallback data (`get_short_epg`) → `EpgItem` -The short EPG path now exists only for the active-panel fallback flow. +The short EPG path serves the two fallback flows: the active panel and the +throttled row-preview queue. Key mapped fields: @@ -220,9 +226,14 @@ playlists. 2. The component ensures playback link resolution as before 3. The component ensures `ensureBulkItvEpg(168)` has run; the eager row effect normally started the same de-duplicated request before playback -4. `selectedItvEpgPrograms()` feeds `app-epg-timeline` -5. If the selected channel has no bulk programs, the component falls back to - `get_short_epg` +4. `selectedItvEpgPrograms()` merged with the short-EPG fallback feeds + `app-epg-timeline` (`mergeEpgProgramLists`; bulk wins an exact start-time + collision) +5. The component falls back to `get_short_epg` whenever the bulk list cannot + answer "what's on now" — because it is empty **or** because it only carries + future programmes (some portals' `get_epg_info` omits the currently airing + one). The fallback fills the gap; the bulk data keeps providing the days + ahead. The active panel no longer uses local EPG pagination or a "Load more" button. When Stalker live TV is playing through an internal player, the active panel is @@ -234,13 +245,38 @@ stream URL has been resolved; external playback keeps the full EPG-only panel. Once non-radio ITV channels render, the post-reset component effect calls `ensureBulkItvEpg(168)`. It starts eagerly before playback and is de-duplicated -against the active-channel path. Individual rows never issue per-row requests. -As soon as the bulk request completes, visible row previews derive locally -from `bulkItvEpgByChannel`: +against the active-channel path. As soon as the bulk request completes, visible +row previews derive locally from `bulkItvEpgByChannel`: - pick the current program for the channel, if one exists - compute progress from the cached program timestamps -- leave the row in its existing placeholder state when no current program exists + +Rows the bulk guide cannot answer fall back to per-channel `get_short_epg` +through `StalkerEpgPreviewQueue` +(`stalker-live-stream-layout/stalker-live-epg-preview.ts`), mirroring the +Xtream `EpgQueueService`: the queue only starts after the bulk request has +settled (so it never races the answer it is a fallback for), fetches the +currently rendered channels with bounded concurrency and inter-request +spacing, caches results — including empty ones — for five minutes, and is +reset on playlist switch because channel ids are only unique per portal. +Each sync's backlog is additionally capped (30 channels, top of the list +first) and the sidebar's scroll handler re-syncs (throttled) to fill the +next gaps, so request count tracks how far the user actually scrolls rather +than how many rows are rendered. Channels with a manual XMLTV mapping are +excluded from the fallback entirely — their bulk record holds the mapped +schedule, and the portal short EPG must not stand in for the data the +mapping deliberately replaces. Because a fetch can be enqueued before the +mapping lookup resolves, the queue's completion callback revalidates +ownership: a row claimed in the meantime by a mapping override or by bulk +data is never overwritten by the late portal response. Mapping ownership is +a fact of the saved mapping row, independent of whether the mapped guide +currently has programs — an empty mapped guide still keeps the portal EPG +out. Ownership changes are published reactively (`applyMappedItvEpg` +re-patches the bulk record even when the mapped guide contributed nothing), +so a fallback row rendered before the mapping lookup finished is removed by +the rerun sync. The backlog is superseded whenever the rendered list empties (a +legacy-paged category switch) or the view leaves ITV (radio), so abandoned +rows stop consuming portal request capacity. ## Cache Lifecycle @@ -268,12 +304,28 @@ therefore falls back to `get_short_epg` when: - the bulk request fails - the bulk response is empty -- the selected channel has no programs in the cached bulk map +- the selected channel has no **currently airing** program in the cached bulk + map — a bulk list of future-only programmes is treated as incomplete, not as + an answer -This keeps the panel usable even on limited portals, while still taking -advantage of the richer bulk API when it is available. Row previews do not -fallback to per-channel requests in this mode; they remain empty until bulk EPG -is available. +The fallback is merged with the bulk list rather than replacing it, so the +panel shows "now" from the short EPG and the days ahead from the bulk guide. +The stored fallback is tagged with the channel it was fetched for and the +merge only applies while that channel is still selected — a channel switch +moves the selection synchronously, while the old fallback is replaced only +after the new channel's EPG load runs, so an unscoped merge would leak the +previous channel's programmes into the new panel during (or after a failed) +playback resolution. +Row previews use the same per-channel fallback through the throttled +`StalkerEpgPreviewQueue` once the bulk request has settled (see "Channel row +preview flow"). + +Manually mapped channels never take the portal fallback, on either path: the +component resolves the channel's mapping before falling back +(`applyMappedItvEpg` for the one id, then +`hasItvEpgMappingOverride`) and keeps mapped channels on their mapped +schedule even when it has no currently airing entry — the mapping exists to +replace the portal EPG, so portal data must not be merged back in. ## Manual EPG Mapping diff --git a/libs/portal/stalker/data-access/src/lib/stores/features/with-stalker-epg.feature.spec.ts b/libs/portal/stalker/data-access/src/lib/stores/features/with-stalker-epg.feature.spec.ts index 355f1aefe..bdc3ac141 100644 --- a/libs/portal/stalker/data-access/src/lib/stores/features/with-stalker-epg.feature.spec.ts +++ b/libs/portal/stalker/data-access/src/lib/stores/features/with-stalker-epg.feature.spec.ts @@ -264,6 +264,41 @@ describe('withStalkerEpg', () => { expect(epgBridge.getEpgMappingsBatch).not.toHaveBeenCalled(); }); + + it('reports which channels carry a mapping override', async () => { + epgBridge.getEpgMappingsBatch.mockResolvedValue({ + 'stalker:playlist-1:10001': 'mapped.channel.id', + }); + epgBridge.getChannelPrograms.mockResolvedValue([MAPPED_PROGRAM]); + + expect(store.hasItvEpgMappingOverride('10001')).toBe(false); + + await store.applyMappedItvEpg(['10001', '10002']); + + // Callers use this to keep mapped channels away from the portal + // short-EPG fallback — the mapping replaces the portal schedule. + expect(store.hasItvEpgMappingOverride('10001')).toBe(true); + expect(store.hasItvEpgMappingOverride('10002')).toBe(false); + }); + + it('keeps ownership for a mapping whose mapped guide is currently empty', async () => { + epgBridge.getEpgMappingsBatch.mockResolvedValue({ + 'stalker:playlist-1:10001': 'mapped.channel.id', + }); + epgBridge.getChannelPrograms.mockResolvedValue([]); + const bulkBefore = store.bulkItvEpgByChannel(); + + await store.applyMappedItvEpg(['10001']); + + // The mapping row exists, so the channel is owned even though it + // contributes no programs — the portal fallback must stay out. + expect(store.hasItvEpgMappingOverride('10001')).toBe(true); + // Ownership is published reactively (same content, new map + // reference): a short-EPG fallback that finished before the + // mapping lookup may already have rendered a portal row, and the + // preview effect only reruns — and removes it — on a state patch. + expect(store.bulkItvEpgByChannel()).not.toBe(bulkBefore); + }); }); }); diff --git a/libs/portal/stalker/data-access/src/lib/stores/features/with-stalker-epg.feature.ts b/libs/portal/stalker/data-access/src/lib/stores/features/with-stalker-epg.feature.ts index b4975ac0f..e593d5f14 100644 --- a/libs/portal/stalker/data-access/src/lib/stores/features/with-stalker-epg.feature.ts +++ b/libs/portal/stalker/data-access/src/lib/stores/features/with-stalker-epg.feature.ts @@ -121,11 +121,16 @@ export function withStalkerEpg() { // whenever it replaces the bulk record. const mappingOverridesById = new Map(); const mappingCheckedIds = new Set(); + // Ownership is a fact of the saved mapping row, independent + // of whether the mapped XMLTV guide currently has programs — + // an empty mapped guide must still keep the portal EPG out. + const mappingOwnedIds = new Set(); let mappingPlaylistId: string | null = null; const resetMappingOverrides = (): void => { mappingOverridesById.clear(); mappingCheckedIds.clear(); + mappingOwnedIds.clear(); mappingPlaylistId = null; }; @@ -357,6 +362,7 @@ export function withStalkerEpg() { } let changed = false; + let ownershipChanged = false; for (const [channelId, key] of keyById) { const mappedEpgId = mappings[key]?.trim(); if (!mappedEpgId) { @@ -365,6 +371,16 @@ export function withStalkerEpg() { mappingCheckedIds.add(channelId); continue; } + if (!mappingOwnedIds.has(channelId)) { + mappingOwnedIds.add(channelId); + // Ownership must reach the preview effect + // even when the mapped guide contributes no + // programs: a concurrently fetched short-EPG + // fallback may already have rendered a portal + // row, and only a state patch reruns the sync + // that removes it. + ownershipChanged = true; + } try { const programs = (await epgBridge.getChannelPrograms( @@ -391,10 +407,13 @@ export function withStalkerEpg() { // portal EPG stays in place meanwhile. } } - if (!changed || isStale()) { + if ((!changed && !ownershipChanged) || isStale()) { return; } + // An ownership-only change patches an identical map + // under a new reference — that is deliberate, it is + // what reruns the preview effect. patchState(store, { bulkItvEpgByChannel: { ...store.bulkItvEpgByChannel(), @@ -403,6 +422,25 @@ export function withStalkerEpg() { }); }, + /** + * True when the channel has a saved manual XMLTV mapping + * — even one whose mapped guide currently has no + * programs. Mapped channels must never fall back to the + * portal's short EPG: the mapping exists to replace the + * portal data, and merging the two schedules could + * surface the portal's programme instead. + */ + hasItvEpgMappingOverride( + channelId: string | number + ): boolean { + const normalizedId = + normalizeStalkerEntityId(channelId); + return ( + mappingOwnedIds.has(normalizedId) || + mappingOverridesById.has(normalizedId) + ); + }, + clearBulkItvEpgCache(): void { resetMappingOverrides(); patchState(store, initialEpgState); diff --git a/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-epg-preview.spec.ts b/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-epg-preview.spec.ts new file mode 100644 index 000000000..ecac95bf5 --- /dev/null +++ b/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-epg-preview.spec.ts @@ -0,0 +1,186 @@ +import type { EpgProgram } from '@iptvnator/shared/interfaces'; +import { + StalkerEpgPreviewQueue, + mergeEpgProgramLists, +} from './stalker-live-epg-preview'; + +function buildProgram( + channelId: string, + title: string, + startOffsetMinutes: number, + durationMinutes = 30 +): EpgProgram { + const startTimestamp = Math.floor( + (Date.now() + startOffsetMinutes * 60 * 1000) / 1000 + ); + const stopTimestamp = startTimestamp + durationMinutes * 60; + + return { + start: new Date(startTimestamp * 1000).toISOString(), + stop: new Date(stopTimestamp * 1000).toISOString(), + channel: channelId, + title, + desc: null, + category: null, + startTimestamp, + stopTimestamp, + }; +} + +function flushQueue(ms = 600): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +describe('mergeEpgProgramLists', () => { + it('returns the other list when one side is empty', () => { + const programs = [buildProgram('1', 'Now', -10)]; + + expect(mergeEpgProgramLists(programs, [])).toEqual(programs); + expect(mergeEpgProgramLists([], programs)).toEqual(programs); + }); + + it('fills the missing current programme from the fallback list', () => { + // A bulk guide that only carries future programmes — the reported + // portal shape — merged with a short EPG that starts at "now". + const future = buildProgram('1', 'Later', 120); + const current = buildProgram('1', 'Now', -10); + + const merged = mergeEpgProgramLists([future], [current]); + + expect(merged.map((program) => program.title)).toEqual([ + 'Now', + 'Later', + ]); + }); + + it('keeps the primary entry on an exact start-time collision', () => { + const primary = buildProgram('1', 'Bulk title', -10); + const duplicate = { + ...buildProgram('1', 'Fallback title', -10), + startTimestamp: primary.startTimestamp, + start: primary.start, + }; + + const merged = mergeEpgProgramLists([primary], [duplicate]); + + expect(merged).toHaveLength(1); + expect(merged[0].title).toBe('Bulk title'); + }); +}); + +describe('StalkerEpgPreviewQueue', () => { + it('fetches each synced channel once and reuses the cache afterwards', async () => { + const fetchPrograms = jest.fn(async (channelId: string) => [ + buildProgram(channelId, `Now ${channelId}`, -10), + ]); + const onPrograms = jest.fn(); + const queue = new StalkerEpgPreviewQueue({ fetchPrograms, onPrograms }); + + queue.sync(['1', '2']); + await flushQueue(); + + expect(fetchPrograms).toHaveBeenCalledTimes(2); + expect(onPrograms).toHaveBeenCalledWith('1', [ + expect.objectContaining({ title: 'Now 1' }), + ]); + expect(onPrograms).toHaveBeenCalledWith('2', [ + expect.objectContaining({ title: 'Now 2' }), + ]); + expect(queue.getCachedPrograms('1')).toHaveLength(1); + + queue.sync(['1', '2']); + await flushQueue(300); + + expect(fetchPrograms).toHaveBeenCalledTimes(2); + queue.destroy(); + }); + + it('caches empty results without reporting them', async () => { + const fetchPrograms = jest.fn(async () => [] as EpgProgram[]); + const onPrograms = jest.fn(); + const queue = new StalkerEpgPreviewQueue({ fetchPrograms, onPrograms }); + + queue.sync(['1']); + await flushQueue(300); + queue.sync(['1']); + await flushQueue(300); + + // The portal answered "no EPG" — remembered, not re-asked and not + // surfaced as a preview. + expect(fetchPrograms).toHaveBeenCalledTimes(1); + expect(onPrograms).not.toHaveBeenCalled(); + expect(queue.getCachedPrograms('1')).toEqual([]); + queue.destroy(); + }); + + it('drops channels that were superseded before their fetch started', async () => { + const fetchPrograms = jest.fn(async (channelId: string) => [ + buildProgram(channelId, `Now ${channelId}`, -10), + ]); + const onPrograms = jest.fn(); + const queue = new StalkerEpgPreviewQueue({ fetchPrograms, onPrograms }); + + // '1' starts immediately; '2' and '3' wait behind the throttle + // delay. The second sync (a re-render without '3') must supersede + // the first work list before the throttle releases them. + queue.sync(['1', '2', '3']); + queue.sync(['1', '2']); + await flushQueue(); + + expect(fetchPrograms).toHaveBeenCalledWith('1'); + expect(fetchPrograms).toHaveBeenCalledWith('2'); + expect(fetchPrograms).not.toHaveBeenCalledWith('3'); + queue.destroy(); + }); + + it('caps each sync at the per-sync backlog limit and refills on the next sync', async () => { + const fetchPrograms = jest.fn(async (channelId: string) => [ + buildProgram(channelId, `Now ${channelId}`, -10), + ]); + const queue = new StalkerEpgPreviewQueue( + { fetchPrograms, onPrograms: jest.fn() }, + { delayMs: 0, maxPerSync: 2 } + ); + + // Request count must track user engagement, not render size: only + // the first slice is fetched per sync, the rest waits for the next + // (scroll-driven) sync. + queue.sync(['1', '2', '3', '4']); + await flushQueue(50); + + expect(fetchPrograms).toHaveBeenCalledTimes(2); + expect(fetchPrograms).not.toHaveBeenCalledWith('3'); + + queue.sync(['1', '2', '3', '4']); + await flushQueue(50); + + expect(fetchPrograms).toHaveBeenCalledTimes(4); + expect(fetchPrograms).toHaveBeenCalledWith('3'); + expect(fetchPrograms).toHaveBeenCalledWith('4'); + queue.destroy(); + }); + + it('discards in-flight results after a reset', async () => { + let resolveFetch!: (programs: EpgProgram[]) => void; + const fetchPrograms = jest.fn( + () => + new Promise((resolve) => { + resolveFetch = resolve; + }) + ); + const onPrograms = jest.fn(); + const queue = new StalkerEpgPreviewQueue({ fetchPrograms, onPrograms }); + + queue.sync(['1']); + expect(fetchPrograms).toHaveBeenCalledTimes(1); + + // Portal switch: the pending answer belongs to the old playlist. + queue.reset(); + resolveFetch([buildProgram('1', 'Stale', -10)]); + await flushQueue(50); + + expect(onPrograms).not.toHaveBeenCalled(); + expect(queue.getCachedPrograms('1')).toBeNull(); + queue.destroy(); + }); +}); diff --git a/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-epg-preview.ts b/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-epg-preview.ts new file mode 100644 index 000000000..c0ec5de63 --- /dev/null +++ b/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-epg-preview.ts @@ -0,0 +1,223 @@ +import type { EpgProgram } from '@iptvnator/shared/interfaces'; + +/** + * Per-channel short-EPG fallback for the ITV channel-list previews. + * + * The bulk `get_epg_info` guide is the primary source for the "now playing" + * row previews, but some portals return only future programmes from it (the + * currently airing one is missing) or no usable data at all. `get_short_epg` + * always starts at the current programme, so channels the bulk guide cannot + * answer are fetched individually — throttled and cached, mirroring the + * Xtream `EpgQueueService`, so scrolling a large list cannot flood the + * portal. + */ + +/** Programmes requested per channel: current + a small safety margin. */ +export const EPG_PREVIEW_FETCH_SIZE = 3; + +const PREVIEW_CACHE_TTL_MS = 5 * 60 * 1000; +const PREVIEW_MAX_CONCURRENCY = 2; +const PREVIEW_DELAY_MS = 200; +/** + * Per-sync backlog cap. The list can render 100+ rows at once, and request + * count must track user engagement, not render size: one sync fetches at + * most this many channels (top of the list first — where a freshly opened + * category is scrolled to), and the host re-syncs on scroll to fill the + * next gaps as the user moves through the list. + */ +const PREVIEW_MAX_PER_SYNC = 30; + +interface StalkerEpgPreviewQueueHost { + /** Fetch the short EPG for one channel; resolves [] on failure. */ + fetchPrograms: (channelId: string) => Promise; + /** Called for each non-empty result so the host can update its previews. */ + onPrograms: (channelId: string, programs: EpgProgram[]) => void; +} + +interface StalkerEpgPreviewQueueOptions { + /** Test-only overrides for the throttling constants. */ + delayMs?: number; + maxPerSync?: number; +} + +interface PreviewCacheEntry { + programs: EpgProgram[]; + timestamp: number; +} + +export class StalkerEpgPreviewQueue { + private readonly cache = new Map(); + private readonly inFlight = new Set(); + private queue: string[] = []; + private visibleSet = new Set(); + private processing = false; + /** Bumped by reset() so an in-flight result of the old portal is dropped. */ + private generation = 0; + private destroyed = false; + private readonly delayMs: number; + private readonly maxPerSync: number; + + constructor( + private readonly host: StalkerEpgPreviewQueueHost, + options: StalkerEpgPreviewQueueOptions = {} + ) { + this.delayMs = options.delayMs ?? PREVIEW_DELAY_MS; + this.maxPerSync = options.maxPerSync ?? PREVIEW_MAX_PER_SYNC; + } + + getCachedPrograms(channelId: string): EpgProgram[] | null { + const entry = this.cache.get(channelId); + if (!entry) { + return null; + } + if (Date.now() - entry.timestamp > PREVIEW_CACHE_TTL_MS) { + this.cache.delete(channelId); + return null; + } + return entry.programs; + } + + /** + * Replace the work list with the currently rendered channels that still + * need a preview. Later calls supersede earlier ones, so fast scrolling + * never accumulates stale requests. + */ + sync(channelIds: readonly string[]): void { + if (this.destroyed) { + return; + } + this.visibleSet = new Set(channelIds); + this.queue = channelIds + .filter((id) => this.shouldFetch(id)) + .slice(0, this.maxPerSync); + if (!this.processing && this.queue.length > 0) { + void this.processQueue(); + } + } + + /** Drop all cached data — channel ids are only unique per portal. */ + reset(): void { + this.generation += 1; + this.cache.clear(); + this.inFlight.clear(); + this.queue = []; + this.visibleSet = new Set(); + } + + destroy(): void { + this.destroyed = true; + this.reset(); + } + + private shouldFetch(channelId: string): boolean { + return ( + this.getCachedPrograms(channelId) === null && + !this.inFlight.has(channelId) + ); + } + + private async processQueue(): Promise { + this.processing = true; + try { + while (this.queue.length > 0 && !this.destroyed) { + if (this.inFlight.size >= PREVIEW_MAX_CONCURRENCY) { + await delay(this.delayMs); + continue; + } + + const channelId = this.queue.shift(); + if ( + !channelId || + !this.visibleSet.has(channelId) || + !this.shouldFetch(channelId) + ) { + continue; + } + + this.inFlight.add(channelId); + void this.fetchOne(channelId); + + await delay(this.delayMs); + } + } finally { + this.processing = false; + } + } + + private async fetchOne(channelId: string): Promise { + const generation = this.generation; + try { + const programs = await this.host.fetchPrograms(channelId); + if (this.destroyed || generation !== this.generation) { + return; + } + // Empty results are cached too: they mean the portal has no + // short EPG for the channel, and refetching on every render + // would hammer it for nothing. + this.cache.set(channelId, { programs, timestamp: Date.now() }); + if (programs.length > 0) { + this.host.onPrograms(channelId, programs); + } + } finally { + if (generation === this.generation) { + this.inFlight.delete(channelId); + } + } + } +} + +/** + * Merge the bulk-EPG programme list with the short-EPG fallback for the + * active-channel panel. The bulk guide may cover days ahead yet miss the + * currently airing programme; the short EPG starts at "now" but only spans a + * few entries. Primary entries win on an exact start-time collision. + */ +export function mergeEpgProgramLists( + primary: EpgProgram[], + secondary: EpgProgram[] +): EpgProgram[] { + if (secondary.length === 0) { + return primary; + } + if (primary.length === 0) { + return secondary; + } + + const primaryStarts = new Set(); + for (const program of primary) { + const startMs = getEpgProgramStartMs(program); + if (startMs !== null) { + primaryStarts.add(startMs); + } + } + + const merged = [...primary]; + for (const program of secondary) { + const startMs = getEpgProgramStartMs(program); + if (startMs === null || !primaryStarts.has(startMs)) { + merged.push(program); + } + } + + return merged.sort( + (left, right) => + (getEpgProgramStartMs(left) ?? 0) - + (getEpgProgramStartMs(right) ?? 0) + ); +} + +function getEpgProgramStartMs(program: EpgProgram): number | null { + if ( + Number.isFinite(program.startTimestamp) && + Number(program.startTimestamp) > 0 + ) { + return Number(program.startTimestamp) * 1000; + } + + const parsedDate = Date.parse(program.start); + return Number.isFinite(parsedDate) ? parsedDate : null; +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} diff --git a/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-stream-layout.component.spec.ts b/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-stream-layout.component.spec.ts index 3b0034850..8aee00df1 100644 --- a/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-stream-layout.component.spec.ts +++ b/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-stream-layout.component.spec.ts @@ -32,6 +32,7 @@ import { SettingsStore, } from '@iptvnator/services'; import { + EpgItem, EpgProgram, ResolvedPortalPlayback, } from '@iptvnator/shared/interfaces'; @@ -253,6 +254,7 @@ describe('StalkerLiveStreamLayoutComponent', () => { fetchChannelEpg: jest.fn(), ensureBulkItvEpg: jest.fn(), applyMappedItvEpg: jest.fn().mockResolvedValue(undefined), + hasItvEpgMappingOverride: jest.fn(() => false), clearBulkItvEpgCache: jest.fn(() => { bulkItvEpgByChannel.set({}); bulkItvEpgLoaded.set(false); @@ -336,19 +338,12 @@ describe('StalkerLiveStreamLayoutComponent', () => { portalPlayer.openExternalPlayback.mockClear(); fetchChannelEpg.mockReset(); fetchChannelEpg.mockResolvedValue([]); + // mockReset drops the implementation; undefined reads as "unmapped". + stalkerStore.hasItvEpgMappingOverride.mockReset(); ensureBulkItvEpg.mockReset(); - ensureBulkItvEpg.mockImplementation(async () => { - const bulkPrograms = { - '10001': [buildProgram('10001', 'Current Show')], - '10002': [buildProgram('10002', 'Next Channel Show')], - }; - bulkItvEpgByChannel.set(bulkPrograms); - bulkItvEpgLoaded.set(true); - bulkItvEpgPlaylistId.set('playlist-1'); - bulkItvEpgPeriodHours.set(168); - selectedItvEpgPrograms.set( - bulkPrograms[selectedItvId() ?? ''] ?? [] - ); + mockBulkEpg({ + '10001': [buildProgram('10001', 'Current Show')], + '10002': [buildProgram('10002', 'Next Channel Show')], }); stalkerStore.setItvChannels.mockClear(); stalkerStore.setRadioChannels.mockClear(); @@ -1129,6 +1124,94 @@ describe('StalkerLiveStreamLayoutComponent', () => { expect(fetchChannelEpg).not.toHaveBeenCalled(); }); + function mockBulkEpg(bulkPrograms: Record): void { + ensureBulkItvEpg.mockImplementation(async () => { + bulkItvEpgByChannel.set(bulkPrograms); + bulkItvEpgLoaded.set(true); + bulkItvEpgPlaylistId.set('playlist-1'); + bulkItvEpgPeriodHours.set(168); + selectedItvEpgPrograms.set( + bulkPrograms[selectedItvId() ?? ''] ?? [] + ); + }); + } + + /** + * The reported portal shape: bulk get_epg_info returns only programmes + * that start in the future — the currently airing one is missing — while + * get_short_epg answers with the current programme. + */ + function mockFutureOnlyBulkEpg(): void { + mockBulkEpg({ + '10001': [buildFutureProgram('10001', 'Future Show')], + '10002': [buildFutureProgram('10002', 'Future Beta Show')], + }); + fetchChannelEpg.mockImplementation(async (channelId: string) => [ + buildEpgItem(String(channelId), `Now ${channelId}`), + ]); + } + + it('merges the short-EPG fallback into the panel when bulk EPG has only future programmes', async () => { + // The old either/or gate skipped the short-EPG fallback whenever bulk + // was non-empty, leaving the panel without a current programme. + mockFutureOnlyBulkEpg(); + + fixture.detectChanges(); + await component.playChannel(itvChannels()[0]); + await fixture.whenStable(); + fixture.detectChanges(); + + expect(fetchChannelEpg).toHaveBeenCalledWith('10001'); + expect(component.currentProgram()?.title).toBe('Now 10001'); + expect( + component.activeEpgPrograms().map((program) => program.title) + ).toEqual(['Now 10001', 'Future Show']); + }); + + it('fills row previews from the short EPG when bulk EPG misses the current programmes', async () => { + mockFutureOnlyBulkEpg(); + + await settleEagerEpg(); + // The throttled per-channel fallback queue drains the two rows + // (one request immediately, the next behind a 200 ms delay). + await new Promise((resolve) => setTimeout(resolve, 600)); + fixture.detectChanges(); + + expect( + ['10001', '10002'].map( + (id) => component.epgPreviewPrograms.get(id)?.title + ) + ).toEqual(['Now 10001', 'Now 10002']); + }); + + it('keeps manually mapped channels away from the portal short-EPG fallback', async () => { + // A mapping replaces the portal schedule; merging the portal's short + // EPG back in could surface the portal's programme instead. + mockFutureOnlyBulkEpg(); + stalkerStore.hasItvEpgMappingOverride.mockImplementation( + (id: string | number) => String(id) === '10001' + ); + + fixture.detectChanges(); + await component.playChannel(itvChannels()[0]); + await fixture.whenStable(); + await new Promise((resolve) => setTimeout(resolve, 600)); + fixture.detectChanges(); + + // Neither the panel nor the row queue asked the portal for 10001. + expect( + fetchChannelEpg.mock.calls.map(([id]: [unknown]) => String(id)) + ).not.toContain('10001'); + // Panel keeps the mapped (future-only) schedule: no "on now" entry. + expect(component.currentProgram()).toBeNull(); + // The unmapped sibling still gets the row fallback; the mapped + // channel's row stays on its (empty) mapped schedule. + expect(component.epgPreviewPrograms.get('10001')).toBeUndefined(); + expect(component.epgPreviewPrograms.get('10002')?.title).toBe( + 'Now 10002' + ); + }); + it('does not re-fetch bulk EPG when switching channels once it is loaded', async () => { await settleEagerEpg(); // Bulk EPG has loaded (eagerly, on entry). @@ -1291,8 +1374,14 @@ describe('StalkerLiveStreamLayoutComponent', () => { }); }); -function buildProgram(channelId: string, title: string): EpgProgram { - const startTimestamp = Math.floor((Date.now() - 10 * 60 * 1000) / 1000); +function buildProgram( + channelId: string, + title: string, + // Started 10 minutes ago (currently airing) unless shifted. + startOffsetMinutes = -10 +): EpgProgram { + const startTimestamp = + Math.floor(Date.now() / 1000) + startOffsetMinutes * 60; const stopTimestamp = startTimestamp + 30 * 60; return { @@ -1306,3 +1395,26 @@ function buildProgram(channelId: string, title: string): EpgProgram { stopTimestamp, }; } + +/** A programme that starts two hours from now — nothing airing "now". */ +function buildFutureProgram(channelId: string, title: string): EpgProgram { + return buildProgram(channelId, title, 120); +} + +/** A currently airing short-EPG entry in the store's EpgItem shape. */ +function buildEpgItem(channelId: string, title: string): EpgItem { + const program = buildProgram(channelId, title); + return { + id: `${channelId}-${title}`, + epg_id: '', + title, + lang: '', + start: program.start, + end: program.stop, + stop: program.stop, + description: `${title} description`, + channel_id: channelId, + start_timestamp: String(program.startTimestamp), + stop_timestamp: String(program.stopTimestamp), + }; +} diff --git a/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-stream-layout.component.ts b/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-stream-layout.component.ts index 3b795996c..ac570060d 100644 --- a/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-stream-layout.component.ts +++ b/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-stream-layout.component.ts @@ -76,6 +76,11 @@ import { normalizeStalkerEntityId, } from '@iptvnator/portal/stalker/data-access'; import { StalkerItvAllItemsComponent } from './stalker-itv-all-items.component'; +import { + EPG_PREVIEW_FETCH_SIZE, + StalkerEpgPreviewQueue, + mergeEpgProgramLists, +} from './stalker-live-epg-preview'; import { createPlaybackSessionKey } from '@iptvnator/playback/util'; type StalkerPlayableChannel = StalkerPortalItem & { @@ -307,13 +312,32 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { ); /** EPG */ - readonly fallbackEpgPrograms = signal([]); + /** Short-EPG panel fallback, tagged with the channel it was fetched for. */ + readonly fallbackEpgPrograms = signal<{ + channelId: string; + programs: EpgProgram[]; + } | null>(null); readonly isLoadingFallbackEpg = signal(false); + // Merged, not either/or: some portals' bulk get_epg_info carries only + // future programmes, so a non-empty bulk list can still miss the one + // airing now — the short-EPG fallback fills exactly that gap. The merge + // is scoped to the fallback's own channel: a channel switch moves the + // selection synchronously while the old fallback is only replaced once + // the new channel's EPG load runs, and an unscoped merge would mix the + // previous channel's programmes into the new panel meanwhile. readonly activeEpgPrograms = computed(() => { - const bulkPrograms = this.stalkerStore.selectedItvEpgPrograms(); - return bulkPrograms.length > 0 - ? bulkPrograms - : this.fallbackEpgPrograms(); + const fallback = this.fallbackEpgPrograms(); + const selectedId = this.selectedChannelId(); + const fallbackPrograms = + fallback && + selectedId && + fallback.channelId === normalizeStalkerEntityId(selectedId) + ? fallback.programs + : []; + return mergeEpgProgramLists( + this.stalkerStore.selectedItvEpgPrograms(), + fallbackPrograms + ); }); readonly currentProgram = computed(() => this.findCurrentProgram(this.activeEpgPrograms()) @@ -372,6 +396,18 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { readonly epgPreviewPrograms = new Map(); readonly currentProgramsProgress = new Map(); private readonly cdr = inject(ChangeDetectorRef); + /** Short-EPG fallback for rows the bulk guide cannot answer. */ + private readonly epgPreviewQueue = new StalkerEpgPreviewQueue({ + fetchPrograms: async (channelId) => + ( + await this.stalkerStore.fetchChannelEpg( + channelId, + EPG_PREVIEW_FETCH_SIZE + ) + ).map((item) => this.toProgram(item, channelId)), + onPrograms: (channelId, programs) => + this.applyFallbackPreviewPrograms(channelId, programs), + }); /** Favorites */ readonly favorites = new Map(); @@ -379,6 +415,8 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { /** Scroll */ readonly scrollContainer = viewChild('scrollContainer'); private scrollListener: (() => void) | null = null; + private epgPreviewRefreshTimer: ReturnType | null = + null; private unsubscribeRemoteChannelChange?: () => void; private unsubscribeRemoteCommand?: () => void; private epgLoadRequestId = 0; @@ -440,7 +478,7 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { this.stalkerStore.setPage(0); this.clearEpgPreviewMaps(); this.epgLoadRequestId += 1; - this.fallbackEpgPrograms.set([]); + this.fallbackEpgPrograms.set(null); this.isLoadingFallbackEpg.set(false); }); }); @@ -457,6 +495,9 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { if (this.isRadioMode() || !this.supportsEpg) { this.clearEpgPreviewMaps(); + // Supersede the queue too — an abandoned ITV view must not + // keep spending portal requests on rows that are gone. + this.epgPreviewQueue.sync([]); this.cdr.markForCheck(); return; } @@ -481,8 +522,11 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { this.lastPlaylistId = playlistId; this.epgLoadRequestId += 1; - this.fallbackEpgPrograms.set([]); + this.fallbackEpgPrograms.set(null); this.isLoadingFallbackEpg.set(false); + // Channel ids are only unique per portal — cached previews of the + // previous playlist must not leak into the new one. + this.epgPreviewQueue.reset(); this.stalkerStore.clearBulkItvEpgCache(); }); @@ -579,6 +623,11 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { this.unsubscribeRemoteChannelChange?.(); this.unsubscribeRemoteCommand?.(); this.removeScrollListener(); + if (this.epgPreviewRefreshTimer !== null) { + clearTimeout(this.epgPreviewRefreshTimer); + this.epgPreviewRefreshTimer = null; + } + this.epgPreviewQueue.destroy(); // Invalidate any playback continuation still awaiting its header // IPC, then drop the radio credentials — they must not outlive this // layout. The service no-ops when a newer playback already owns the @@ -930,7 +979,7 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { private async loadEpgForChannel(item: StalkerItvChannel) { if (!this.supportsEpg) { - this.fallbackEpgPrograms.set([]); + this.fallbackEpgPrograms.set(null); this.isLoadingFallbackEpg.set(false); this.clearEpgPreviewMaps(); return; @@ -944,7 +993,7 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { this.stalkerStore.bulkItvEpgPlaylistId() !== playlistId || this.stalkerStore.bulkItvEpgPeriodHours() !== 168; - this.fallbackEpgPrograms.set([]); + this.fallbackEpgPrograms.set(null); this.isLoadingFallbackEpg.set(false); try { @@ -955,7 +1004,36 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { } } - if (this.stalkerStore.selectedItvEpgPrograms().length > 0) { + // Skip the short-EPG fallback only when the bulk guide can + // actually answer "what's on now" — a non-empty bulk list of + // future-only programmes still needs the fallback merged in. + if ( + this.findCurrentProgram( + this.stalkerStore.selectedItvEpgPrograms() + ) + ) { + return; + } + + // Resolve this channel's manual mapping before falling back: a + // mapped channel must show the mapped XMLTV schedule only — + // merging the portal's short EPG in could surface the portal's + // programme, defeating the mapping the user created to replace + // it. The store dedupes per channel id, so this is cheap. + await this.stalkerStore.applyMappedItvEpg([item.id]); + if (!this.isCurrentEpgRequest(requestId, normalizedChannelId)) { + return; + } + if ( + this.findCurrentProgram( + this.stalkerStore.selectedItvEpgPrograms() + ) + ) { + return; + } + if ( + this.stalkerStore.hasItvEpgMappingOverride(normalizedChannelId) + ) { return; } @@ -967,15 +1045,16 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { return; } - this.fallbackEpgPrograms.set( - fallbackItems.map((epgItem) => + this.fallbackEpgPrograms.set({ + channelId: normalizedChannelId, + programs: fallbackItems.map((epgItem) => this.toProgram(epgItem, normalizedChannelId) - ) - ); + ), + }); } catch (error) { this.logger.warn('Failed to load Stalker live EPG', error); if (this.isCurrentEpgRequest(requestId, normalizedChannelId)) { - this.fallbackEpgPrograms.set([]); + this.fallbackEpgPrograms.set(null); } } finally { if (this.isCurrentEpgRequest(requestId, normalizedChannelId)) { @@ -993,21 +1072,39 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { this.clearEpgPreviewMaps(); const bulkProgramsByChannel = this.stalkerStore.bulkItvEpgByChannel(); - if ( - channels.length === 0 || - Object.keys(bulkProgramsByChannel).length === 0 - ) { + if (channels.length === 0) { + // A legacy-paged category switch clears the list before the new + // channels arrive — supersede the backlog so the disappeared + // rows stop consuming portal request capacity. + this.epgPreviewQueue.sync([]); this.cdr.markForCheck(); return; } + const channelsWithoutCurrent: string[] = []; for (const channel of channels) { const channelId = normalizeStalkerEntityId(channel.id); - const currentProgram = this.findCurrentProgram( - bulkProgramsByChannel[channelId] ?? [] - ); + // Manually mapped channels are bulk-only: their programs in the + // bulk record come from the uploaded XMLTV guide, and the portal + // short EPG must not stand in for the schedule the mapping + // deliberately replaces. + const hasMappingOverride = + this.stalkerStore.hasItvEpgMappingOverride(channelId); + const currentProgram = + this.findCurrentProgram( + bulkProgramsByChannel[channelId] ?? [] + ) ?? + (hasMappingOverride + ? null + : this.findCurrentProgram( + this.epgPreviewQueue.getCachedPrograms(channelId) ?? + [] + )); if (!currentProgram) { + if (!hasMappingOverride) { + channelsWithoutCurrent.push(channelId); + } continue; } @@ -1015,6 +1112,45 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { this.updateProgramProgress(channelId, currentProgram); } + // Rows the bulk guide cannot answer (portals whose get_epg_info + // returns only future programmes, or none at all) fall back to + // per-channel short EPG. Deferred until the bulk request settles so + // the queue never races the answer it is a fallback for; the effect + // tracks bulkItvEpgLoaded, so it re-runs when that happens. + if (this.stalkerStore.bulkItvEpgLoaded()) { + this.epgPreviewQueue.sync(channelsWithoutCurrent); + } + + this.cdr.markForCheck(); + } + + private applyFallbackPreviewPrograms( + channelId: string, + programs: EpgProgram[] + ): void { + if (this.isRadioMode() || !this.supportsEpg) { + return; + } + + // Revalidate ownership: the fetch was enqueued before mapping + // resolution (or a bulk refresh) could finish, and an owner installed + // in the meantime must not be overwritten by a late portal response. + if ( + this.stalkerStore.hasItvEpgMappingOverride(channelId) || + this.findCurrentProgram( + this.stalkerStore.bulkItvEpgByChannel()[channelId] ?? [] + ) + ) { + return; + } + + const currentProgram = this.findCurrentProgram(programs); + if (!currentProgram) { + return; + } + + this.epgPreviewPrograms.set(channelId, currentProgram); + this.updateProgramProgress(channelId, currentProgram); this.cdr.markForCheck(); } @@ -1063,6 +1199,7 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { if (!container) return; const onScroll = () => { + this.scheduleEpgPreviewRefresh(); if (this.isLoadingMore() || !this.hasMoreItems()) return; const { scrollTop, scrollHeight, clientHeight } = container; @@ -1096,6 +1233,25 @@ export class StalkerLiveStreamLayoutComponent implements OnDestroy { } } + /** + * The preview queue caps each sync's backlog so request count tracks + * user engagement, not render size — scrolling therefore re-syncs to + * fetch the next rows the user is moving toward. Throttled; a fully + * cached list makes the re-sync a no-op. + */ + private scheduleEpgPreviewRefresh(): void { + if (this.epgPreviewRefreshTimer !== null) { + return; + } + this.epgPreviewRefreshTimer = setTimeout(() => { + this.epgPreviewRefreshTimer = null; + if (this.isRadioMode() || !this.supportsEpg) { + return; + } + this.syncBulkEpgPreviews(this.visibleChannels()); + }, 300); + } + private toProgram(item: EpgItem, channelId: string | number): EpgProgram { return { start: item.start, diff --git a/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-stream-layout.epg-races.spec.ts b/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-stream-layout.epg-races.spec.ts new file mode 100644 index 000000000..cbcf3b653 --- /dev/null +++ b/libs/portal/stalker/feature/src/lib/stalker-live-stream-layout/stalker-live-stream-layout.epg-races.spec.ts @@ -0,0 +1,301 @@ +import { signal } from '@angular/core'; +import { ComponentFixture, TestBed } from '@angular/core/testing'; +import { MatDialog } from '@angular/material/dialog'; +import { MatSnackBar } from '@angular/material/snack-bar'; +import { TranslateService } from '@ngx-translate/core'; +import { of } from 'rxjs'; +import { EpgRuntimeBridgeService } from '@iptvnator/epg/data-access'; +import { + LiveLayoutSidebarStateService, + PORTAL_PLAYER, +} from '@iptvnator/portal/shared/util'; +import { StalkerStore } from '@iptvnator/portal/stalker/data-access'; +import { + PlaylistsService, + RuntimeCapabilitiesService, + SettingsStore, +} from '@iptvnator/services'; +import type { EpgItem, EpgProgram } from '@iptvnator/shared/interfaces'; +import { ElectronStreamHeadersService } from '@iptvnator/ui/playback'; +import { StalkerLiveStreamLayoutComponent } from './stalker-live-stream-layout.component'; + +/** + * Race regressions for the short-EPG fallback: + * + * 1. The panel fallback is scoped to the channel it was fetched for — a + * channel switch moves the selection synchronously, while the previous + * channel's fallback is only replaced once the new channel's EPG load + * runs. An unscoped merge mixed channel A's programmes into channel B's + * panel during (or after a failed) playback resolution. + * 2. A queued row-preview fetch can complete after a manual XMLTV mapping + * (or a bulk refresh) claimed the row; the late portal response must not + * overwrite the installed owner. + */ +describe('StalkerLiveStreamLayoutComponent EPG fallback races', () => { + let fixture: ComponentFixture; + let component: StalkerLiveStreamLayoutComponent; + const playlist = signal({ _id: 'playlist-one', title: 'Portal One' }); + const channels = [ + { + id: 'channel-one', + cmd: 'ffrt4://itv/channel-one', + name: 'One', + o_name: 'One', + logo: 'one.png', + }, + { + id: 'channel-two', + cmd: 'ffrt4://itv/channel-two', + name: 'Two', + o_name: 'Two', + logo: 'two.png', + }, + ]; + const itvChannels = signal(channels); + const selectedItvId = signal(channels[0].id); + const selectedItem = signal<(typeof channels)[number] | null>(channels[0]); + const selectedContentType = signal<'itv' | 'radio'>('itv'); + const selectedItvEpgPrograms = signal([]); + const bulkItvEpgByChannel = signal>({}); + const bulkItvEpgLoaded = signal(false); + const resolveItvPlayback = jest.fn(); + const fetchChannelEpg = jest.fn(); + const hasItvEpgMappingOverride = jest.fn(() => false); + const store = { + getSelectedCategoryName: signal('All'), + currentPlaylist: playlist, + selectedContentType, + selectedCategoryId: signal('all'), + selectedItvId, + selectedItem, + itvChannels, + radioChannels: signal([]), + searchPhrase: signal(''), + hasMoreChannels: signal(false), + itvFullListActive: signal(false), + itvSelectedCategoryFromCache: signal(false), + itvFullListLoading: signal(false), + itvFullListProgress: signal(null), + itvFullChannelList: signal([]), + isPaginatedContentLoading: signal(false), + selectedItvEpgPrograms, + bulkItvEpgByChannel, + bulkItvEpgLoaded, + bulkItvEpgPlaylistId: signal('playlist-one'), + bulkItvEpgPeriodHours: signal(168), + isLoadingBulkItvEpg: signal(false), + setItvChannels: jest.fn(), + setRadioChannels: jest.fn(), + setPage: jest.fn(), + preloadItvChannels: jest.fn(), + applyMappedItvEpg: jest.fn().mockResolvedValue(undefined), + hasItvEpgMappingOverride, + clearBulkItvEpgCache: jest.fn(), + ensureBulkItvEpg: jest.fn().mockResolvedValue(undefined), + fetchChannelEpg, + resolveItvPlayback, + resolveRadioPlayback: jest.fn(), + addToFavorites: jest.fn(), + removeFromFavorites: jest.fn(), + setSelectedItem: jest.fn((item: (typeof channels)[number]) => { + selectedItem.set(item); + selectedItvId.set(String(item.id)); + selectedItvEpgPrograms.set( + bulkItvEpgByChannel()[String(item.id)] ?? [] + ); + }), + }; + + beforeEach(async () => { + playlist.set({ _id: 'playlist-one', title: 'Portal One' }); + selectedContentType.set('itv'); + selectedItvId.set(channels[0].id); + selectedItem.set(channels[0]); + bulkItvEpgByChannel.set({ + 'channel-one': [buildProgram('channel-one', 'Future A', 120)], + 'channel-two': [buildProgram('channel-two', 'Future B', 120)], + }); + bulkItvEpgLoaded.set(true); + selectedItvEpgPrograms.set(bulkItvEpgByChannel()['channel-one']); + resolveItvPlayback.mockReset(); + resolveItvPlayback.mockResolvedValue({ + streamUrl: 'https://one.example/live.m3u8', + }); + fetchChannelEpg.mockReset(); + fetchChannelEpg.mockImplementation( + async (channelId: string | number) => [ + buildEpgItem(String(channelId), `Now ${channelId}`), + ] + ); + hasItvEpgMappingOverride.mockReset(); + hasItvEpgMappingOverride.mockReturnValue(false); + await TestBed.configureTestingModule({ + imports: [StalkerLiveStreamLayoutComponent], + providers: [ + { provide: StalkerStore, useValue: store }, + { + provide: RuntimeCapabilitiesService, + useValue: { + supportsEpg: true, + isElectron: true, + supportsEpgMapping: false, + }, + }, + { + provide: PlaylistsService, + useValue: { getPortalFavorites: () => of([]) }, + }, + { + provide: SettingsStore, + useValue: { openStreamOnDoubleClick: signal(false) }, + }, + { + provide: PORTAL_PLAYER, + useValue: { + isEmbeddedPlayer: () => true, + openResolvedPlayback: jest.fn(), + }, + }, + { + provide: ElectronStreamHeadersService, + useValue: { apply: jest.fn(), clear: jest.fn() }, + }, + { + provide: LiveLayoutSidebarStateService, + useValue: { isCollapsed: signal(false), toggle: jest.fn() }, + }, + { provide: EpgRuntimeBridgeService, useValue: {} }, + { provide: MatDialog, useValue: { open: jest.fn() } }, + { provide: MatSnackBar, useValue: { open: jest.fn() } }, + { + provide: TranslateService, + useValue: { instant: (key: string) => key }, + }, + ], + }) + .overrideComponent(StalkerLiveStreamLayoutComponent, { + set: { template: '' }, + }) + .compileComponents(); + fixture = TestBed.createComponent(StalkerLiveStreamLayoutComponent); + component = fixture.componentInstance; + fixture.detectChanges(); + }); + + afterEach(() => fixture.destroy()); + + it('drops the previous channel fallback from the panel as soon as the selection moves', async () => { + await component.playChannel(channels[0]); + await fixture.whenStable(); + + expect( + component.activeEpgPrograms().map((program) => program.title) + ).toEqual(['Now channel-one', 'Future A']); + + // A channel switch moves the selection synchronously; the EPG load + // that replaces the fallback only runs after (slow or failing) + // playback resolution. The stale fallback must not leak into B. + store.setSelectedItem(channels[1]); + + expect( + component.activeEpgPrograms().map((program) => program.title) + ).toEqual(['Future B']); + }); + + it('stops the preview backlog when the view leaves ITV', async () => { + // Let init settle: the playlist effect's first run resets the queue, + // discarding whatever the init sync dispatched. + await fixture.whenStable(); + await new Promise((resolve) => setTimeout(resolve, 250)); + fetchChannelEpg.mockClear(); + + // Re-arm the backlog: a bulk-loaded transition re-runs the preview + // sync, which enqueues both future-only channels and dispatches the + // first one immediately. + bulkItvEpgLoaded.set(false); + fixture.detectChanges(); + bulkItvEpgLoaded.set(true); + fixture.detectChanges(); + const callsAtSwitch = fetchChannelEpg.mock.calls.length; + expect(callsAtSwitch).toBeGreaterThan(0); + + // Leaving ITV must supersede the rest of the backlog — an abandoned + // view must not keep spending portal requests on vanished rows. + selectedContentType.set('radio'); + fixture.detectChanges(); + await new Promise((resolve) => setTimeout(resolve, 600)); + + expect(fetchChannelEpg.mock.calls.length).toBe(callsAtSwitch); + }); + + it('does not let a late queued fallback overwrite an installed owner', () => { + const apply = (channelId: string) => + ( + component as unknown as { + applyFallbackPreviewPrograms( + id: string, + programs: EpgProgram[] + ): void; + } + ).applyFallbackPreviewPrograms(channelId, [ + buildProgram(channelId, `Portal ${channelId}`, -10), + ]); + + // Mapping override installed while the fetch was in flight. + hasItvEpgMappingOverride.mockReturnValue(true); + apply('channel-one'); + expect(component.epgPreviewPrograms.get('channel-one')).toBeUndefined(); + + // Bulk data claimed the row while the fetch was in flight. + hasItvEpgMappingOverride.mockReturnValue(false); + bulkItvEpgByChannel.set({ + 'channel-two': [buildProgram('channel-two', 'Bulk Now', -10)], + }); + apply('channel-two'); + expect(component.epgPreviewPrograms.get('channel-two')).toBeUndefined(); + + // No owner — the fallback applies. + apply('channel-one'); + expect(component.epgPreviewPrograms.get('channel-one')?.title).toBe( + 'Portal channel-one' + ); + }); +}); + +function buildProgram( + channelId: string, + title: string, + startOffsetMinutes: number +): EpgProgram { + const startTimestamp = + Math.floor(Date.now() / 1000) + startOffsetMinutes * 60; + const stopTimestamp = startTimestamp + 30 * 60; + + return { + start: new Date(startTimestamp * 1000).toISOString(), + stop: new Date(stopTimestamp * 1000).toISOString(), + channel: channelId, + title, + desc: null, + category: null, + startTimestamp, + stopTimestamp, + }; +} + +function buildEpgItem(channelId: string, title: string): EpgItem { + const program = buildProgram(channelId, title, -10); + return { + id: `${channelId}-${title}`, + epg_id: '', + title, + lang: '', + start: program.start, + end: program.stop, + stop: program.stop, + description: `${title} description`, + channel_id: channelId, + start_timestamp: String(program.startTimestamp), + stop_timestamp: String(program.stopTimestamp), + }; +}