Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 13 additions & 3 deletions server/config.js
Original file line number Diff line number Diff line change
Expand Up @@ -276,11 +276,21 @@ router.post('/api/reload/channels', requireAuth, async (req, res) => {

router.post('/api/reload/epg', requireAuth, async (req, res) => {
try {
await refreshEPG();
notifyWebhooks('epg.refreshed', {}).catch(err =>
const outcome = await refreshEPG();
if (outcome.status === 'failed') {
return res.status(503).json({
error: outcome.message,
status: outcome.status,
sources: outcome.sourceResults,
});
}
notifyWebhooks('epg.refreshed', { status: outcome.status }).catch(err =>
console.warn('[Config] Webhook notification failed:', err.message)
);
res.json({ status: 'reloaded' });
res.status(outcome.status === 'degraded' ? 207 : 200).json({
status: outcome.status,
sources: outcome.sourceResults,
});
Comment on lines +290 to +293
} catch (e) {
res.status(500).json({
error: 'Failed to reload EPG',
Expand Down
100 changes: 89 additions & 11 deletions server/epg.js
Original file line number Diff line number Diff line change
Expand Up @@ -29,12 +29,31 @@ const epgLimiter = RateLimit({

// Module-level refresher that will be set when routes are initialized
let refreshImpl = null;
let refreshPromise = null;
let requestedRefreshRevision = 0;
let completedRefreshRevision = 0;
export async function refreshEPG() {
if (typeof refreshImpl === 'function') {
await refreshImpl();
} else {
if (typeof refreshImpl !== 'function') {
throw new Error('EPG refresher not initialized');
}

requestedRefreshRevision += 1;
if (!refreshPromise) {
refreshPromise = (async () => {
let outcome;
// Calls made while a refresh is in progress request one follow-up pass,
// ensuring an old channel/guide selection never becomes the final state.
while (completedRefreshRevision < requestedRefreshRevision) {
const revision = requestedRefreshRevision;
outcome = await refreshImpl(revision);
completedRefreshRevision = revision;
}
return outcome;
})().finally(() => {
refreshPromise = null;
});
}
return refreshPromise;
}

export function hasEPGRefresh() {
Expand All @@ -56,6 +75,8 @@ let epgCache = null;
// Module-level merged EPG XML string (set by setupEPGRoutes)
let mergedEPG = null;
let lastEPGSourceResults = [];
let lastEPGRefreshOutcome = null;
const lastGoodSourceXml = new Map();

// Maximum number of programmes returned per guide query (matches /api/guide behaviour)
const MAX_GUIDE_PROGRAMMES = 20;
Expand Down Expand Up @@ -203,6 +224,15 @@ export function getGuideData(tvgId, hours = 24) {
export function _resetMergedEPGForTesting() {
mergedEPG = null;
lastEPGSourceResults = [];
lastEPGRefreshOutcome = null;
lastGoodSourceXml.clear();
refreshPromise = null;
requestedRefreshRevision = 0;
completedRefreshRevision = 0;
}

export function getEPGRefreshOutcome() {
return lastEPGRefreshOutcome;
}

function loadOutputChannels(slug = '') {
Expand Down Expand Up @@ -341,7 +371,7 @@ export async function setupEPGRoutes(app) {
epgCache = cacheManager.createCache('epg', epgTTL);
console.log(`EPG cache initialized with TTL: ${epgTTL / 1000}s`);

async function fetchAndMergeEPGs() {
async function fetchAndMergeEPGs(revision) {
const startTime = Date.now();
const allChannels = loadChannelsForGuideMerge();
const { bindingsBySource: guideBindingsBySource, selectedCanonicalIds } =
Expand All @@ -364,6 +394,8 @@ export async function setupEPGRoutes(app) {

const fallbackTvgIds = new Set(fallbackChannels.map(ch => ch.tvg_id).filter(Boolean));
const fallbackNames = new Set(fallbackChannels.map(ch => ch.name).filter(Boolean));
let usedStaleSource = false;
let sourceFetchError = null;

try {
console.log(`Loading EPG: ${sourceName} (${sourceUrl})`);
Expand All @@ -375,7 +407,13 @@ export async function setupEPGRoutes(app) {
try {
xmlData = fs.readFileSync(path, 'utf-8');
} catch (fileErr) {
throw new Error(`Failed to read file: ${fileErr.message}`);
sourceFetchError = new Error(`Failed to read file: ${fileErr.message}`);
const lastGood = lastGoodSourceXml.get(sourceName);
if (!lastGood || lastGood.url !== sourceUrl) {
throw sourceFetchError;
}
Comment on lines 409 to +414
usedStaleSource = true;
xmlData = lastGood.xmlData;
}
} else {
try {
Expand All @@ -385,7 +423,13 @@ export async function setupEPGRoutes(app) {
});
xmlData = response.data;
} catch (httpErr) {
throw new Error(`Failed to fetch EPG: ${httpErr.message}`);
sourceFetchError = new Error(`Failed to fetch EPG: ${httpErr.message}`);
const lastGood = lastGoodSourceXml.get(sourceName);
if (!lastGood || lastGood.url !== sourceUrl) {
throw sourceFetchError;
}
usedStaleSource = true;
xmlData = lastGood.xmlData;
}
}

Expand All @@ -412,6 +456,10 @@ export async function setupEPGRoutes(app) {
throw new Error('Invalid XMLTV structure - missing <tv> root element');
}

if (!usedStaleSource) {
lastGoodSourceXml.set(sourceName, { url: sourceUrl, xmlData });
}

if (parsed.tv?.channel) {
const channels = []
.concat(parsed.tv.channel)
Expand Down Expand Up @@ -448,7 +496,8 @@ export async function setupEPGRoutes(app) {
sourceResults.push({
source: sourceName,
url: sourceUrl,
status: 'ok',
status: usedStaleSource ? 'stale' : 'ok',
...(sourceFetchError ? { error: sourceFetchError.message } : {}),
channelCount: channels.length,
programmeCount: 0,
});
Expand Down Expand Up @@ -491,7 +540,8 @@ export async function setupEPGRoutes(app) {
sourceResults.push({
source: sourceName,
url: sourceUrl,
status: 'ok',
status: usedStaleSource ? 'stale' : 'ok',
...(sourceFetchError ? { error: sourceFetchError.message } : {}),
channelCount: 0,
programmeCount: sourceProgrammeCount,
});
Expand Down Expand Up @@ -556,6 +606,24 @@ export async function setupEPGRoutes(app) {
}

lastEPGSourceResults = sourceResults;
const usableSources = sourceResults.filter(result => result.status !== 'error');
const staleSources = sourceResults.filter(result => result.status === 'stale');
const errors = sourceResults.filter(result => result.status === 'error');

if (usableSources.length === 0) {
const message = epgSources.length
? 'All configured EPG sources failed; retained the previous guide.'
: 'No EPG sources are configured.';
Comment on lines +614 to +616
lastEPGRefreshOutcome = {
status: 'failed',
revision,
message,
sourceResults,
refreshedAt: new Date().toISOString(),
};
console.error(`[EPG] ${message}`);
return lastEPGRefreshOutcome;
}

try {
mergedEPG = builder.build(merged);
Expand All @@ -573,17 +641,26 @@ export async function setupEPGRoutes(app) {
}

const duration = Date.now() - startTime;
lastEPGRefreshOutcome = {
status: errors.length || staleSources.length ? 'degraded' : 'ok',
revision,
message:
errors.length || staleSources.length
? 'Guide refreshed with stale or unavailable source data.'
: 'Guide refreshed successfully.',
sourceResults,
refreshedAt: new Date().toISOString(),
};
console.log(
`EPG merge completed in ${duration}ms (${merged.tv.channel.length} channels, ${merged.tv.programme.length} programmes)`
);
return lastEPGRefreshOutcome;
}

// Expose refresher
refreshImpl = fetchAndMergeEPGs;

await fetchAndMergeEPGs();
// unref() so the interval does not prevent the process from exiting cleanly
setInterval(fetchAndMergeEPGs, 6 * 60 * 60 * 1000).unref();
await refreshEPG();

const handleXmltvRequest = asyncHandler(async (req, res) => {
const profileSlug = req.params.slug ? String(req.params.slug).trim() : '';
Expand Down Expand Up @@ -669,6 +746,7 @@ export async function setupEPGRoutes(app) {
const sources = {
total: lastEPGSourceResults.length,
valid: lastEPGSourceResults.filter(source => source.status === 'ok').length,
stale: lastEPGSourceResults.filter(source => source.status === 'stale').length,
failed: lastEPGSourceResults.filter(source => source.status === 'error').length,
results: lastEPGSourceResults,
};
Expand Down
22 changes: 18 additions & 4 deletions server/mcp.js
Original file line number Diff line number Diff line change
Expand Up @@ -1041,13 +1041,27 @@ function createMcpServer() {
{},
async () => {
try {
await refreshEPG();
const outcome = await refreshEPG();
if (outcome.status === 'failed') {
return createError('reload_epg', {
code: 'epg-reload-failed',
message: outcome.message,
nextSuggestedTools: ['get_status', 'list_guide_bindings'],
});
}
return createSuccess(
'reload_epg',
{ triggered: true },
{ triggered: true, status: outcome.status, sources: outcome.sourceResults },
{
summary: 'EPG reload triggered successfully.',
sideEffects: ['Guide data was reloaded from configured XMLTV sources.'],
summary:
outcome.status === 'degraded'
? 'EPG reload completed with stale source data retained.'
: 'EPG reload completed successfully.',
sideEffects: [
outcome.status === 'degraded'
? 'Guide data was rebuilt while retaining last-known-good data for failed sources.'
: 'Guide data was reloaded from configured XMLTV sources.',
],
nextSuggestedTools: ['get_status', 'get_guide', 'diagnose_agent_readiness'],
}
);
Expand Down
8 changes: 7 additions & 1 deletion server/scheduler.js
Original file line number Diff line number Diff line change
Expand Up @@ -177,7 +177,13 @@ export function initDefaultJobs() {
registerJob(
'EPG Refresh',
'0 */6 * * *',
refreshEPG,
async () => {
const outcome = await refreshEPG();
if (outcome.status === 'failed') {
throw new Error(outcome.message);
}
return outcome;
},
false // EPG is already loaded during setupEPGRoutes
);

Expand Down
34 changes: 33 additions & 1 deletion test/integration/guide.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,8 @@ describe('GET /api/guide', () => {
let epgFilePath;
let server;
let baseUrl;
let refreshEPG;
let resetMergedEPG;
before(async () => {
tmpDataDir = await fs.mkdtemp(path.join(os.tmpdir(), 'iptv-guide-data-'));
process.env.DATA_PATH = tmpDataDir;
Expand Down Expand Up @@ -109,7 +111,10 @@ describe('GET /api/guide', () => {
const app = express();
app.use(express.json());

const { setupEPGRoutes } = await import('../../server/epg.js');
const epgModule = await import('../../server/epg.js');
const { setupEPGRoutes } = epgModule;
refreshEPG = epgModule.refreshEPG;
resetMergedEPG = epgModule._resetMergedEPGForTesting;
const { errorHandler } = await import('../../server/error-handler.js');

await setupEPGRoutes(app);
Expand Down Expand Up @@ -173,4 +178,31 @@ describe('GET /api/guide', () => {
expect(res.data.programmes).to.be.an('array').with.lengthOf(0);
expect(res.data.total).to.equal(0);
});

it('retains last-known-good guide data on source failure and reports recovery', async () => {
await fs.unlink(epgFilePath);

const degraded = await refreshEPG();
expect(degraded.status).to.equal('degraded');
const staleSource = degraded.sourceResults.find(source => source.source === 'GuideProvider');
expect(staleSource).to.include({
url: pathToFileURL(epgFilePath).href,
status: 'stale',
channelCount: 2,
programmeCount: 3,
});
expect(staleSource.error).to.be.a('string');

const retained = await axios.get(`${baseUrl}/api/guide?tvgId=${TVG_ID}`);
expect(retained.data.programmes.map(programme => programme.title)).to.include('Current Show');

await fs.writeFile(epgFilePath, buildFutureXMLTV(TVG_ID), 'utf8');
const recovered = await refreshEPG();
expect(recovered.status).to.equal('ok');

resetMergedEPG();
await fs.unlink(epgFilePath);
const failed = await refreshEPG();
expect(failed.status).to.equal('failed');
});
});
Loading