From 649d55a1c1b83554e081658242d2a3c359c950e0 Mon Sep 17 00:00:00 2001 From: 4gray Date: Wed, 11 Feb 2026 23:11:14 +0100 Subject: [PATCH] refactor(epg): implement throttled EPG request queue with caching and concurrency control Entire-Checkpoint: 3ba3a7503b59 --- .../portal-channels-list.component.ts | 122 ++++++++++------ .../services/epg-queue.service.ts | 134 ++++++++++++++++++ 2 files changed, 210 insertions(+), 46 deletions(-) create mode 100644 apps/web/src/app/xtream-tauri/services/epg-queue.service.ts diff --git a/apps/web/src/app/xtream-tauri/portal-channels-list/portal-channels-list.component.ts b/apps/web/src/app/xtream-tauri/portal-channels-list/portal-channels-list.component.ts index ecbd28058..cee27954f 100644 --- a/apps/web/src/app/xtream-tauri/portal-channels-list/portal-channels-list.component.ts +++ b/apps/web/src/app/xtream-tauri/portal-channels-list/portal-channels-list.component.ts @@ -7,11 +7,11 @@ import { AfterViewInit, ChangeDetectorRef, Component, - EventEmitter, inject, - Output, + OnDestroy, + output, signal, - ViewChild, + viewChild, } from '@angular/core'; import { FormsModule } from '@angular/forms'; import { MatFormFieldModule } from '@angular/material/form-field'; @@ -20,7 +20,11 @@ import { MatInputModule } from '@angular/material/input'; import { ActivatedRoute } from '@angular/router'; import { FilterPipe } from '@iptvnator/pipes'; import { TranslatePipe } from '@ngx-translate/core'; +import { Subscription } from 'rxjs'; +import { debounceTime } from 'rxjs/operators'; import { XtreamCategory, XtreamItem } from 'shared-interfaces'; +import { EpgQueueService } from '../services/epg-queue.service'; +import { XtreamCredentials } from '../services/xtream-api.service'; import { FavoritesService } from '../services/favorites.service'; import { XtreamStore } from '../stores/xtream.store'; @@ -48,11 +52,12 @@ interface EpgProgram { TranslatePipe, ], }) -export class PortalChannelsListComponent implements AfterViewInit { - @Output() playClicked = new EventEmitter(); +export class PortalChannelsListComponent implements AfterViewInit, OnDestroy { + readonly playClicked = output(); readonly xtreamStore = inject(XtreamStore); private readonly favoritesService = inject(FavoritesService); + private readonly epgQueueService = inject(EpgQueueService); private readonly route = inject(ActivatedRoute); readonly channels = this.xtreamStore.selectItemsFromSelectedCategory; @@ -60,10 +65,11 @@ export class PortalChannelsListComponent implements AfterViewInit { searchString = signal(''); currentPrograms = new Map(); currentProgramsProgress = new Map(); - programTimings = new Map(); // Changed to store timestamps - private requestedChannels = new Set(); + programTimings = new Map(); - @ViewChild(CdkVirtualScrollViewport) viewport?: CdkVirtualScrollViewport; + readonly viewport = viewChild(CdkVirtualScrollViewport); + + private subscriptions = new Subscription(); constructor(private cdr: ChangeDetectorRef) {} @@ -86,58 +92,78 @@ export class PortalChannelsListComponent implements AfterViewInit { }); }); } + + // Subscribe to EPG results from the queue service + this.subscriptions.add( + this.epgQueueService.epgResult$.subscribe( + ({ streamId, items }) => { + if (items && items.length > 0) { + this.currentPrograms.set(streamId, items[0].title); + this.updateProgramProgress( + streamId, + items[0] as unknown as EpgProgram + ); + this.cdr.detectChanges(); + } + } + ) + ); } ngAfterViewInit() { - if ( - this.viewport && - this.xtreamStore.selectedContentType() === 'live' - ) { - this.viewport.renderedRangeStream.subscribe((range) => { - const visibleChannels = this.channels().slice( - range.start, - range.end - ); - this.loadEpgForVisibleChannels(visibleChannels); - }); + const vp = this.viewport(); + if (vp && this.xtreamStore.selectedContentType() === 'live') { + this.subscriptions.add( + vp.renderedRangeStream + .pipe(debounceTime(300)) + .subscribe((range) => { + const visibleChannels = this.channels().slice( + range.start, + range.end + ); + this.loadEpgForVisibleChannels(visibleChannels); + }) + ); } } - private async loadEpgForVisibleChannels(channels: any[]): Promise { + private loadEpgForVisibleChannels(channels: any[]): void { const playlist = this.xtreamStore.currentPlaylist(); if (!playlist) return; + const credentials: XtreamCredentials = { + serverUrl: playlist.serverUrl, + username: playlist.username, + password: playlist.password, + }; + + const visibleIds = new Set( + channels.map((ch) => ch.xtream_id) + ); + const uncachedIds: number[] = []; + + // Apply cached results immediately for (const channel of channels) { - // Skip if we already requested or have EPG data for this channel - if ( - this.requestedChannels.has(channel.xtream_id) || - this.currentPrograms.has(channel.xtream_id) - ) { - continue; - } - - // Mark as requested before making the API call - this.requestedChannels.add(channel.xtream_id); - - try { - const epgData = await this.xtreamStore.loadChannelEpg( - channel.xtream_id - ); - if (epgData && epgData.length > 0) { + const cached = this.epgQueueService.getCached(channel.xtream_id); + if (cached && cached.length > 0) { + if (!this.currentPrograms.has(channel.xtream_id)) { this.currentPrograms.set( channel.xtream_id, - epgData[0].title + cached[0].title + ); + this.updateProgramProgress( + channel.xtream_id, + cached[0] as unknown as EpgProgram ); - this.updateProgramProgress(channel.xtream_id, epgData[0]); - this.cdr.detectChanges(); } - } catch (error) { - console.error( - `Failed to load EPG for channel ${channel.xtream_id}:`, - error - ); + } else if (!this.currentPrograms.has(channel.xtream_id)) { + uncachedIds.push(channel.xtream_id); } } + + if (uncachedIds.length > 0) { + this.epgQueueService.enqueue(uncachedIds, visibleIds, credentials); + } } private updateProgramProgress(streamId: number, program: EpgProgram) { @@ -152,8 +178,8 @@ export class PortalChannelsListComponent implements AfterViewInit { this.currentProgramsProgress.set(streamId, progress); this.programTimings.set(streamId, { - start: start * 1000, // Convert to milliseconds for date pipe - end: end * 1000, // Convert to milliseconds for date pipe + start: start * 1000, + end: end * 1000, }); } } @@ -180,4 +206,8 @@ export class PortalChannelsListComponent implements AfterViewInit { this.cdr.detectChanges(); }); } + + ngOnDestroy(): void { + this.subscriptions.unsubscribe(); + } } diff --git a/apps/web/src/app/xtream-tauri/services/epg-queue.service.ts b/apps/web/src/app/xtream-tauri/services/epg-queue.service.ts new file mode 100644 index 000000000..41a23327e --- /dev/null +++ b/apps/web/src/app/xtream-tauri/services/epg-queue.service.ts @@ -0,0 +1,134 @@ +import { inject, Injectable, OnDestroy } from '@angular/core'; +import { Subject } from 'rxjs'; +import { EpgItem } from 'shared-interfaces'; +import { XtreamApiService, XtreamCredentials } from './xtream-api.service'; + +interface CacheEntry { + data: EpgItem[]; + timestamp: number; +} + +/** + * Throttled EPG request queue with concurrency control, inter-request + * delay, and in-memory caching. Prevents Xtream providers from + * rate-limiting / banning the client when scrolling through large + * channel lists. + */ +@Injectable({ providedIn: 'root' }) +export class EpgQueueService implements OnDestroy { + private readonly apiService = inject(XtreamApiService); + + private readonly cache = new Map(); + private queue: number[] = []; + private readonly inFlight = new Set(); + private visibleSet = new Set(); + private processing = false; + + private readonly maxConcurrency = 2; + private readonly delayMs = 200; + private readonly cacheTtlMs = 5 * 60 * 1000; // 5 minutes + + /** Emits EPG results as they arrive. */ + readonly epgResult$ = new Subject<{ streamId: number; items: EpgItem[] }>(); + + /** Return cached EPG items if still valid, otherwise null. */ + getCached(streamId: number): EpgItem[] | null { + const entry = this.cache.get(streamId); + if (!entry) return null; + if (Date.now() - entry.timestamp > this.cacheTtlMs) { + this.cache.delete(streamId); + return null; + } + return entry.data; + } + + /** + * Enqueue stream IDs for EPG fetching. + * + * - IDs already cached or in-flight are skipped. + * - The internal queue is replaced so that stale IDs (no longer in + * `visibleIds`) are dropped on next dequeue. + */ + enqueue( + streamIds: number[], + visibleIds: Set, + credentials: XtreamCredentials + ): void { + this.visibleSet = visibleIds; + + const toFetch = streamIds.filter( + (id) => + !this.getCached(id) && + !this.inFlight.has(id) + ); + + // Replace the queue – any previously queued but now-invisible IDs + // will be skipped during processing via the visibleSet check. + this.queue = toFetch; + + if (!this.processing) { + this.processQueue(credentials); + } + } + + private async processQueue(credentials: XtreamCredentials): Promise { + this.processing = true; + + while (this.queue.length > 0) { + // Wait until a concurrency slot is available + if (this.inFlight.size >= this.maxConcurrency) { + await this.delay(this.delayMs); + continue; + } + + const streamId = this.queue.shift()!; + + // Drop stale entries that are no longer visible + if (!this.visibleSet.has(streamId)) continue; + + // Skip if it got cached while queued (e.g. duplicate) + if (this.getCached(streamId)) continue; + + this.inFlight.add(streamId); + this.fetchEpg(credentials, streamId); + + // Space out requests + await this.delay(this.delayMs); + } + + this.processing = false; + } + + private async fetchEpg( + credentials: XtreamCredentials, + streamId: number + ): Promise { + try { + const items = await this.apiService.getShortEpg( + credentials, + streamId, + 1 + ); + this.cache.set(streamId, { + data: items, + timestamp: Date.now(), + }); + this.epgResult$.next({ streamId, items }); + } catch (error) { + console.error( + `EpgQueueService: failed to load EPG for stream ${streamId}:`, + error + ); + } finally { + this.inFlight.delete(streamId); + } + } + + private delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); + } + + ngOnDestroy(): void { + this.epgResult$.complete(); + } +}