Files
Nodecast/server/services/liveProjectionService.js
blomios 9f97020996 feat: precompute stream grouping during source sync (#13)
Move variant grouping work out of Live TV rendering into the
catalog pipeline: a liveProjectionService builds the logical channel
projection (schema in liveProjectionSchema) whenever a source is
added, synced or modified (name change, new address), and the
projection is served through a /live route. ChannelList now consumes
the precomputed projection via a liveChannelProjection client
service, recalculating only when the catalog changes. Cover
projection building, incremental recompute and client consumption
with tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-10-02 15:35:02 +02:00

252 lines
9.9 KiB
JavaScript

const crypto = require('crypto');
const QUALITY_TOKENS = new Set([
'4k', 'uhd', 'fhd', 'fullhd', 'hd', 'sd', 'hevc', 'h265', 'h264',
'1080p', '1080i', '720p', '576p', '480p'
]);
const REGION_PREFIX = /^(?:(?:\|\s*[a-z]{2,3}\s*\|)|(?:[a-z]{2,3}(?:-[a-z]{2,3})?\s*\|))\s*/i;
function words(value) {
return String(value || '')
.normalize('NFKD')
.replace(/[\u0300-\u036f]/g, '')
.toLowerCase()
.replace(REGION_PREFIX, '')
.replace(/[|()[\]{}_.,:;+/\\-]+/g, ' ')
.split(/\s+/)
.filter(Boolean);
}
function normalizeChannelName(name) {
return words(name).filter(token => !QUALITY_TOKENS.has(token)).join(' ').trim();
}
function displayNameForFingerprint(fingerprint) {
return fingerprint.split(' ').map(token => {
if (token === 'bein') return 'beIN';
if (/^\d+$/.test(token)) return token;
return token.charAt(0).toUpperCase() + token.slice(1);
}).join(' ');
}
function qualityForName(name) {
const tokens = words(name);
if (tokens.includes('4k') || tokens.includes('uhd')) return { label: '4K', rank: 4 };
if (tokens.includes('fhd') || tokens.includes('fullhd') || tokens.includes('1080p') || tokens.includes('1080i')) {
return { label: 'Full HD', rank: 3 };
}
if (tokens.includes('hd') || tokens.includes('720p')) return { label: 'HD', rank: 2 };
if (tokens.includes('sd') || tokens.includes('576p') || tokens.includes('480p')) return { label: 'SD', rank: 0 };
return { label: 'Standard', rank: 1 };
}
function stableHash(value) {
return crypto.createHash('sha256').update(value).digest('hex');
}
function projectCatalog(items) {
const buckets = new Map();
for (const item of items) {
const fallback = `channel-${item.source_id}:${item.item_id}`;
const fingerprint = normalizeChannelName(item.name) || fallback;
const quality = qualityForName(item.name);
const projected = {
...item,
fingerprint,
groupingInputHash: stableHash(`${item.name || ''}\0${item.category_id || ''}`),
qualityLabel: quality.label,
qualityRank: quality.rank
};
if (!buckets.has(fingerprint)) buckets.set(fingerprint, []);
buckets.get(fingerprint).push(projected);
}
return Array.from(buckets, ([fingerprint, variants]) => {
variants.sort((a, b) => b.qualityRank - a.qualityRank || a.source_id - b.source_id || String(a.item_id).localeCompare(String(b.item_id)));
variants.forEach((variant, index) => { variant.variantRank = index; });
return {
id: `logical:${stableHash(fingerprint).slice(0, 24)}`,
fingerprint,
displayName: displayNameForFingerprint(fingerprint),
preferredVariantId: variants[0].id,
variants
};
}).sort((a, b) => a.displayName.localeCompare(b.displayName));
}
function catalogInputHash(items) {
return stableHash(items
.map(item => `${item.id}\0${item.name}\0${item.category_id || ''}`)
.sort()
.join('\n'));
}
function defaultDatabase() {
return require('../db/sqlite').getDb();
}
function projectionState(database = null) {
database ||= defaultDatabase();
return database.prepare('SELECT * FROM catalog_projection_state WHERE projection = ?').get('live') || {
projection: 'live', status: 'missing', started_at: null, completed_at: null,
input_hash: null, projected_count: 0, error: null
};
}
function rebuild(database = null) {
database ||= defaultDatabase();
const items = database.prepare(`
SELECT id, source_id, item_id, name, category_id
FROM playlist_items WHERE type = 'live'
`).all();
const groups = projectCatalog(items);
const inputHash = catalogInputHash(items);
const startedAt = Date.now();
database.prepare(`
INSERT INTO catalog_projection_state (projection, status, started_at, completed_at, input_hash, projected_count, error)
VALUES ('live', 'building', ?, NULL, NULL, 0, NULL)
ON CONFLICT(projection) DO UPDATE SET status = 'building', started_at = excluded.started_at,
completed_at = NULL, error = NULL
`).run(startedAt);
try {
const apply = database.transaction(() => {
database.prepare('DELETE FROM logical_channels').run();
database.prepare(`
UPDATE playlist_items SET logical_channel_id = NULL, grouping_fingerprint = NULL,
grouping_input_hash = NULL, quality_label = NULL, quality_rank = NULL, variant_rank = NULL
WHERE type = 'live'
`).run();
const insertLogical = database.prepare(`
INSERT INTO logical_channels
(id, fingerprint, display_name, preferred_variant_id, variant_count, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
`);
const updateVariant = database.prepare(`
UPDATE playlist_items SET logical_channel_id = ?, grouping_fingerprint = ?,
grouping_input_hash = ?, quality_label = ?, quality_rank = ?, variant_rank = ?
WHERE id = ? AND type = 'live'
`);
const now = Date.now();
for (const group of groups) {
insertLogical.run(group.id, group.fingerprint, group.displayName, group.preferredVariantId, group.variants.length, now, now);
for (const variant of group.variants) {
updateVariant.run(group.id, group.fingerprint, variant.groupingInputHash,
variant.qualityLabel, variant.qualityRank, variant.variantRank, variant.id);
}
}
database.prepare(`
UPDATE catalog_projection_state SET status = 'success', completed_at = ?,
input_hash = ?, projected_count = ?, error = NULL WHERE projection = 'live'
`).run(Date.now(), inputHash, groups.length);
});
apply();
return projectionState(database);
} catch (error) {
database.prepare(`
UPDATE catalog_projection_state SET status = 'error', completed_at = ?, error = ?
WHERE projection = 'live'
`).run(Date.now(), 'Live catalog projection failed');
throw error;
}
}
let rebuildQueue = Promise.resolve();
function queueRebuild() {
const next = rebuildQueue.catch(() => {}).then(() => rebuild());
rebuildQueue = next;
return next;
}
function parseData(value) {
try { return value ? JSON.parse(value) : {}; } catch { return {}; }
}
function getChannels({ sourceId = null, includeHidden = false, userId = null }, database = null) {
database ||= defaultDatabase();
const params = [];
let sourceFilter = '';
if (sourceId !== null) {
sourceFilter = 'AND p.source_id = ?';
params.push(sourceId);
}
const rows = database.prepare(`
SELECT p.*, c.name AS category_name, COALESCE(c.is_hidden, 0) AS category_hidden,
CASE WHEN f.id IS NULL THEN 0 ELSE 1 END AS user_favorite,
l.display_name, l.preferred_variant_id
FROM playlist_items p
JOIN logical_channels l ON l.id = p.logical_channel_id
LEFT JOIN categories c ON c.source_id = p.source_id AND c.type = 'live' AND c.category_id = p.category_id
LEFT JOIN favorites f ON f.user_id = ? AND f.source_id = p.source_id
AND f.item_id = p.item_id AND f.item_type = 'channel'
WHERE p.type = 'live' ${sourceFilter}
ORDER BY l.display_name, p.variant_rank
`).all(userId ?? -1, ...params);
return serializeChannels(rows, includeHidden);
}
function serializeChannels(rows, includeHidden = false) {
const groups = new Map();
for (const row of rows) {
const hidden = Boolean(row.is_hidden || row.category_hidden);
if (hidden && !includeHidden) continue;
const raw = parseData(row.data);
const categoryName = raw.groupTitle || row.category_name || 'Uncategorized';
const variant = {
...raw,
id: row.item_id,
sourceId: row.source_id,
name: row.name,
categoryId: row.category_id,
categoryName,
groupTitle: categoryName,
streamIcon: row.stream_icon,
streamUrl: row.stream_url,
qualityLabel: row.quality_label,
qualityRank: row.quality_rank,
variantRank: row.variant_rank,
isHidden: hidden,
isFavorite: Boolean(row.user_favorite)
};
if (!groups.has(row.logical_channel_id)) {
groups.set(row.logical_channel_id, {
id: row.logical_channel_id,
fingerprint: row.grouping_fingerprint,
displayName: row.display_name,
name: row.display_name,
variants: [],
isFavorite: false,
isHidden: true
});
}
const group = groups.get(row.logical_channel_id);
group.variants.push(variant);
group.isFavorite ||= variant.isFavorite;
group.isHidden &&= variant.isHidden;
if (row.id === row.preferred_variant_id) group.preferredVariant = variant;
}
return Array.from(groups.values()).map(group => {
group.preferredVariant ||= group.variants[0];
group.categoryId = group.preferredVariant?.categoryId || null;
group.categoryName = group.preferredVariant?.categoryName || 'Uncategorized';
group.groupTitle = group.preferredVariant?.groupTitle || group.categoryName;
group.selectedVariantId = group.preferredVariant
? `${group.preferredVariant.sourceId}:${group.preferredVariant.id}`
: null;
return group;
});
}
module.exports = {
catalogInputHash,
displayNameForFingerprint,
getChannels,
normalizeChannelName,
projectCatalog,
projectionState,
qualityForName,
queueRebuild,
rebuild,
serializeChannels
};