Skip to content

Fix streamed EPG ingestion and atomic profile saves #185

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cbulock merged 1 commit into from
Sep 6, 2026
Merged

Fix streamed EPG ingestion and atomic profile saves #185

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
194 changes: 29 additions & 165 deletions admin/src/App.vue
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'));
}
Comment on lines +2481 to 2484

await loadChannelAuthoringData();
Expand Down
46 changes: 36 additions & 10 deletions libs/cache-manager.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down Expand Up @@ -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;
Comment on lines +65 to +70
}

estimateBytes(value) {
if (typeof value === 'string') return Buffer.byteLength(value);
try {
return Buffer.byteLength(JSON.stringify(value));
} catch {
return 0;
}
}

/**
Expand All @@ -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);
}
Expand Down Expand Up @@ -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);
}
Expand All @@ -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);
}
}

/**
Expand All @@ -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;
}
Expand All @@ -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}%`,
Expand Down
Loading
Loading