mirror of
https://github.com/4gray/iptvnator.git
synced 2026-10-14 05:06:16 -08:00
feat(dashboard): portal EPG on the live rails, loaded lazily per visible card (#1638)
Xtream and Stalker live cards on the dashboard carried no XMLTV key, so the rails never asked anything for them and showed only the LIVE chip. Their programme now comes from the portal, one card at a time and only once the card is on screen. - `lib-dashboard-rail` reports the cards inside its viewport through an IntersectionObserver rooted at the track; `DashboardLiveEpgPresenter` unions them with the pinned hero row and hands the set to `DashboardPortalLiveEpgPresenter`. - `DashboardPortalLiveEpgService` runs the bounded queue — two requests in flight, 200 ms apart, one card each — through `StreamResolverService.loadEpgForItems`, publishing every answer the moment it lands, so the page never waits and a slow portal delays no other card. A card scrolled past before its turn is never requested. - A programme is trusted for 60 s, an empty answer for 30 s (the resolver reports a dead portal and a guide-less channel identically), and a completion captures both the display offset and the EPG source revision, requeueing itself when either moved. - The presenter hands its wanted set back on destroy. Desktop only: the shared collection resolver is gated on the local XMLTV bridge. - A shimmer placeholder shows only before a card's first portal answer; M3U cards keep the batched XMLTV lookup.
This commit is contained in:
4gray authored and GitHub committed
1 parent
4e575a810e
commit
6578e4073c
19 files changed
+1937
-46
No files matched your search
@@ -1,4 +1,6 @@
|
||||
export * from './lib/dashboard-data.service';
|
||||
export * from './lib/dashboard-portal-live-epg.service';
|
||||
export * from './lib/dashboard-portal-live-epg.util';
|
||||
export * from './lib/dashboard-recommendations.service';
|
||||
export * from './lib/dashboard-recommendations.util';
|
||||
export * from './lib/dashboard-source-expiry.service';
|
||||
|
||||
+324
@@ -0,0 +1,324 @@
|
||||
import { TestBed } from '@angular/core/testing';
|
||||
import { Subject } from 'rxjs';
|
||||
import type { EpgProgram } from '@iptvnator/shared/interfaces';
|
||||
import {
|
||||
EpgSourceSettingsService,
|
||||
RuntimeCapabilitiesService,
|
||||
SettingsStore,
|
||||
} from '@iptvnator/services';
|
||||
import { StreamResolverService } from '@iptvnator/portal/shared/data-access';
|
||||
import {
|
||||
DASHBOARD_PORTAL_LIVE_EPG_TIMING,
|
||||
DashboardPortalLiveEpgService,
|
||||
} from './dashboard-portal-live-epg.service';
|
||||
import type { DashboardPortalLiveEpgEntry } from './dashboard-portal-live-epg.util';
|
||||
|
||||
interface Deferred {
|
||||
readonly key: string;
|
||||
resolve: (program: EpgProgram | null) => void;
|
||||
reject: (error: unknown) => void;
|
||||
}
|
||||
|
||||
const entry = (id: number): DashboardPortalLiveEpgEntry => ({
|
||||
key: `xtream::p::${id}`,
|
||||
item: {
|
||||
uid: `xtream::p::${id}`,
|
||||
name: `Channel ${id}`,
|
||||
contentType: 'live',
|
||||
sourceType: 'xtream',
|
||||
playlistId: 'p',
|
||||
playlistName: 'Portal',
|
||||
xtreamId: id,
|
||||
tvgId: String(id),
|
||||
},
|
||||
});
|
||||
|
||||
const program = (title: string, stopIso?: string): EpgProgram =>
|
||||
({
|
||||
channel: 'x',
|
||||
title,
|
||||
start: '2026-05-23T10:00:00.000Z',
|
||||
stop: stopIso ?? '2026-05-23T11:00:00.000Z',
|
||||
}) as EpgProgram;
|
||||
|
||||
describe('DashboardPortalLiveEpgService', () => {
|
||||
let service: DashboardPortalLiveEpgService;
|
||||
let loadEpgForItems: jest.Mock;
|
||||
let deferred: Deferred[];
|
||||
let offsetMinutes: number;
|
||||
let sourceChanged: Subject<void>;
|
||||
let sourceRevision: number;
|
||||
let supportsEpgProgramLookup: boolean;
|
||||
|
||||
const { delayMs, ttlMs, emptyTtlMs, endedRefetchFloorMs } =
|
||||
DASHBOARD_PORTAL_LIVE_EPG_TIMING;
|
||||
|
||||
/** What a reconciliation does: bump the fence, then announce it. */
|
||||
const changeEpgSources = () => {
|
||||
sourceRevision++;
|
||||
sourceChanged.next();
|
||||
};
|
||||
|
||||
/** Let the queue loop take its next step (one inter-request delay). */
|
||||
const step = async (rounds = 1): Promise<void> => {
|
||||
for (let i = 0; i < rounds; i++) {
|
||||
await jest.advanceTimersByTimeAsync(delayMs);
|
||||
}
|
||||
};
|
||||
|
||||
/** Settle the LATEST request for `key` — a key can be asked more than once. */
|
||||
const settle = async (key: string, value: EpgProgram | null) => {
|
||||
const request = [...deferred]
|
||||
.reverse()
|
||||
.find((request) => request.key === key);
|
||||
if (!request) throw new Error(`no request in flight for ${key}`);
|
||||
request.resolve(value);
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
jest.useFakeTimers();
|
||||
jest.setSystemTime(new Date('2026-05-23T10:30:00.000Z'));
|
||||
deferred = [];
|
||||
offsetMinutes = 0;
|
||||
sourceRevision = 0;
|
||||
supportsEpgProgramLookup = true;
|
||||
sourceChanged = new Subject<void>();
|
||||
loadEpgForItems = jest.fn((items: { tvgId?: string }[]) => {
|
||||
const key = `xtream::p::${items[0].tvgId}`;
|
||||
return new Promise<Map<string, EpgProgram | null>>(
|
||||
(resolve, reject) => {
|
||||
deferred.push({
|
||||
key,
|
||||
resolve: (value) =>
|
||||
resolve(new Map([[String(items[0].tvgId), value]])),
|
||||
reject,
|
||||
});
|
||||
}
|
||||
);
|
||||
});
|
||||
|
||||
TestBed.configureTestingModule({
|
||||
providers: [
|
||||
DashboardPortalLiveEpgService,
|
||||
{
|
||||
provide: StreamResolverService,
|
||||
useValue: { loadEpgForItems },
|
||||
},
|
||||
{
|
||||
provide: SettingsStore,
|
||||
useValue: {
|
||||
resolvedEpgOffsetMinutes: () => offsetMinutes,
|
||||
},
|
||||
},
|
||||
{
|
||||
provide: EpgSourceSettingsService,
|
||||
useValue: {
|
||||
changed$: sourceChanged,
|
||||
revision: () => sourceRevision,
|
||||
},
|
||||
},
|
||||
{
|
||||
provide: RuntimeCapabilitiesService,
|
||||
useValue: {
|
||||
get supportsEpgProgramLookup() {
|
||||
return supportsEpgProgramLookup;
|
||||
},
|
||||
},
|
||||
},
|
||||
],
|
||||
});
|
||||
service = TestBed.inject(DashboardPortalLiveEpgService);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
jest.useRealTimers();
|
||||
});
|
||||
|
||||
it('answers each card as its own request lands, never waiting for the slowest', async () => {
|
||||
service.sync([entry(1), entry(2)]);
|
||||
expect(service.pending()).toEqual(
|
||||
new Set(['xtream::p::1', 'xtream::p::2'])
|
||||
);
|
||||
await step(2);
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
|
||||
await settle('xtream::p::2', program('Fast answer'));
|
||||
expect(service.programs().get('xtream::p::2')?.title).toBe(
|
||||
'Fast answer'
|
||||
);
|
||||
expect(service.programs().has('xtream::p::1')).toBe(false);
|
||||
expect(service.pending()).toEqual(new Set(['xtream::p::1']));
|
||||
|
||||
await settle('xtream::p::1', null);
|
||||
expect(service.programs().get('xtream::p::1')).toBeNull();
|
||||
expect(service.pending().size).toBe(0);
|
||||
});
|
||||
|
||||
it('never has more than two requests in flight and spaces starts by the delay', async () => {
|
||||
service.sync([entry(1), entry(2), entry(3), entry(4)]);
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(1);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
await step(3);
|
||||
// Two in flight: the loop keeps waiting instead of starting a third.
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
|
||||
await settle('xtream::p::1', null);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it('drops a queued card that scrolled away before its turn and keeps the rest', async () => {
|
||||
service.sync([entry(1), entry(2), entry(3)]);
|
||||
// 1 started; 2 and 3 still queued.
|
||||
service.sync([entry(1), entry(3)]);
|
||||
expect(service.pending()).toEqual(
|
||||
new Set(['xtream::p::1', 'xtream::p::3'])
|
||||
);
|
||||
await step(2);
|
||||
const requested = loadEpgForItems.mock.calls.map(
|
||||
([items]) => items[0].tvgId
|
||||
);
|
||||
expect(requested).toEqual(['1', '3']);
|
||||
});
|
||||
|
||||
it('serves a fresh answer from cache and asks again once the TTL has passed', async () => {
|
||||
service.sync([entry(1)]);
|
||||
await settle('xtream::p::1', program('Cached'));
|
||||
await step();
|
||||
|
||||
service.sync([entry(1)]);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(1);
|
||||
|
||||
jest.setSystemTime(Date.now() + ttlMs);
|
||||
service.sync([entry(1)]);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('asks again when the programme on air has ended, but not more often than the floor', async () => {
|
||||
service.sync([entry(1)]);
|
||||
await settle(
|
||||
'xtream::p::1',
|
||||
program('Ending soon', '2026-05-23T10:35:00.000Z')
|
||||
);
|
||||
await step();
|
||||
|
||||
// Ended 5 min in, but the floor (30 s) already passed → refetch.
|
||||
jest.setSystemTime(Date.now() + 6 * 60_000);
|
||||
service.sync([entry(1)]);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
|
||||
// Portal keeps returning the ended row: not re-asked within the floor.
|
||||
await settle(
|
||||
'xtream::p::1',
|
||||
program('Still ended', '2026-05-23T10:35:00.000Z')
|
||||
);
|
||||
await step();
|
||||
jest.setSystemTime(Date.now() + endedRefetchFloorMs / 2);
|
||||
service.sync([entry(1)]);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('expires an answer with no programme sooner than one with a programme', async () => {
|
||||
// The resolver reports a failed portal and a guide-less channel the
|
||||
// same way, so the short TTL is what lets an outage recover.
|
||||
service.sync([entry(1)]);
|
||||
await settle('xtream::p::1', null);
|
||||
await step();
|
||||
expect(service.programs().get('xtream::p::1')).toBeNull();
|
||||
|
||||
jest.setSystemTime(Date.now() + emptyTtlMs / 2);
|
||||
service.sync([entry(1)]);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(1);
|
||||
|
||||
jest.setSystemTime(Date.now() + emptyTtlMs / 2);
|
||||
service.sync([entry(1)]);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
|
||||
// A programme keeps the full TTL, which outlives the empty one.
|
||||
await settle('xtream::p::1', program('On air'));
|
||||
await step();
|
||||
jest.setSystemTime(Date.now() + emptyTtlMs);
|
||||
service.sync([entry(1)]);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('treats a rejected resolver call as an answer with no programme', async () => {
|
||||
service.sync([entry(1)]);
|
||||
deferred[0].reject(new Error('portal down'));
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
|
||||
expect(service.programs().get('xtream::p::1')).toBeNull();
|
||||
expect(service.pending().size).toBe(0);
|
||||
|
||||
jest.setSystemTime(Date.now() + emptyTtlMs);
|
||||
service.sync([entry(1)]);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('retires an answer evaluated under a previous display offset and asks again', async () => {
|
||||
service.sync([entry(1)]);
|
||||
offsetMinutes = 60;
|
||||
await settle('xtream::p::1', program('Old clock'));
|
||||
await step();
|
||||
expect(service.programs().has('xtream::p::1')).toBe(false);
|
||||
// Still wanted → requeued under the new clock.
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
|
||||
await settle('xtream::p::1', program('New clock'));
|
||||
expect(service.programs().get('xtream::p::1')?.title).toBe('New clock');
|
||||
});
|
||||
|
||||
it('drops every answer when the EPG sources change and asks the wanted cards again', async () => {
|
||||
service.sync([entry(1)]);
|
||||
await settle('xtream::p::1', program('Before import'));
|
||||
await step();
|
||||
|
||||
changeEpgSources();
|
||||
expect(service.programs().size).toBe(0);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('discards an answer computed before an EPG source change and asks again', async () => {
|
||||
// The key is in flight when the sources change, so the retire pass
|
||||
// cannot requeue it; the completion must not publish the old guide's
|
||||
// answer and must ask again itself.
|
||||
service.sync([entry(1)]);
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(1);
|
||||
|
||||
changeEpgSources();
|
||||
await settle('xtream::p::1', program('Removed guide'));
|
||||
expect(service.programs().has('xtream::p::1')).toBe(false);
|
||||
|
||||
await step();
|
||||
expect(loadEpgForItems).toHaveBeenCalledTimes(2);
|
||||
await settle('xtream::p::1', program('Current guide'));
|
||||
expect(service.programs().get('xtream::p::1')?.title).toBe(
|
||||
'Current guide'
|
||||
);
|
||||
});
|
||||
|
||||
it('does nothing at all without the local EPG program-lookup capability', async () => {
|
||||
// PWA: the collection resolver is gated on the desktop XMLTV bridge
|
||||
// and answers nothing, so no request is worth queuing.
|
||||
supportsEpgProgramLookup = false;
|
||||
|
||||
service.sync([entry(1), entry(2)]);
|
||||
await step(3);
|
||||
|
||||
expect(loadEpgForItems).not.toHaveBeenCalled();
|
||||
expect(service.pending().size).toBe(0);
|
||||
expect(service.programs().size).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,275 @@
|
||||
import { inject, Injectable, OnDestroy, signal } from '@angular/core';
|
||||
import {
|
||||
epgProviderClockMs,
|
||||
type EpgProgram,
|
||||
} from '@iptvnator/shared/interfaces';
|
||||
import {
|
||||
EpgSourceSettingsService,
|
||||
RuntimeCapabilitiesService,
|
||||
SettingsStore,
|
||||
} from '@iptvnator/services';
|
||||
import { StreamResolverService } from '@iptvnator/portal/shared/data-access';
|
||||
import {
|
||||
dashboardPortalLiveEpgProgramStopMs,
|
||||
resolveDashboardPortalLiveEpgProgram,
|
||||
type DashboardPortalLiveEpgEntry,
|
||||
} from './dashboard-portal-live-epg.util';
|
||||
|
||||
interface CachedProgram {
|
||||
readonly program: EpgProgram | null;
|
||||
readonly fetchedAt: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Same numbers `EpgQueueService` uses against real Xtream panels: two
|
||||
* requests in flight, 200 ms between starts. A programme is trusted for a
|
||||
* minute; a programme that ended is asked again, but never more often than
|
||||
* every 30 s in case the portal keeps returning the stale row.
|
||||
*
|
||||
* An answer with no programme expires sooner (`emptyTtlMs`) because the
|
||||
* collection resolver reports a failed portal and a channel with no guide
|
||||
* identically — it catches per-channel failures and files them as `null`.
|
||||
* There is therefore no failure cooldown to keep: the short TTL is what lets
|
||||
* a transient outage recover on the next tick, at the price of re-asking a
|
||||
* genuinely guide-less channel while its card stays on screen.
|
||||
*/
|
||||
export const DASHBOARD_PORTAL_LIVE_EPG_TIMING = Object.freeze({
|
||||
maxConcurrency: 2,
|
||||
delayMs: 200,
|
||||
ttlMs: 60_000,
|
||||
emptyTtlMs: 30_000,
|
||||
endedRefetchFloorMs: 30_000,
|
||||
});
|
||||
|
||||
/**
|
||||
* Lazy, per-card "what is on air" for the dashboard's Xtream and Stalker
|
||||
* live cards. `sync()` receives the cards currently worth asking for — the
|
||||
* visible ones — and the service answers each through the collection
|
||||
* resolver one card at a time, publishing every answer the moment it lands
|
||||
* (`programs`) so a slow portal never delays a fast one. The queue holds
|
||||
* only wanted keys, so a card scrolled away before its turn is never
|
||||
* requested. One instance for the app: favourites and recent rows of the
|
||||
* same channel share the key and therefore the answer.
|
||||
*/
|
||||
@Injectable({ providedIn: 'root' })
|
||||
export class DashboardPortalLiveEpgService implements OnDestroy {
|
||||
private readonly streamResolver = inject(StreamResolverService);
|
||||
private readonly settingsStore = inject(SettingsStore);
|
||||
private readonly runtime = inject(RuntimeCapabilitiesService);
|
||||
private readonly sourceSettings = inject(EpgSourceSettingsService);
|
||||
private readonly sourceSubscription =
|
||||
this.sourceSettings.changed$.subscribe(() => this.retireAnswers());
|
||||
|
||||
private readonly cache = new Map<string, CachedProgram>();
|
||||
private wanted = new Map<string, DashboardPortalLiveEpgEntry>();
|
||||
private queue: string[] = [];
|
||||
private readonly inFlight = new Set<string>();
|
||||
private processing = false;
|
||||
/** Display offset every cached answer was evaluated under. */
|
||||
private stateOffsetMinutes = this.offsetMinutes();
|
||||
|
||||
private readonly programsState = signal<
|
||||
ReadonlyMap<string, EpgProgram | null>
|
||||
>(new Map());
|
||||
private readonly pendingState = signal<ReadonlySet<string>>(new Set());
|
||||
|
||||
/** Answers by entry key; absent = not asked yet, `null` = nothing on air. */
|
||||
readonly programs = this.programsState.asReadonly();
|
||||
/** Keys queued or in flight — the cards that may show a placeholder. */
|
||||
readonly pending = this.pendingState.asReadonly();
|
||||
|
||||
/**
|
||||
* Replace the wanted set. Keys without a fresh answer are queued; keys no
|
||||
* longer wanted are dropped from the queue (an in-flight request is
|
||||
* allowed to finish and is cached for when the card scrolls back).
|
||||
*/
|
||||
sync(entries: readonly DashboardPortalLiveEpgEntry[]): void {
|
||||
// The collection resolver this queue asks through is gated on the
|
||||
// local XMLTV bridge (`supportsProgramLookup`, desktop only) and
|
||||
// answers nothing without it, so the PWA never queues at all rather
|
||||
// than filing an empty answer for every card.
|
||||
if (!this.runtime.supportsEpgProgramLookup) {
|
||||
return;
|
||||
}
|
||||
this.retireStateOfPreviousOffset();
|
||||
this.wanted = new Map(entries.map((entry) => [entry.key, entry]));
|
||||
this.queue = this.queue.filter((key) => this.wanted.has(key));
|
||||
const now = Date.now();
|
||||
for (const entry of entries) {
|
||||
if (
|
||||
this.needsFetch(entry.key, now) &&
|
||||
!this.queue.includes(entry.key)
|
||||
) {
|
||||
this.queue.push(entry.key);
|
||||
}
|
||||
}
|
||||
this.publishPending();
|
||||
if (!this.processing && this.queue.length > 0) {
|
||||
void this.processQueue();
|
||||
}
|
||||
}
|
||||
|
||||
ngOnDestroy(): void {
|
||||
this.sourceSubscription.unsubscribe();
|
||||
}
|
||||
|
||||
private offsetMinutes(): number {
|
||||
return this.settingsStore.resolvedEpgOffsetMinutes();
|
||||
}
|
||||
|
||||
private needsFetch(key: string, now: number): boolean {
|
||||
if (this.inFlight.has(key)) {
|
||||
return false;
|
||||
}
|
||||
const cached = this.cache.get(key);
|
||||
if (!cached) {
|
||||
return true;
|
||||
}
|
||||
const ttlMs = cached.program
|
||||
? DASHBOARD_PORTAL_LIVE_EPG_TIMING.ttlMs
|
||||
: DASHBOARD_PORTAL_LIVE_EPG_TIMING.emptyTtlMs;
|
||||
if (now - cached.fetchedAt >= ttlMs) {
|
||||
return true;
|
||||
}
|
||||
const stopMs = dashboardPortalLiveEpgProgramStopMs(cached.program);
|
||||
return (
|
||||
stopMs !== null &&
|
||||
epgProviderClockMs(now, this.stateOffsetMinutes) >= stopMs &&
|
||||
now - cached.fetchedAt >=
|
||||
DASHBOARD_PORTAL_LIVE_EPG_TIMING.endedRefetchFloorMs
|
||||
);
|
||||
}
|
||||
|
||||
private async processQueue(): Promise<void> {
|
||||
this.processing = true;
|
||||
try {
|
||||
while (this.queue.length > 0) {
|
||||
if (
|
||||
this.inFlight.size >=
|
||||
DASHBOARD_PORTAL_LIVE_EPG_TIMING.maxConcurrency
|
||||
) {
|
||||
await this.delay();
|
||||
continue;
|
||||
}
|
||||
const key = this.queue.shift();
|
||||
if (
|
||||
key == null ||
|
||||
!this.wanted.has(key) ||
|
||||
!this.needsFetch(key, Date.now())
|
||||
) {
|
||||
this.publishPending();
|
||||
continue;
|
||||
}
|
||||
this.inFlight.add(key);
|
||||
void this.fetch(key);
|
||||
await this.delay();
|
||||
}
|
||||
} finally {
|
||||
this.processing = false;
|
||||
}
|
||||
}
|
||||
|
||||
private async fetch(key: string): Promise<void> {
|
||||
const entry = this.wanted.get(key);
|
||||
if (!entry) {
|
||||
this.inFlight.delete(key);
|
||||
this.publishPending();
|
||||
return;
|
||||
}
|
||||
// Both facts this answer is evaluated against. The revision is the
|
||||
// same fence `EpgService.guard()` uses: a reconciliation bumps it, so
|
||||
// a result computed against the previous XMLTV source set can be told
|
||||
// apart from one computed against the current one.
|
||||
const offsetMinutes = this.offsetMinutes();
|
||||
const revision = this.sourceSettings.revision();
|
||||
let program: EpgProgram | null = null;
|
||||
try {
|
||||
const epgMap = await this.streamResolver.loadEpgForItems([
|
||||
entry.item,
|
||||
]);
|
||||
program = resolveDashboardPortalLiveEpgProgram(epgMap, entry);
|
||||
} catch {
|
||||
// The resolver files a failed portal as `null` itself, so this
|
||||
// only catches a resolver-level throw. Same answer either way,
|
||||
// and `emptyTtlMs` is what makes it recoverable.
|
||||
program = null;
|
||||
}
|
||||
this.inFlight.delete(key);
|
||||
|
||||
// A setting or the source set changed while the request was on the
|
||||
// wire: this answer belongs to the previous provider clock or the
|
||||
// previous guide. `retireAnswers()` could not requeue the key while
|
||||
// it was in flight, so the requeue happens here — otherwise the stale
|
||||
// answer would be published and trusted for a full TTL.
|
||||
if (
|
||||
offsetMinutes !== this.offsetMinutes() ||
|
||||
revision !== this.sourceSettings.revision()
|
||||
) {
|
||||
this.retireStateOfPreviousOffset();
|
||||
this.requeueIfWanted(key);
|
||||
return;
|
||||
}
|
||||
|
||||
this.cache.set(key, { program, fetchedAt: Date.now() });
|
||||
this.programsState.update((programs) => {
|
||||
const next = new Map(programs);
|
||||
next.set(key, program);
|
||||
return next;
|
||||
});
|
||||
this.publishPending();
|
||||
}
|
||||
|
||||
private requeueIfWanted(key: string): void {
|
||||
if (this.wanted.has(key) && !this.queue.includes(key)) {
|
||||
this.queue.push(key);
|
||||
}
|
||||
this.publishPending();
|
||||
if (!this.processing && this.queue.length > 0) {
|
||||
void this.processQueue();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Every answer here is "what is on at the provider clock", so a changed
|
||||
* display offset drops all of them. The caller's own `sync` (the
|
||||
* presenter re-syncs on an offset change) asks the wanted cards again.
|
||||
*/
|
||||
private retireStateOfPreviousOffset(): void {
|
||||
const current = this.offsetMinutes();
|
||||
if (current === this.stateOffsetMinutes) {
|
||||
return;
|
||||
}
|
||||
this.stateOffsetMinutes = current;
|
||||
this.dropAnswers();
|
||||
}
|
||||
|
||||
/** Removed or re-imported XMLTV: drop every answer and ask again now. */
|
||||
private retireAnswers(): void {
|
||||
this.dropAnswers();
|
||||
if (this.wanted.size > 0) {
|
||||
this.sync(Array.from(this.wanted.values()));
|
||||
}
|
||||
}
|
||||
|
||||
private dropAnswers(): void {
|
||||
this.cache.clear();
|
||||
this.programsState.set(new Map());
|
||||
}
|
||||
|
||||
private publishPending(): void {
|
||||
const pending = new Set<string>();
|
||||
for (const key of this.queue) {
|
||||
if (this.wanted.has(key)) pending.add(key);
|
||||
}
|
||||
for (const key of this.inFlight) {
|
||||
if (this.wanted.has(key)) pending.add(key);
|
||||
}
|
||||
this.pendingState.set(pending);
|
||||
}
|
||||
|
||||
private delay(): Promise<void> {
|
||||
return new Promise((resolve) =>
|
||||
setTimeout(resolve, DASHBOARD_PORTAL_LIVE_EPG_TIMING.delayMs)
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,160 @@
|
||||
import type {
|
||||
EpgProgram,
|
||||
PortalActivityItem,
|
||||
} from '@iptvnator/shared/interfaces';
|
||||
import {
|
||||
buildDashboardPortalLiveEpgEntry,
|
||||
buildDashboardPortalLiveEpgKey,
|
||||
dashboardPortalLiveEpgProgramStopMs,
|
||||
resolveDashboardPortalLiveEpgProgram,
|
||||
} from './dashboard-portal-live-epg.util';
|
||||
|
||||
const baseItem = (overrides: Partial<PortalActivityItem>): PortalActivityItem =>
|
||||
({
|
||||
id: 'item-1',
|
||||
title: 'News 24',
|
||||
type: 'live',
|
||||
playlist_id: 'playlist-1',
|
||||
playlist_name: 'My portal',
|
||||
category_id: '7',
|
||||
xtream_id: 42,
|
||||
poster_url: 'https://example.com/logo.png',
|
||||
...overrides,
|
||||
}) as PortalActivityItem;
|
||||
|
||||
describe('buildDashboardPortalLiveEpgEntry', () => {
|
||||
it('builds an Xtream live entry keyed by the collection uid with the stream id as tvgId', () => {
|
||||
const entry = buildDashboardPortalLiveEpgEntry(
|
||||
baseItem({ source: 'xtream', xtream_id: 42 })
|
||||
);
|
||||
|
||||
expect(entry?.key).toBe('xtream::playlist-1::42');
|
||||
expect(entry?.item).toMatchObject({
|
||||
uid: 'xtream::playlist-1::42',
|
||||
name: 'News 24',
|
||||
contentType: 'live',
|
||||
sourceType: 'xtream',
|
||||
playlistId: 'playlist-1',
|
||||
playlistName: 'My portal',
|
||||
logo: 'https://example.com/logo.png',
|
||||
xtreamId: 42,
|
||||
tvgId: '42',
|
||||
categoryId: '7',
|
||||
});
|
||||
});
|
||||
|
||||
it('accepts a numeric string Xtream id and rejects ids that cannot name a stream', () => {
|
||||
expect(
|
||||
buildDashboardPortalLiveEpgEntry(
|
||||
baseItem({ source: 'xtream', xtream_id: '17' })
|
||||
)?.item.xtreamId
|
||||
).toBe(17);
|
||||
for (const xtream_id of ['', 'abc', 0, -3, 1.5]) {
|
||||
expect(
|
||||
buildDashboardPortalLiveEpgEntry(
|
||||
baseItem({ source: 'xtream', xtream_id })
|
||||
)
|
||||
).toBeNull();
|
||||
}
|
||||
});
|
||||
|
||||
it('builds a Stalker live entry from the stored portal item, keyed by the extracted id', () => {
|
||||
const stalkerItem = {
|
||||
id: 'ch-9',
|
||||
cmd: 'ffrt http://portal/9',
|
||||
radio: false,
|
||||
};
|
||||
const entry = buildDashboardPortalLiveEpgEntry(
|
||||
baseItem({
|
||||
source: 'stalker',
|
||||
id: 'ch-9',
|
||||
xtream_id: 'ch-9',
|
||||
stalker_item: stalkerItem,
|
||||
})
|
||||
);
|
||||
|
||||
expect(entry?.key).toBe('stalker::playlist-1::ch-9');
|
||||
expect(entry?.item).toMatchObject({
|
||||
sourceType: 'stalker',
|
||||
stalkerId: 'ch-9',
|
||||
tvgId: 'ch-9',
|
||||
stalkerCmd: 'ffrt http://portal/9',
|
||||
stalkerItem,
|
||||
});
|
||||
});
|
||||
|
||||
it('skips Stalker radio rows and rows without the stored item, like the collection resolver', () => {
|
||||
expect(
|
||||
buildDashboardPortalLiveEpgEntry(
|
||||
baseItem({
|
||||
source: 'stalker',
|
||||
stalker_item: { id: 'r-1', radio: 'true' },
|
||||
})
|
||||
)
|
||||
).toBeNull();
|
||||
expect(
|
||||
buildDashboardPortalLiveEpgEntry(
|
||||
baseItem({ source: 'stalker', stalker_item: undefined })
|
||||
)
|
||||
).toBeNull();
|
||||
});
|
||||
|
||||
it('leaves M3U rows and non-live rows to the XMLTV batch', () => {
|
||||
expect(
|
||||
buildDashboardPortalLiveEpgEntry(baseItem({ source: 'm3u' }))
|
||||
).toBeNull();
|
||||
expect(
|
||||
buildDashboardPortalLiveEpgEntry(
|
||||
baseItem({ source: 'xtream', type: 'movie' })
|
||||
)
|
||||
).toBeNull();
|
||||
expect(
|
||||
buildDashboardPortalLiveEpgKey(baseItem({ source: 'xtream' }))
|
||||
).toBe('xtream::playlist-1::42');
|
||||
expect(
|
||||
buildDashboardPortalLiveEpgKey(baseItem({ source: 'm3u' }))
|
||||
).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe('resolveDashboardPortalLiveEpgProgram', () => {
|
||||
it('reads the resolver map by the entry tvgId and treats a missing key as nothing on air', () => {
|
||||
const entry = buildDashboardPortalLiveEpgEntry(
|
||||
baseItem({ source: 'xtream', xtream_id: 42 })
|
||||
);
|
||||
const program = { title: 'Evening news' } as EpgProgram;
|
||||
if (!entry) throw new Error('expected an entry');
|
||||
|
||||
expect(
|
||||
resolveDashboardPortalLiveEpgProgram(
|
||||
new Map([['42', program]]),
|
||||
entry
|
||||
)
|
||||
).toBe(program);
|
||||
expect(
|
||||
resolveDashboardPortalLiveEpgProgram(new Map(), entry)
|
||||
).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe('dashboardPortalLiveEpgProgramStopMs', () => {
|
||||
it('prefers the unix-seconds stop, falls back to the ISO string, and reports null otherwise', () => {
|
||||
expect(
|
||||
dashboardPortalLiveEpgProgramStopMs({
|
||||
stopTimestamp: 1_700_000_000,
|
||||
stop: '2026-01-01T00:00:00.000Z',
|
||||
} as EpgProgram)
|
||||
).toBe(1_700_000_000_000);
|
||||
expect(
|
||||
dashboardPortalLiveEpgProgramStopMs({
|
||||
stop: '2026-01-01T00:00:00.000Z',
|
||||
} as EpgProgram)
|
||||
).toBe(Date.parse('2026-01-01T00:00:00.000Z'));
|
||||
expect(
|
||||
dashboardPortalLiveEpgProgramStopMs({
|
||||
stop: 'garbage',
|
||||
} as EpgProgram)
|
||||
).toBeNull();
|
||||
expect(dashboardPortalLiveEpgProgramStopMs(null)).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,136 @@
|
||||
import {
|
||||
isStalkerRadioItem,
|
||||
type EpgProgram,
|
||||
type PortalActivityItem,
|
||||
} from '@iptvnator/shared/interfaces';
|
||||
import {
|
||||
buildCollectionUid,
|
||||
type UnifiedCollectionItem,
|
||||
} from '@iptvnator/portal/shared/util';
|
||||
|
||||
/**
|
||||
* One dashboard live card's portal EPG request: the key its answer is filed
|
||||
* under (the collection uid, stable across favourites and recent rows of the
|
||||
* same channel) and the item `StreamResolverService.loadEpgForItems` fetches
|
||||
* for — the same shape the "See all" pages hand it, so the two surfaces
|
||||
* resolve a channel identically.
|
||||
*/
|
||||
export interface DashboardPortalLiveEpgEntry {
|
||||
readonly key: string;
|
||||
readonly item: UnifiedCollectionItem;
|
||||
}
|
||||
|
||||
/** The key alone, for cards that only need to find their answer. */
|
||||
export function buildDashboardPortalLiveEpgKey(
|
||||
item: PortalActivityItem
|
||||
): string | null {
|
||||
return buildDashboardPortalLiveEpgEntry(item)?.key ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* M3U channels keep the dashboard's batched XMLTV lookup; Xtream and Stalker
|
||||
* live rows have no XMLTV key of their own and are answered by their portal.
|
||||
* Radio rows are skipped like the collection resolver skips them.
|
||||
*/
|
||||
export function buildDashboardPortalLiveEpgEntry(
|
||||
item: PortalActivityItem
|
||||
): DashboardPortalLiveEpgEntry | null {
|
||||
if (item.type !== 'live') {
|
||||
return null;
|
||||
}
|
||||
if (item.source === 'xtream') {
|
||||
return buildXtreamEntry(item);
|
||||
}
|
||||
if (item.source === 'stalker') {
|
||||
return buildStalkerEntry(item);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* The resolver keys its map by the item's `tvgId`; the entry sets it to the
|
||||
* provider id, so a missing key means "asked, nothing on air".
|
||||
*/
|
||||
export function resolveDashboardPortalLiveEpgProgram(
|
||||
epgMap: ReadonlyMap<string, EpgProgram | null>,
|
||||
entry: DashboardPortalLiveEpgEntry
|
||||
): EpgProgram | null {
|
||||
const key = entry.item.tvgId?.trim();
|
||||
return key ? (epgMap.get(key) ?? null) : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* When the programme on air ends, in wall-clock milliseconds; `null` when the
|
||||
* row states no usable end. Reads the pre-computed unix-seconds field first,
|
||||
* then the ISO string, like the dashboard's own time-range helper.
|
||||
*/
|
||||
export function dashboardPortalLiveEpgProgramStopMs(
|
||||
program: EpgProgram | null
|
||||
): number | null {
|
||||
if (!program) {
|
||||
return null;
|
||||
}
|
||||
const cached = Number(program.stopTimestamp);
|
||||
if (Number.isFinite(cached) && cached > 0) {
|
||||
return cached * 1000;
|
||||
}
|
||||
const parsed = program.stop ? Date.parse(program.stop) : NaN;
|
||||
return Number.isFinite(parsed) ? parsed : null;
|
||||
}
|
||||
|
||||
function buildXtreamEntry(
|
||||
item: PortalActivityItem
|
||||
): DashboardPortalLiveEpgEntry | null {
|
||||
const xtreamId = Number(item.xtream_id);
|
||||
if (!Number.isInteger(xtreamId) || xtreamId <= 0) {
|
||||
return null;
|
||||
}
|
||||
const key = buildCollectionUid('xtream', item.playlist_id, xtreamId);
|
||||
return {
|
||||
key,
|
||||
item: {
|
||||
uid: key,
|
||||
name: item.title,
|
||||
contentType: 'live',
|
||||
sourceType: 'xtream',
|
||||
playlistId: item.playlist_id,
|
||||
playlistName: item.playlist_name ?? 'Xtream',
|
||||
logo: item.poster_url ?? null,
|
||||
xtreamId,
|
||||
tvgId: String(xtreamId),
|
||||
categoryId: item.category_id,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function buildStalkerEntry(
|
||||
item: PortalActivityItem
|
||||
): DashboardPortalLiveEpgEntry | null {
|
||||
const raw = item.stalker_item;
|
||||
if (!raw || isStalkerRadioItem(raw)) {
|
||||
return null;
|
||||
}
|
||||
// Both dashboard mappers already store the extracted portal id as `id`.
|
||||
const stalkerId = String(item.id ?? '').trim();
|
||||
if (!stalkerId) {
|
||||
return null;
|
||||
}
|
||||
const key = buildCollectionUid('stalker', item.playlist_id, stalkerId);
|
||||
return {
|
||||
key,
|
||||
item: {
|
||||
uid: key,
|
||||
name: item.title,
|
||||
contentType: 'live',
|
||||
sourceType: 'stalker',
|
||||
playlistId: item.playlist_id,
|
||||
playlistName: item.playlist_name ?? 'Stalker',
|
||||
logo: item.poster_url ?? null,
|
||||
stalkerId,
|
||||
tvgId: stalkerId,
|
||||
stalkerCmd: raw.cmd,
|
||||
categoryId: item.category_id,
|
||||
stalkerItem: raw,
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user