import { MonoTypeOperatorFunction, Observable, of } from 'rxjs'; import { finalize, shareReplay, tap } from 'rxjs/operators'; import { EpgProgram } from '@iptvnator/shared/interfaces'; import { normalizeEpgUrls } from '@iptvnator/shared/m3u-utils'; export interface CachedProgram { program: EpgProgram | null; timestamp: number; /** Display offset the "now" verdict was computed with; a changed setting invalidates the entry. */ offsetMinutes: number; } const CACHE_TTL_MS = 60000; /** * The 60 s "currently airing" memory behind every lookup in `EpgService`, * plus the in-flight requests sharing one answer. * * Owning it here keeps one rule in one place: an entry answers "what is on * at the provider clock, in this source scope", so the scope and the display * offset are both part of its identity. A lookup issued after the offset * changed must neither read the previous entry nor join a request still in * flight for it. * * Deliberately not `@Injectable`: it has no dependencies of its own, it * takes the two facts it cannot know (the current offset, and the operator * that retires work across an EPG source change) from its owner. */ export class EpgProgramCache { private readonly programs = new Map(); private readonly inFlight = new Map< string, Observable >(); private readonly inFlightBatches = new Map< string, Observable> >(); constructor( private readonly offsetMinutes: () => number, private readonly guard: () => MonoTypeOperatorFunction ) {} /** Cache and in-flight identity of a single-channel lookup. */ keyFor(channelId: string, sourceUrls: string[] = []): string { const normalizedSourceUrls = normalizeEpgUrls(sourceUrls); const key = normalizedSourceUrls.length === 0 ? channelId : `source:${channelId}:${JSON.stringify(normalizedSourceUrls)}`; const offsetMinutes = this.offsetMinutes(); return offsetMinutes === 0 ? key : `${key}|offset:${offsetMinutes}`; } /** Identity of a batch request; order-insensitive in the channel ids. */ batchKeyFor( channelIds: string[], sourceUrls: string[], fallbackSourceUrls: string[] ): string { return JSON.stringify({ channelIds: [...channelIds].sort(), sourceUrls: normalizeEpgUrls(sourceUrls), fallbackSourceUrls: normalizeEpgUrls(fallbackSourceUrls), offsetMinutes: this.offsetMinutes(), }); } /** A live entry, or `undefined` once it expired or the offset moved. */ get(cacheKey: string): CachedProgram | undefined { const cached = this.programs.get(cacheKey); if (!cached) { return undefined; } if ( Date.now() - cached.timestamp >= CACHE_TTL_MS || cached.offsetMinutes !== this.offsetMinutes() ) { this.programs.delete(cacheKey); return undefined; } return cached; } /** * `offsetMinutes` is the offset the verdict was computed with, not the * one current when the answer lands. */ set( cacheKey: string, program: EpgProgram | null, offsetMinutes: number, timestamp = Date.now() ): void { this.programs.set(cacheKey, { program, timestamp, offsetMinutes }); } /** Unexpired entry without the scope/offset checks `get` applies. */ getFresh(cacheKey: string, now: number): CachedProgram | undefined { const cached = this.programs.get(cacheKey); return cached && now - cached.timestamp < CACHE_TTL_MS ? cached : undefined; } /** The cached answer, an in-flight one, else one started from `fetch`. */ getOrFetch( cacheKey: string, fetchProgram: () => Observable ): Observable { const cached = this.get(cacheKey); if (cached) { return of(cached.program); } const existingRequest = this.inFlight.get(cacheKey); if (existingRequest) { return existingRequest; } const offsetMinutes = this.offsetMinutes(); const request$ = fetchProgram().pipe( this.guard(), tap((program) => this.set(cacheKey, program, offsetMinutes)), finalize(() => { if (this.inFlight.get(cacheKey) === request$) this.inFlight.delete(cacheKey); }), shareReplay({ bufferSize: 1, refCount: false }) ); this.inFlight.set(cacheKey, request$); return request$; } batchInFlight( batchCacheKey: string ): Observable> | undefined { return this.inFlightBatches.get(batchCacheKey); } registerBatch( batchCacheKey: string, request$: Observable> ): void { this.inFlightBatches.set(batchCacheKey, request$); } /** Only the request that owns the slot may release it. */ releaseBatch( batchCacheKey: string, request$: Observable> ): void { if (this.inFlightBatches.get(batchCacheKey) === request$) { this.inFlightBatches.delete(batchCacheKey); } } clear(): void { this.programs.clear(); this.inFlight.clear(); this.inFlightBatches.clear(); } }