mirror of
https://github.com/4gray/iptvnator.git
synced 2026-10-13 12:36:15 -08:00
fix(epg): address review feedback on the #1165 follow-up
- portal-channels-list: forward the already-validated playlistId into the mapping dialog instead of re-reading currentPlaylist() after the async getEpgMapping roundtrip (which could return undefined on navigation) - EpgQueueService.invalidate(): bump a per-stream invalidation epoch and clear inFlight so a request already running when the mapping changes has its (pre-change) result discarded via an epoch check in fetchEpg, and the immediate re-enqueue can schedule a fresh mapping-aware fetch - getMapping Xtream fallback: order the join deterministically before limit(1) so the resolved mapping is stable; documented that this backend layer has no caller playlist context and is a best-effort net behind the renderer's playlist-scoped resolution Tests: in-flight staleness discard for invalidate(). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
1 parent
ef1cff25ed
commit
47e4171f78
5 files changed
+69
-4
No files matched your search
@@ -417,7 +417,7 @@ describe('EpgQueryService', () => {
|
||||
});
|
||||
joinChain.where = jest.fn((condition: unknown) => {
|
||||
whereCalls.push(condition);
|
||||
return { limit: joinLimit };
|
||||
return { orderBy: jest.fn(() => ({ limit: joinLimit })) };
|
||||
});
|
||||
|
||||
// Program lookup for the resolved (mapped) channel id.
|
||||
|
||||
@@ -947,6 +947,13 @@ export class EpgQueryService {
|
||||
// The join key is built in SQL to mirror
|
||||
// buildXtreamEpgMappingKey(playlistId, xtreamId) →
|
||||
// `xtream:{playlistId}:{xtreamId}`; keep the two in sync.
|
||||
//
|
||||
// This layer only receives the provider epg_channel_id, not the
|
||||
// caller's playlist, so when the same id exists in several
|
||||
// playlists it cannot scope to the caller's own mapping — the
|
||||
// renderer resolves the playlist-scoped key directly and this is
|
||||
// a best-effort fallback. Order deterministically so the chosen
|
||||
// mapping is at least stable rather than storage-order dependent.
|
||||
const mapped = await db
|
||||
.select({
|
||||
epgChannelId: schema.epgChannelMappings.epgChannelId,
|
||||
@@ -964,6 +971,10 @@ export class EpgQueryService {
|
||||
)
|
||||
)
|
||||
.where(eq(schema.content.epgChannelId, channelKey))
|
||||
.orderBy(
|
||||
schema.categories.playlistId,
|
||||
schema.content.xtreamId
|
||||
)
|
||||
.limit(1);
|
||||
if (mapped.length > 0) {
|
||||
return mapped[0].epgChannelId;
|
||||
|
||||
@@ -372,6 +372,30 @@ describe('EpgQueueService', () => {
|
||||
expect(priv().epgChannelByStreamId.has(555)).toBe(false);
|
||||
expect(priv().xmltvPreviewByStreamId.has(555)).toBe(false);
|
||||
});
|
||||
|
||||
it('discards an in-flight result when the mapping is invalidated mid-request', async () => {
|
||||
let resolveEpg!: (items: EpgItem[]) => void;
|
||||
xtreamApi.getShortEpg.mockReturnValue(
|
||||
new Promise<EpgItem[]>((resolve) => {
|
||||
resolveEpg = resolve;
|
||||
})
|
||||
);
|
||||
const emitted: number[] = [];
|
||||
const sub = service.epgResult$.subscribe(({ streamId }) =>
|
||||
emitted.push(streamId)
|
||||
);
|
||||
|
||||
const fetchPromise = priv().fetchEpg(credentials, 707);
|
||||
// The user changes the mapping while the request is still running.
|
||||
service.invalidate(707);
|
||||
// The provider now returns the pre-change (stale) result.
|
||||
resolveEpg([makeEpgItem('rtl.de', 'Stale')]);
|
||||
await fetchPromise;
|
||||
|
||||
expect(service.getCached(707)).toBeNull();
|
||||
expect(emitted).not.toContain(707);
|
||||
sub.unsubscribe();
|
||||
});
|
||||
});
|
||||
|
||||
function makeEpgItem(channelId: string, title: string): EpgItem {
|
||||
|
||||
@@ -55,6 +55,8 @@ export class EpgQueueService implements OnDestroy {
|
||||
private readonly cache = new Map<number, CacheEntry>();
|
||||
private queue: number[] = [];
|
||||
private readonly inFlight = new Set<number>();
|
||||
/** Bumped by invalidate() so a stale in-flight result is discarded. */
|
||||
private readonly invalidationEpoch = new Map<number, number>();
|
||||
private readonly epgChannelByStreamId = new Map<number, string>();
|
||||
private readonly xmltvPreviewByStreamId = new Map<number, EpgItem>();
|
||||
private visibleSet = new Set<number>();
|
||||
@@ -84,12 +86,23 @@ export class EpgQueueService implements OnDestroy {
|
||||
* Drop every cached artifact for a stream so the next enqueue refetches
|
||||
* it. Used when a manual EPG mapping for the stream changes, since the
|
||||
* cached preview/resolution was computed for the previous mapping.
|
||||
*
|
||||
* Also bumps an invalidation epoch and clears `inFlight`: a request that
|
||||
* was already running when the mapping changed carries the pre-change
|
||||
* resolution, so its result is discarded (epoch mismatch in `fetchEpg`)
|
||||
* and clearing `inFlight` lets the immediate re-enqueue schedule a fresh
|
||||
* fetch through the mapping-aware `enqueue()` path.
|
||||
*/
|
||||
invalidate(streamId: number): void {
|
||||
this.cache.delete(streamId);
|
||||
this.failureTimestamps.delete(streamId);
|
||||
this.epgChannelByStreamId.delete(streamId);
|
||||
this.xmltvPreviewByStreamId.delete(streamId);
|
||||
this.inFlight.delete(streamId);
|
||||
this.invalidationEpoch.set(
|
||||
streamId,
|
||||
(this.invalidationEpoch.get(streamId) ?? 0) + 1
|
||||
);
|
||||
}
|
||||
|
||||
private isFailureCoolingDown(streamId: number): boolean {
|
||||
@@ -317,6 +330,9 @@ export class EpgQueueService implements OnDestroy {
|
||||
credentials: XtreamCredentials,
|
||||
streamId: number
|
||||
): Promise<void> {
|
||||
const startEpoch = this.invalidationEpoch.get(streamId) ?? 0;
|
||||
const isStale = (): boolean =>
|
||||
(this.invalidationEpoch.get(streamId) ?? 0) !== startEpoch;
|
||||
try {
|
||||
const apiItems = await this.apiService.getShortEpg(
|
||||
credentials,
|
||||
@@ -325,6 +341,11 @@ export class EpgQueueService implements OnDestroy {
|
||||
{ suppressErrorLog: true }
|
||||
);
|
||||
|
||||
// A mapping change during the request invalidated this result.
|
||||
if (isStale()) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (apiItems.length > 0) {
|
||||
this.recordSuccess(streamId, apiItems);
|
||||
return;
|
||||
@@ -333,6 +354,9 @@ export class EpgQueueService implements OnDestroy {
|
||||
const xmltv = this.xmltvPreviewByStreamId.get(streamId);
|
||||
this.recordSuccess(streamId, xmltv ? [xmltv] : []);
|
||||
} catch (error) {
|
||||
if (isStale()) {
|
||||
return;
|
||||
}
|
||||
this.failureTimestamps.set(streamId, Date.now());
|
||||
this.logger.error(
|
||||
`Failed to load EPG for stream ${streamId}`,
|
||||
|
||||
+9
-3
@@ -514,20 +514,26 @@ export class PortalChannelsListComponent implements AfterViewInit, OnDestroy {
|
||||
}
|
||||
|
||||
const channelKey = buildXtreamEpgMappingKey(playlistId, xtreamId);
|
||||
void this.openEpgMappingDialog(channelKey, channel, xtreamId);
|
||||
void this.openEpgMappingDialog(
|
||||
channelKey,
|
||||
channel,
|
||||
xtreamId,
|
||||
playlistId
|
||||
);
|
||||
}
|
||||
|
||||
private async openEpgMappingDialog(
|
||||
channelKey: string,
|
||||
channel: XtreamChannelListItem,
|
||||
streamId: number
|
||||
streamId: number,
|
||||
playlistId: string
|
||||
): Promise<void> {
|
||||
const mappingBefore = await this.readEpgMapping(channelKey);
|
||||
|
||||
EpgMappingDialogComponent.open(this.dialog, {
|
||||
channelKey,
|
||||
channelName: channel.title ?? channel.name ?? String(streamId),
|
||||
playlistId: this.xtreamStore.currentPlaylist()?.id,
|
||||
playlistId,
|
||||
})
|
||||
.afterClosed()
|
||||
.subscribe(async () => {
|
||||
|
||||
Reference in new issue
Block a user