refactor(epg): implement throttled EPG request queue with caching and concurrency control

Entire-Checkpoint: 3ba3a7503b59

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<any>();
+export class PortalChannelsListComponent implements AfterViewInit, OnDestroy {
+    readonly playClicked = output<any>();
 
     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<string>('');
     currentPrograms = new Map<number, string>();
     currentProgramsProgress = new Map<number, number>();
-    programTimings = new Map<number, { start: number; end: number }>(); // Changed to store timestamps
-    private requestedChannels = new Set<number>();
+    programTimings = new Map<number, { start: number; end: number }>();
 
-    @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<void> {
+    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<number>(
+            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();
+    }
 }

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<number, CacheEntry>();
+    private queue: number[] = [];
+    private readonly inFlight = new Set<number>();
+    private visibleSet = new Set<number>();
+    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<number>,
+        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<void> {
+        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<void> {
+        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<void> {
+        return new Promise((resolve) => setTimeout(resolve, ms));
+    }
+
+    ngOnDestroy(): void {
+        this.epgResult$.complete();
+    }
+}