fix(epg): resolve second-order review findings on the mapping follow-up

Two P2 issues Codex raised on the previous fixes:

- Unscoped program lookups could return the same programme twice now that
  the dedup index preserves per-source rows: the M3U timeline calls
  getChannelPrograms without sourceUrls, so two sources sharing
  channel/start/title both surfaced. Added toEpgProgams(), which collapses
  duplicate channel|start|title slots after mapping/validation, applied at
  every getChannelPrograms return.
- EpgQueueService.fetchEpg unconditionally cleared the in-flight marker in
  its finally, which could drop a marker a re-enqueued request took over
  after invalidate(). It now only releases the marker when the completing
  request still owns it (epoch unchanged), preserving per-stream dedup and
  concurrency accounting.

Tests: unscoped duplicate-slot collapse; stale request preserving a fresh
in-flight marker.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
4grayandClaude Fable 5 committed 2026-07-19 19:17:38 +02:00
1 parent e9d90120f3
commit 2521092753
4 files changed
+112 -10

No files matched your search

@@ -463,4 +463,58 @@ describe('EpgQueryService', () => {
expect(result).toHaveLength(1);
expect(result[0].title).toBe('Mapped Programme');
});
it('collapses duplicate programme slots from multiple sources in an unscoped lookup', async () => {
const whereCalls: unknown[] = [];
const dupRow = {
id: 1,
channelId: 'bbc.one.uk',
start: '2026-06-21T20:00:00.000Z',
stop: '2026-06-21T21:00:00.000Z',
title: 'News',
description: null,
category: null,
iconUrl: null,
rating: null,
episodeNum: null,
};
const programChain = {
from: jest.fn(() => ({
where: jest.fn(() => ({
orderBy: jest.fn(() => ({
limit: jest.fn().mockResolvedValue([
{ ...dupRow, sourceUrl: 'https://a.example/g.xml' },
{ ...dupRow, id: 2, sourceUrl: 'https://b.example/g.xml' },
]),
})),
})),
})),
};
const select = jest
.fn()
// getMapping direct + Xtream fallback — both miss.
.mockReturnValueOnce(createLimitedSelectChain([], whereCalls))
.mockReturnValueOnce({
from: jest.fn(() => ({
innerJoin: jest.fn(function inner() {
return this;
}),
where: jest.fn(() => ({
orderBy: jest.fn(() => ({
limit: jest.fn().mockResolvedValue([]),
})),
})),
})),
})
// selectChannelPrograms — two sources, same channel/start/title.
.mockReturnValueOnce(programChain);
getDatabase.mockResolvedValue({ select });
const result = await service.getChannelPrograms('bbc.one.uk');
expect(result).toHaveLength(1);
expect(result[0].title).toBe('News');
});
});
@@ -68,9 +68,7 @@ export class EpgQueryService {
}
if (results.length > 0) {
return results
.map(this.transformDbRowToEpgProgram)
.filter(this.isValidEpgProgram);
return this.toEpgPrograms(results);
}
let channel = await this.selectChannelById(
@@ -101,9 +99,7 @@ export class EpgQueryService {
}
if (results.length > 0) {
return results
.map(this.transformDbRowToEpgProgram)
.filter(this.isValidEpgProgram);
return this.toEpgPrograms(results);
}
}
@@ -153,9 +149,7 @@ export class EpgQueryService {
);
}
return results
.map(this.transformDbRowToEpgProgram)
.filter(this.isValidEpgProgram);
return this.toEpgPrograms(results);
}
return [];
@@ -894,6 +888,31 @@ export class EpgQueryService {
);
}
/**
* Map program rows to EpgProgram and drop invalid ones, then collapse
* duplicate programme slots. The dedup index is source-aware, so an
* unscoped lookup (e.g. the M3U timeline, which queries without
* `sourceUrls`) can return the same channel/start/title from two EPG
* sources; keep the first so a programme is never shown twice.
*/
private toEpgPrograms(rows: EpgProgramRow[]): EpgProgram[] {
const seen = new Set<string>();
const programs: EpgProgram[] = [];
for (const row of rows) {
const program = this.transformDbRowToEpgProgram(row);
if (!this.isValidEpgProgram(program)) {
continue;
}
const key = `${program.channel}|${program.start}|${program.title}`;
if (seen.has(key)) {
continue;
}
seen.add(key);
programs.push(program);
}
return programs;
}
private transformDbRowToEpgProgram(row: EpgProgramRow): EpgProgram {
return {
start: row.start,
@@ -28,6 +28,7 @@ describe('EpgQueueService invalidation', () => {
shouldFetch: (streamId: number) => boolean;
xmltvPreviewByStreamId: Map<number, EpgItem>;
epgChannelByStreamId: Map<number, string>;
inFlight: Set<number>;
};
beforeEach(() => {
@@ -116,4 +117,25 @@ describe('EpgQueueService invalidation', () => {
expect(emitted).not.toContain(707);
sub.unsubscribe();
});
it('leaves a fresh in-flight marker intact when a stale request completes', async () => {
let resolveA!: (items: EpgItem[]) => void;
xtreamApi.getShortEpg.mockReturnValue(
new Promise<EpgItem[]>((resolve) => {
resolveA = resolve;
})
);
const aPromise = priv().fetchEpg(credentials, 909);
// Mapping changes mid-flight; a re-enqueue (request B) then starts and
// takes ownership of the in-flight marker.
service.invalidate(909);
priv().inFlight.add(909);
resolveA([makeItem('rtl.de', 'Stale')]);
await aPromise;
// Request A was stale, so its finally must not clear B's marker.
expect(priv().inFlight.has(909)).toBe(true);
});
});
@@ -363,7 +363,14 @@ export class EpgQueueService implements OnDestroy {
error
);
} finally {
this.inFlight.delete(streamId);
// Only release the in-flight marker if this request still owns it.
// When invalidate() cleared it mid-flight, a later re-enqueue may
// already have started a new request for the same stream; an
// unconditional delete here would drop that request's marker and
// let a third concurrent fetch start.
if (!isStale()) {
this.inFlight.delete(streamId);
}
}
}