diff --git a/admin/src/App.vue b/admin/src/App.vue index 17eb071..948d3cc 100644 --- a/admin/src/App.vue +++ b/admin/src/App.vue @@ -2445,178 +2445,42 @@ async function saveOutputProfileChanges() { setChannelRowFeedback(channelId, 'accent', 'Saving changes…'); }); - if (profileMetaDirty.value) { - const profileResponse = await apiFetch( - `/api/output-profiles/${encodeURIComponent(state.selectedOutputProfileSlug)}`, - { - method: 'PATCH', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - name: state.outputProfileDraft.name.trim(), - enabled: state.outputProfileDraft.enabled, - }), - } - ); - const profileJson = await profileResponse.json(); - if (!profileResponse.ok) { - throw new Error(buildApiErrorMessage(profileJson, 'Failed to save output profile details')); - } - } - - for (const row of canonicalRowsToSave) { - if (row.customNameDirty) { - state.updatingCanonicalNameChannelId = row.id; - const response = await apiFetch(`/api/canonical/channels/${encodeURIComponent(row.id)}`, { - method: 'PATCH', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - customName: - typeof row.customNameDraft === 'string' && row.customNameDraft.trim() - ? row.customNameDraft.trim() - : null, - }), - }); - const json = await response.json(); - if (!response.ok) { - throw Object.assign(new Error(buildApiErrorMessage(json, 'Failed to update channel name')), { - channelId: row.id, - }); - } - - const updatedChannel = { - ...json.channel, - customNameDraft: json.channel?.customName || '', - }; - state.canonicalChannels = state.canonicalChannels.map(entry => - entry.id === row.id ? updatedChannel : entry - ); - state.channelBindings = state.channelBindings.map(binding => - binding.canonical?.id === row.id - ? { - ...binding, - canonical: { - ...binding.canonical, - name: updatedChannel.name, - baseName: updatedChannel.baseName, - customName: updatedChannel.customName, - }, - } - : binding - ); - state.guideBindings = state.guideBindings.map(binding => - binding.canonical?.id === row.id - ? { - ...binding, - canonical: { - ...binding.canonical, - name: updatedChannel.name, - baseName: updatedChannel.baseName, - customName: updatedChannel.customName, - }, - } - : binding - ); - state.outputProfileEntries = state.outputProfileEntries.map(entry => - entry.canonical?.id === row.id + const response = await apiFetch( + `/api/output-profiles/${encodeURIComponent(state.selectedOutputProfileSlug)}`, + { + method: 'PUT', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + profile: profileMetaDirty.value ? { - ...entry, - canonical: { - ...entry.canonical, - name: updatedChannel.name, - baseName: updatedChannel.baseName, - customName: updatedChannel.customName, - }, + name: state.outputProfileDraft.name.trim(), + enabled: state.outputProfileDraft.enabled, } - : entry - ); - } - - if (row.preferredStreamDirty) { - state.updatingPreferredStreamChannelId = row.id; - const response = await apiFetch( - `/api/canonical/channels/${encodeURIComponent(row.id)}/preferred-stream`, - { - method: 'PATCH', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ sourceChannelId: row.preferredSourceChannelId }), - } - ); - const json = await response.json(); - if (!response.ok) { - throw Object.assign( - new Error(buildApiErrorMessage(json, 'Failed to update preferred stream')), - { channelId: row.id } - ); - } - - state.channelBindings = state.channelBindings.map(binding => ({ - ...binding, - isPreferredStream: - binding.canonical?.id === row.id ? binding.id === json.binding.id : binding.isPreferredStream, - })); - } - - if (row.guideBindingDirty) { - state.updatingGuideBindingChannelId = row.id; - const response = await apiFetch( - `/api/canonical/channels/${encodeURIComponent(row.id)}/guide-binding`, - { - method: 'PATCH', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify(parseGuideBindingValue(row.selectedGuideBindingValue)), - } - ); - const json = await response.json(); - if (!response.ok) { - throw Object.assign( - new Error(buildApiErrorMessage(json, 'Failed to update guide binding')), - { channelId: row.id } - ); - } - - state.guideBindings = state.guideBindings.map(entry => { - if (entry.canonical?.id !== row.id) { - return entry; - } - - if (entry.source?.id === json.binding.source.id) { - return { - ...entry, - epgChannelId: json.binding.epgChannelId, - priority: json.binding.priority, - selected: true, - }; - } - - return { - ...entry, - selected: false, - priority: entry.priority === 0 ? 1 : entry.priority, - }; - }); - } - } - - if (outputProfileEntriesDirty.value) { - const response = await apiFetch( - `/api/output-profiles/${encodeURIComponent(state.selectedOutputProfileSlug)}/channels`, - { - method: 'PATCH', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - channels: channelWorkflowRows.value.map(row => ({ + : undefined, + canonicalChannels: canonicalRowsToSave.map(row => ({ + id: row.id, + ...(row.customNameDirty ? { customName: row.customNameDraft } : {}), + ...(row.preferredStreamDirty + ? { preferredSourceChannelId: row.preferredSourceChannelId } + : {}), + ...(row.guideBindingDirty + ? { guideBinding: parseGuideBindingValue(row.selectedGuideBindingValue) } + : {}), + })), + channels: outputProfileEntriesDirty.value + ? channelWorkflowRows.value.map(row => ({ canonicalId: row.id, position: row.position, enabled: row.outputEnabled, guideNumberOverride: row.committedGuideNumberOverride, - })), - }), - } - ); - const json = await response.json(); - if (!response.ok) { - throw new Error(buildApiErrorMessage(json, 'Failed to save output profile channels')); + })) + : undefined, + }), } + ); + const json = await response.json(); + if (!response.ok) { + throw new Error(buildApiErrorMessage(json, 'Failed to save output profile')); } await loadChannelAuthoringData(); diff --git a/libs/cache-manager.js b/libs/cache-manager.js index 4ee4805..62e84e2 100644 --- a/libs/cache-manager.js +++ b/libs/cache-manager.js @@ -14,8 +14,8 @@ class CacheManager { * @param {number} ttl - Time to live in milliseconds (0 = no expiration) * @returns {Cache} Cache instance */ - createCache(name, ttl = 0, maxEntries = Infinity) { - const cache = new Cache(name, ttl, maxEntries); + createCache(name, ttl = 0, maxEntries = Infinity, maxBytes = Infinity) { + const cache = new Cache(name, ttl, maxEntries, maxBytes); this.caches.set(name, cache); return cache; } @@ -52,14 +52,31 @@ class CacheManager { } class Cache { - constructor(name, ttl = 0, maxEntries = Infinity) { + constructor(name, ttl = 0, maxEntries = Infinity, maxBytes = Infinity) { this.name = name; this.ttl = ttl; // TTL in milliseconds this.maxEntries = maxEntries; + this.maxBytes = maxBytes; + this.byteSize = 0; this.data = new Map(); this.timestamps = new Map(); this.hits = 0; this.misses = 0; + // Expiry must not depend on a later read of the same cache key. The + // timer is unref'd so a cache never keeps the service/test process alive. + this.cleanupTimer = + this.ttl > 0 + ? setInterval(() => this.purgeExpired(), Math.max(1000, Math.min(this.ttl, 60000))).unref() + : null; + } + + estimateBytes(value) { + if (typeof value === 'string') return Buffer.byteLength(value); + try { + return Buffer.byteLength(JSON.stringify(value)); + } catch { + return 0; + } } /** @@ -71,12 +88,14 @@ class Cache { // Map insertion order gives us a small, dependency-free LRU cache. Refresh // an existing entry so frequently used variants are not evicted first. if (this.data.has(key)) { + this.byteSize -= this.estimateBytes(this.data.get(key)); this.data.delete(key); this.timestamps.delete(key); } this.data.set(key, value); + this.byteSize += this.estimateBytes(value); this.timestamps.set(key, Date.now()); - while (this.data.size > this.maxEntries) { + while (this.data.size > this.maxEntries || this.byteSize > this.maxBytes) { const oldestKey = this.data.keys().next().value; this.delete(oldestKey); } @@ -143,6 +162,7 @@ class Cache { * @param {string} key - Cache key */ delete(key) { + if (this.data.has(key)) this.byteSize -= this.estimateBytes(this.data.get(key)); this.data.delete(key); this.timestamps.delete(key); } @@ -153,6 +173,15 @@ class Cache { clear() { this.data.clear(); this.timestamps.clear(); + this.byteSize = 0; + } + + purgeExpired() { + if (this.ttl <= 0) return; + const now = Date.now(); + for (const [key, timestamp] of this.timestamps) { + if (now - timestamp > this.ttl) this.delete(key); + } } /** @@ -162,12 +191,7 @@ class Cache { size() { // Clean up expired entries before counting if (this.ttl > 0) { - const now = Date.now(); - for (const [key, timestamp] of this.timestamps.entries()) { - if (now - timestamp > this.ttl) { - this.delete(key); - } - } + this.purgeExpired(); } return this.data.size; } @@ -186,6 +210,8 @@ class Cache { size, ttl: this.ttl, maxEntries: this.maxEntries, + maxBytes: this.maxBytes, + bytes: this.byteSize, hits: this.hits, misses: this.misses, hitRate: `${hitRate}%`, diff --git a/libs/output-profile-service.js b/libs/output-profile-service.js index b0c837d..29df366 100644 --- a/libs/output-profile-service.js +++ b/libs/output-profile-service.js @@ -510,6 +510,142 @@ export function updateOutputProfileEntries(slug = DEFAULT_PROFILE_SLUG, channels return listOutputProfileEntries(slug); } +/** + * Apply all edits made by the output-profile authoring screen in one database + * transaction. Keeping this at the service boundary lets HTTP and MCP callers + * invalidate/refresh derived guide state once, after every row is committed. + * + * @param {string} slug + * @param {{profile?: {name?: string, enabled?: boolean}, canonicalChannels?: Array, channels?: Array}} changes + * @returns {{profile?: object, channels?: Array}|{error: string, canonicalId?: string}} + */ +export function saveOutputProfile(slug = DEFAULT_PROFILE_SLUG, changes = {}) { + ensureDatabaseReady(); + ensureDefaultOutputProfile(); + + const db = getDatabase(); + const profile = getProfileBySlug(slug); + if (!profile) return { error: 'profile-not-found' }; + + const profileChanges = changes.profile || {}; + const nextName = + profileChanges.name === undefined ? profile.name : String(profileChanges.name || '').trim(); + if (!nextName) return { error: 'invalid-name' }; + const nextEnabled = + profileChanges.enabled === undefined ? Boolean(profile.enabled) : Boolean(profileChanges.enabled); + if (profile.slug === DEFAULT_PROFILE_SLUG && !nextEnabled) { + return { error: 'default-profile-required' }; + } + + const canonicalChanges = Array.isArray(changes.canonicalChannels) ? changes.canonicalChannels : []; + const channels = Array.isArray(changes.channels) ? changes.channels : []; + const canonicalById = new Map( + db + .prepare('SELECT id FROM canonical_channels') + .all() + .map(row => [row.id, row]) + ); + const entryRows = db + .prepare( + `SELECT opc.canonical_channel_id, cc.guide_number AS canonical_guide_number + FROM output_profile_channels opc + JOIN canonical_channels cc ON cc.id = opc.canonical_channel_id + WHERE opc.output_profile_id = ?` + ) + .all(profile.id); + const entryByCanonicalId = new Map(entryRows.map(row => [row.canonical_channel_id, row])); + + for (const change of canonicalChanges) { + if (!change?.id || !canonicalById.has(change.id)) { + return { error: 'canonical-not-found', canonicalId: change?.id }; + } + if (change.preferredSourceChannelId !== undefined) { + const binding = db + .prepare( + 'SELECT id FROM channel_bindings WHERE canonical_channel_id = ? AND source_channel_id = ?' + ) + .get(change.id, change.preferredSourceChannelId); + if (!binding) return { error: 'binding-not-found', canonicalId: change.id }; + } + if (change.guideBinding !== undefined) { + const binding = db + .prepare( + 'SELECT id FROM guide_bindings WHERE canonical_channel_id = ? AND source_id = ?' + ) + .get(change.id, change.guideBinding?.sourceId); + if (!binding || !String(change.guideBinding?.epgChannelId || '').trim()) { + return { error: 'guide-binding-not-found', canonicalId: change.id }; + } + } + } + + for (const channel of channels) { + if (!channel?.canonicalId || !entryByCanonicalId.has(channel.canonicalId)) { + return { error: 'entry-not-found', canonicalId: channel?.canonicalId }; + } + if (!Number.isInteger(channel.position) || channel.position < 0 || typeof channel.enabled !== 'boolean') { + return { error: 'invalid-entry', canonicalId: channel.canonicalId }; + } + if (channel.guideNumberOverride !== null && channel.guideNumberOverride !== undefined && typeof channel.guideNumberOverride !== 'string') { + return { error: 'invalid-entry', canonicalId: channel.canonicalId }; + } + } + + transaction(() => { + db.prepare('UPDATE output_profiles SET name = ?, enabled = ?, updated_at = ? WHERE id = ?').run( + nextName, + nextEnabled ? 1 : 0, + new Date().toISOString(), + profile.id + ); + const updateName = db.prepare('UPDATE canonical_channels SET custom_name = ?, updated_at = ? WHERE id = ?'); + const updatePreferred = db.prepare( + 'UPDATE channel_bindings SET is_preferred_stream = CASE WHEN source_channel_id = ? THEN 1 ELSE 0 END WHERE canonical_channel_id = ?' + ); + const updateGuide = db.prepare( + 'UPDATE guide_bindings SET epg_channel_id = CASE WHEN source_id = ? THEN ? ELSE epg_channel_id END, priority = CASE WHEN source_id = ? THEN 0 ELSE 1 END WHERE canonical_channel_id = ?' + ); + for (const change of canonicalChanges) { + if (Object.hasOwn(change, 'customName')) { + updateName.run( + typeof change.customName === 'string' && change.customName.trim() + ? change.customName.trim() + : null, + new Date().toISOString(), + change.id + ); + } + if (change.preferredSourceChannelId !== undefined) updatePreferred.run(change.preferredSourceChannelId, change.id); + if (change.guideBinding !== undefined) { + updateGuide.run( + change.guideBinding.sourceId, + change.guideBinding.epgChannelId, + change.guideBinding.sourceId, + change.id + ); + } + } + const updateEntry = db.prepare( + 'UPDATE output_profile_channels SET position = ?, guide_number_override = ?, enabled = ? WHERE output_profile_id = ? AND canonical_channel_id = ?' + ); + for (const channel of channels) { + const effectiveGuideNumber = resolveEffectiveGuideNumber( + { guideNumberOverride: channel.guideNumberOverride }, + { guide_number: entryByCanonicalId.get(channel.canonicalId).canonical_guide_number } + ); + updateEntry.run( + channel.position, + channel.guideNumberOverride || null, + channel.enabled && effectiveGuideNumber ? 1 : 0, + profile.id, + channel.canonicalId + ); + } + })(); + + return { profile: getOutputProfile(slug), channels: listOutputProfileEntries(slug) }; +} + export function createOutputProfile({ name, copyFromSlug = null, enabled = true } = {}) { ensureDatabaseReady(); @@ -648,6 +784,7 @@ export default { listOutputProfiles, syncAllOutputProfiles, syncDefaultOutputProfile, + saveOutputProfile, updateOutputProfile, updateOutputProfileEntries, }; diff --git a/server/canonical.js b/server/canonical.js index 285d443..1901481 100644 --- a/server/canonical.js +++ b/server/canonical.js @@ -16,6 +16,7 @@ import { getOutputProfileChannels, listOutputProfileEntries, listOutputProfiles, + saveOutputProfile, syncAllOutputProfiles, updateOutputProfileEntries, updateOutputProfile, @@ -376,4 +377,44 @@ router.patch('/api/output-profiles/:slug/channels', requireAuth, async (req, res } }); +// The authoring UI may change profile metadata, canonical channel choices, and +// output entries together. Save them atomically, then rebuild the guide once. +router.put('/api/output-profiles/:slug', requireAuth, async (req, res) => { + try { + await ensureCanonicalModelReady(); + const result = saveOutputProfile(req.params.slug, { + profile: req.body?.profile, + canonicalChannels: req.body?.canonicalChannels, + channels: req.body?.channels, + }); + if (result?.error === 'profile-not-found') { + return res.status(404).json({ error: 'Output profile not found' }); + } + if (result?.error === 'default-profile-required') { + return res.status(400).json({ error: 'Default output profile must remain enabled' }); + } + if (result?.error === 'invalid-name') { + return res.status(400).json({ error: 'name must be a non-empty string' }); + } + if (result?.error === 'canonical-not-found') { + return res.status(404).json({ error: `Canonical channel not found: ${result.canonicalId}` }); + } + if (result?.error === 'binding-not-found' || result?.error === 'guide-binding-not-found') { + return res.status(404).json({ error: `Channel binding not found for canonical channel ${result.canonicalId}` }); + } + if (result?.error === 'entry-not-found') { + return res.status(404).json({ error: `Output profile entry not found for canonical channel ${result.canonicalId}` }); + } + if (result?.error === 'invalid-entry') { + return res.status(400).json({ error: `Invalid output profile entry for canonical channel ${result.canonicalId}` }); + } + + invalidateLineupCaches(); + if (hasEPGRefresh()) await refreshEPG(); + res.json({ status: 'saved', ...result }); + } catch (err) { + res.status(500).json({ error: 'Failed to save output profile', detail: err.message }); + } +}); + export default router; diff --git a/server/epg.js b/server/epg.js index c5a03d1..e60240c 100644 --- a/server/epg.js +++ b/server/epg.js @@ -3,6 +3,7 @@ import { fileURLToPath } from 'url'; import axios from 'axios'; import RateLimit from 'express-rate-limit'; import { XMLParser, XMLBuilder } from 'fast-xml-parser'; +import { decodeXmltvStream, readXmltvRecords } from './xmltv-stream.js'; import { loadAppConfigFromStore } from '../libs/app-settings-service.js'; import { getChannels } from '../libs/channels-cache.js'; import { asyncHandler, AppError } from './error-handler.js'; @@ -79,6 +80,7 @@ let mergedEPG = null; let lastEPGSourceResults = []; let lastEPGRefreshOutcome = null; const lastGoodSourceXml = new Map(); +let guideProgrammeIndex = new Map(); // Maximum number of programmes returned per guide query (matches /api/guide behaviour) const MAX_GUIDE_PROGRAMMES = 20; @@ -86,6 +88,8 @@ const MAX_GUIDE_PROGRAMMES = 20; // number of request variants so arbitrary filter combinations cannot retain // unbounded copies of the guide in memory. const MAX_EPG_CACHE_ENTRIES = 32; +const MAX_EPG_CACHE_BYTES = 16 * 1024 * 1024; +const MAX_XMLTV_INPUT_BYTES = 64 * 1024 * 1024; /** * Extract a plain string from an XMLTV text field which may be a raw string, @@ -185,15 +189,12 @@ export function getGuideData(tvgId, hours = 24) { if (!mergedEPG) return null; const clampedHours = Math.min(48, Math.max(1, hours)); - const parsed = parser.parse(mergedEPG); const now = Date.now(); const cutoff = now + clampedHours * 60 * 60 * 1000; - let programmes = [].concat(parsed.tv?.programme || []); - - if (tvgId) { - programmes = programmes.filter(p => p && p['@_channel'] === tvgId); - } + let programmes = tvgId + ? guideProgrammeIndex.get(tvgId) || [] + : Array.from(guideProgrammeIndex.values()).flat(); programmes = programmes.filter(p => { const start = parseXMLTVDate(p['@_start']); @@ -226,6 +227,7 @@ export function getGuideData(tvgId, hours = 24) { */ export function _resetMergedEPGForTesting() { mergedEPG = null; + guideProgrammeIndex = new Map(); lastEPGSourceResults = []; lastEPGRefreshOutcome = null; lastGoodSourceXml.clear(); @@ -241,6 +243,7 @@ export function getEPGRefreshOutcome() { /** Clear derived guide data after a state restore. */ export function invalidateEPGCache() { mergedEPG = null; + guideProgrammeIndex = new Map(); epgCache = null; lastEPGSourceResults = []; } @@ -424,7 +427,12 @@ export async function setupEPGRoutes(app) { // Initialize EPG cache with TTL from config (default: 6 hours) const epgTTL = (appConfig.cache?.epg_ttl ?? 21600) * 1000; // Convert seconds to milliseconds - epgCache = cacheManager.createCache('epg', epgTTL, MAX_EPG_CACHE_ENTRIES); + epgCache = cacheManager.createCache( + 'epg', + epgTTL, + MAX_EPG_CACHE_ENTRIES, + MAX_EPG_CACHE_BYTES + ); console.log(`EPG cache initialized with TTL: ${epgTTL / 1000}s`); async function fetchAndMergeEPGs(revision) { @@ -462,12 +470,16 @@ export async function setupEPGRoutes(app) { try { console.log(`Loading EPG: ${sourceName} (${sourceUrl})`); - let xmlData; + let sourceRecords; if (sourceUrl.startsWith('file://')) { const path = fileURLToPath(sourceUrl); try { - xmlData = fs.readFileSync(path, 'utf-8'); + const compressed = /\.gz$/i.test(path); + sourceRecords = await readXmltvRecords( + decodeXmltvStream(fs.createReadStream(path), compressed), + { maxBytes: MAX_XMLTV_INPUT_BYTES } + ); } catch (fileErr) { sourceFetchError = new Error(`Failed to read file: ${fileErr.message}`); const lastGood = lastGoodSourceXml.get(sourceName); @@ -475,15 +487,22 @@ export async function setupEPGRoutes(app) { throw sourceFetchError; } usedStaleSource = true; - xmlData = lastGood.xmlData; + sourceRecords = lastGood.sourceRecords; } } else { try { const response = await axios.get(sourceUrl, { timeout: 15000, validateStatus: status => status === 200, + responseType: 'stream', + decompress: false, }); - xmlData = response.data; + const encoding = String(response.headers['content-encoding'] || '').toLowerCase(); + const compressed = encoding.includes('gzip') || /\.gz(?:$|\?)/i.test(sourceUrl); + sourceRecords = await readXmltvRecords( + decodeXmltvStream(response.data, compressed), + { maxBytes: MAX_XMLTV_INPUT_BYTES } + ); } catch (httpErr) { sourceFetchError = new Error(`Failed to fetch EPG: ${httpErr.message}`); const lastGood = lastGoodSourceXml.get(sourceName); @@ -491,40 +510,21 @@ export async function setupEPGRoutes(app) { throw sourceFetchError; } usedStaleSource = true; - xmlData = lastGood.xmlData; + sourceRecords = lastGood.sourceRecords; } } - // Validate that we got XML data - if (!xmlData || typeof xmlData !== 'string' || xmlData.trim().length === 0) { + if (!sourceRecords || (!sourceRecords.channels.length && !sourceRecords.programmes.length)) { throw new Error('Empty or invalid EPG data received'); } - // Validate basic XML structure (allow whitespace/comments before declaration) - const trimmedXml = xmlData.trim(); - if (!trimmedXml.includes(' root element'); - } - if (!usedStaleSource) { - lastGoodSourceXml.set(sourceName, { url: sourceUrl, xmlData }); + lastGoodSourceXml.set(sourceName, { url: sourceUrl, sourceRecords }); } - if (parsed.tv?.channel) { + if (sourceRecords.channels.length) { const channels = [] - .concat(parsed.tv.channel) + .concat(sourceRecords.channels) .flatMap(channel => { if (!channel) { return []; @@ -568,9 +568,9 @@ export async function setupEPGRoutes(app) { } let sourceProgrammeCount = 0; - if (parsed.tv?.programme) { + if (sourceRecords.programmes.length) { const programmes = [] - .concat(parsed.tv.programme) + .concat(sourceRecords.programmes) .flatMap(programme => { if (!programme) { return []; @@ -674,6 +674,13 @@ export async function setupEPGRoutes(app) { const deduplicated = deduplicateXmltvRecords(merged); merged.tv.channel = deduplicated.tv.channel; merged.tv.programme = deduplicated.tv.programme; + guideProgrammeIndex = new Map(); + for (const programme of merged.tv.programme) { + const channelId = programme?.['@_channel']; + if (!channelId) continue; + if (!guideProgrammeIndex.has(channelId)) guideProgrammeIndex.set(channelId, []); + guideProgrammeIndex.get(channelId).push(programme); + } lastEPGSourceResults = sourceResults; const usableSources = sourceResults.filter(result => result.status !== 'error'); diff --git a/server/mcp.js b/server/mcp.js index ec7e4bd..aa8478b 100644 --- a/server/mcp.js +++ b/server/mcp.js @@ -22,6 +22,7 @@ import { getOutputProfileChannels, listOutputProfileEntries, listOutputProfiles, + saveOutputProfile, syncAllOutputProfiles, updateOutputProfileEntries, updateOutputProfile, @@ -114,6 +115,7 @@ function buildWorkflowOverview() { mutation: [ 'create_output_profile', 'update_output_profile', + 'save_output_profile', 'delete_output_profile', 'set_canonical_channel_name', 'set_canonical_channel_published', @@ -162,6 +164,7 @@ function buildWorkflowOverview() { 'set_canonical_channel_preferred_stream', 'set_canonical_channel_guide_binding', 'update_output_profile_channels', + 'save_output_profile', ], }, { @@ -174,6 +177,10 @@ function buildWorkflowOverview() { mutationSideEffects: { create_output_profile: ['Creates profile state and syncs initial output entries.'], update_output_profile: ['Updates profile metadata only.'], + save_output_profile: [ + 'Atomically updates profile metadata, canonical choices, and output entries.', + 'Invalidates lineup caches and triggers at most one EPG refresh.', + ], delete_output_profile: ['Removes profile state for that lineup.'], set_canonical_channel_name: [ 'Updates the canonical channel display name override.', @@ -651,6 +658,81 @@ function createMcpServer() { } ); + registerJsonTool( + server, + 'save_output_profile', + 'Atomically save output profile metadata, canonical choices, and channel entries with at most one EPG refresh.', + { + slug: z.string().default('default').describe('Output profile slug (default: "default")'), + profile: z + .object({ name: z.string().min(1).optional(), enabled: z.boolean().optional() }) + .optional(), + canonical_channels: z + .array( + z.object({ + id: z.string().min(1), + custom_name: z.string().nullable().optional(), + preferred_source_channel_id: z.string().min(1).optional(), + guide_binding: z + .object({ source_id: z.string().min(1), epg_channel_id: z.string().min(1) }) + .optional(), + }) + ) + .optional(), + channels: z + .array( + z.object({ + canonicalId: z.string().min(1), + position: z.number().int().min(0), + enabled: z.boolean(), + guideNumberOverride: z.union([z.string(), z.null()]).optional(), + }) + ) + .optional(), + }, + async ({ slug = 'default', profile, canonical_channels: canonicalChannels, channels }) => { + const result = saveOutputProfile(slug, { + profile, + canonicalChannels: (canonicalChannels || []).map(channel => ({ + id: channel.id, + customName: channel.custom_name, + preferredSourceChannelId: channel.preferred_source_channel_id, + guideBinding: channel.guide_binding + ? { + sourceId: channel.guide_binding.source_id, + epgChannelId: channel.guide_binding.epg_channel_id, + } + : undefined, + })), + channels, + }); + if (result?.error) { + return createError('save_output_profile', { + code: result.error, + message: `Failed to save output profile: ${result.error}`, + details: { slug, canonicalId: result.canonicalId }, + nextSuggestedTools: ['list_output_profiles', 'list_output_profile_entries'], + }); + } + + invalidateLineupCaches(); + if (hasEPGRefresh()) await refreshEPG(); + return createSuccess('save_output_profile', result, { + summary: `Saved output profile ${slug} atomically.`, + sideEffects: [ + 'Output profile and channel state were updated atomically.', + 'Lineup caches were invalidated.', + ...epgRefreshSideEffect(), + ], + nextSuggestedTools: [ + 'list_output_profile_entries', + 'get_output_profile_channels', + 'diagnose_agent_readiness', + ], + }); + } + ); + registerJsonTool( server, 'delete_output_profile', diff --git a/server/xmltv-stream.js b/server/xmltv-stream.js new file mode 100644 index 0000000..6e379bd --- /dev/null +++ b/server/xmltv-stream.js @@ -0,0 +1,95 @@ +import { createGunzip } from 'node:zlib'; +import { XMLParser } from 'fast-xml-parser'; + +const fragmentParser = new XMLParser({ ignoreAttributes: false, attributeNamePrefix: '@_' }); + +/** + * Decode a gzip XMLTV stream when the caller has identified it as compressed. + * Piping preserves Node's normal backpressure semantics; consumers can await + * each chunk instead of accumulating the response in a string. + * + * @param {import('node:stream').Readable} input + * @param {boolean} compressed + */ +export function decodeXmltvStream(input, compressed = false) { + if (!compressed) return input; + const gunzip = createGunzip(); + input.pipe(gunzip); + return gunzip; +} + +/** + * Read only complete XMLTV channel/programme elements from a readable stream. + * This deliberately does not construct a document-wide parser object: the + * retained memory is the selected records plus one currently-open element. + * XMLTV uses non-nested channel/programme records, so element framing is safe + * here and lets fast-xml-parser keep handling XML entities and child fields. + * + * @param {import('node:stream').Readable} input + * @param {{maxBytes?: number, maxElementBytes?: number}} [limits] + * @returns {Promise<{channels: Array, programmes: Array, bytes: number}>} + */ +export async function readXmltvRecords(input, limits = {}) { + const maxBytes = limits.maxBytes ?? 64 * 1024 * 1024; + const maxElementBytes = limits.maxElementBytes ?? 1024 * 1024; + const channels = []; + const programmes = []; + let bytes = 0; + let buffer = ''; + let activeType = null; + let sawTvRoot = false; + + for await (const chunk of input) { + bytes += Buffer.byteLength(chunk); + if (bytes > maxBytes) { + throw new Error(`XMLTV input exceeds the ${maxBytes}-byte limit`); + } + buffer += Buffer.isBuffer(chunk) ? chunk.toString('utf8') : String(chunk); + if (/])/i.test(buffer)) sawTvRoot = true; + + while (buffer.length > 0) { + if (!activeType) { + const match = /<(channel|programme)(?=[\s>])/i.exec(buffer); + if (!match) { + // Retain a small suffix in case the next chunk finishes a tag name. + buffer = buffer.slice(-16); + break; + } + buffer = buffer.slice(match.index); + activeType = match[1].toLowerCase(); + } + + const closeTag = ``; + const openingEnd = buffer.indexOf('>'); + const selfClosing = openingEnd !== -1 && /\/\s*$/.test(buffer.slice(0, openingEnd)); + const closeIndex = selfClosing ? -1 : buffer.toLowerCase().indexOf(closeTag); + const endIndex = selfClosing ? openingEnd + 1 : closeIndex === -1 ? -1 : closeIndex + closeTag.length; + + if (endIndex === -1) { + if (Buffer.byteLength(buffer) > maxElementBytes) { + throw new Error(`XMLTV ${activeType} element exceeds the ${maxElementBytes}-byte limit`); + } + break; + } + + const fragment = buffer.slice(0, endIndex); + buffer = buffer.slice(endIndex); + const type = activeType; + activeType = null; + let parsed; + try { + parsed = fragmentParser.parse(`${fragment}`); + } catch (error) { + throw new Error(`XMLTV ${type} parsing failed: ${error.message}`); + } + const record = parsed?.tv?.[type]; + if (record) (type === 'channel' ? channels : programmes).push(record); + } + } + + if (!sawTvRoot) throw new Error('Invalid XMLTV structure - missing root element'); + if (activeType) { + throw new Error(`Incomplete XMLTV ${activeType} element`); + } + return { channels, programmes, bytes }; +} diff --git a/test/integration/mcp.test.js b/test/integration/mcp.test.js index f788fb3..02953c1 100644 --- a/test/integration/mcp.test.js +++ b/test/integration/mcp.test.js @@ -187,6 +187,7 @@ describe('MCP Route Integration', () => { 'list_output_profiles', 'create_output_profile', 'update_output_profile', + 'save_output_profile', 'delete_output_profile', 'get_output_profile_channels', 'list_output_profile_entries', @@ -329,6 +330,16 @@ describe('MCP Route Integration', () => { expect(profiles).to.be.an('array'); }); + it('save_output_profile atomically saves profile state through MCP', async () => { + const { msg } = await mcpPost(baseUrl, 'tools/call', { + name: 'save_output_profile', + arguments: { slug: 'default', profile: { name: 'MCP Default', enabled: true } }, + }); + const result = expectToolSuccess(msg, 'save_output_profile').data; + expect(result.profile).to.include({ slug: 'default', name: 'MCP Default', enabled: true }); + expect(result.channels).to.be.an('array'); + }); + it('create_output_profile, update_output_profile, and delete_output_profile manage named profiles', async () => { const createResult = await mcpPost(baseUrl, 'tools/call', { name: 'create_output_profile', diff --git a/test/integration/output-profile-routes.test.js b/test/integration/output-profile-routes.test.js index 434687d..07bef37 100644 --- a/test/integration/output-profile-routes.test.js +++ b/test/integration/output-profile-routes.test.js @@ -258,6 +258,51 @@ describe('output profile routes', () => { }); }); + it('atomically saves profile metadata, canonical edits, and output entries', async () => { + const entriesResponse = await axios.get(`${baseUrl}/api/output-profiles/default/entries`); + const [first, second] = entriesResponse.data.channels; + + const response = await axios.put(`${baseUrl}/api/output-profiles/default`, { + profile: { name: 'Living Room', enabled: true }, + canonicalChannels: [{ id: first.canonical.id, customName: 'Renamed One' }], + channels: [ + { + canonicalId: first.canonical.id, + position: 1, + enabled: false, + guideNumberOverride: null, + }, + { + canonicalId: second.canonical.id, + position: 0, + enabled: true, + guideNumberOverride: '900', + }, + ], + }); + + expect(response.data).to.include({ status: 'saved' }); + expect(response.data.profile).to.include({ name: 'Living Room', slug: 'default' }); + expect(response.data.channels.find(entry => entry.canonical.id === first.canonical.id)).to.include({ + position: 1, + enabled: false, + }); + + const profiles = await axios.get(`${baseUrl}/api/output-profiles`); + expect(profiles.data.profiles.find(profile => profile.slug === 'default').name).to.equal('Living Room'); + const channels = await axios.get(`${baseUrl}/api/output-profiles/default/channels`); + expect(channels.data.channels.find(channel => channel.canonicalId === second.canonical.id)).to.include({ + canonicalId: second.canonical.id, + name: 'Canonical Two', + guideNumber: '900', + position: 0, + }); + const canonical = await axios.get(`${baseUrl}/api/canonical/channels`); + expect(canonical.data.channels.find(channel => channel.id === first.canonical.id)).to.include({ + customName: 'Renamed One', + }); + }); + it('creates, updates, and deletes named output profiles', async () => { const createResponse = await axios.post(`${baseUrl}/api/output-profiles`, { name: 'Bedroom TV', diff --git a/test/unit/cache-manager.test.js b/test/unit/cache-manager.test.js index 357e974..2db4c50 100644 --- a/test/unit/cache-manager.test.js +++ b/test/unit/cache-manager.test.js @@ -23,6 +23,14 @@ describe('Cache Manager', () => { expect(bounded.get('second')).to.be.undefined; expect(bounded.get('third')).to.equal(3); }); + it('bounds cached payload bytes as well as entry count', () => { + const bounded = new Cache('byte-bounded', 0, 10, 5); + bounded.set('a', '1234'); + bounded.set('b', '5678'); + expect(bounded.has('a')).to.be.false; + expect(bounded.get('b')).to.equal('5678'); + expect(bounded.getStats().bytes).to.equal(4); + }); it('should set and get values', () => { cache.set('key1', 'value1'); const result = cache.get('key1'); diff --git a/test/unit/xmltv-stream.test.js b/test/unit/xmltv-stream.test.js new file mode 100644 index 0000000..e30a293 --- /dev/null +++ b/test/unit/xmltv-stream.test.js @@ -0,0 +1,48 @@ +import { describe, it } from 'mocha'; +import { expect } from 'chai'; +import { Readable } from 'node:stream'; +import { gzipSync } from 'node:zlib'; +import { decodeXmltvStream, readXmltvRecords } from '../../server/xmltv-stream.js'; + +function source(parts) { + return Readable.from(parts.map(part => Buffer.from(part))); +} + +describe('streaming XMLTV ingestion', () => { + it('parses records split across chunks without buffering a full document', async () => { + const records = await readXmltvRecords( + source([ + 'One', + 'Show', + ]) + ); + expect(records.channels).to.have.length(1); + expect(records.channels[0]['@_id']).to.equal('one'); + expect(records.programmes[0]['@_channel']).to.equal('one'); + expect(records.programmes[0].title).to.equal('Show'); + }); + + it('handles gzip-compressed XMLTV streams', async () => { + const xml = 'Zipped'; + const records = await readXmltvRecords(decodeXmltvStream(source([gzipSync(xml)]), true)); + expect(records.channels).to.have.length(1); + expect(records.programmes[0].title).to.equal('Zipped'); + }); + + it('processes a representative large feed delivered in small chunks', async () => { + const programme = 'Show'; + const records = await readXmltvRecords(source(['', ...Array(1000).fill(programme), ''])); + expect(records.programmes).to.have.length(1000); + expect(records.bytes).to.be.greaterThan(80 * 1024); + }); + + it('enforces a decoded input limit on large feeds', async () => { + const xml = `${'x'.repeat(500)}`; + try { + await readXmltvRecords(source([xml]), { maxBytes: 1024 }); + expect.fail('Expected input size rejection'); + } catch (error) { + expect(error.message).to.include('XMLTV input exceeds'); + } + }); +});