blomios/Nodecast
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: precompute stream grouping during source sync (#13)

Browse Source 
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>
This commit is contained in:
blomios
2026-10-02 15:35:02 +02:00
parent 7b6525708c
commit 9f97020996
16 changed files with 619 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
server/db/liveProjectionSchema.js Normal file
View File

@ -0,0 +1,49 @@
const LIVE_PROJECTION_COLUMNS = {
logical_channel_id: 'TEXT',
grouping_fingerprint: 'TEXT',
grouping_input_hash: 'TEXT',
quality_label: 'TEXT',
quality_rank: 'INTEGER',
variant_rank: 'INTEGER',
language_code: 'TEXT',
language_confidence: 'REAL',
language_scores: 'JSON',
language_evidence: 'JSON',
language_detector: 'TEXT',
language_input_hash: 'TEXT'
};
const LIVE_PROJECTION_SCHEMA_SQL = `
CREATE INDEX IF NOT EXISTS idx_items_logical_channel ON playlist_items(logical_channel_id);
CREATE INDEX IF NOT EXISTS idx_items_grouping_fingerprint ON playlist_items(grouping_fingerprint);
CREATE TABLE IF NOT EXISTS logical_channels (
id TEXT PRIMARY KEY,
fingerprint TEXT NOT NULL UNIQUE,
display_name TEXT NOT NULL,
preferred_variant_id TEXT,
variant_count INTEGER NOT NULL DEFAULT 0,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_logical_channels_fingerprint ON logical_channels(fingerprint);
CREATE TABLE IF NOT EXISTS catalog_projection_state (
projection TEXT PRIMARY KEY,
status TEXT NOT NULL,
started_at INTEGER,
completed_at INTEGER,
input_hash TEXT,
projected_count INTEGER NOT NULL DEFAULT 0,
error TEXT
);
`;
function ensureColumns(database, table, columns) {
const existing = new Set(database.prepare(`PRAGMA table_info(${table})`).all().map(column => column.name));
for (const [name, definition] of Object.entries(columns)) {
if (!existing.has(name)) database.exec(`ALTER TABLE ${table} ADD COLUMN ${name} ${definition}`);
}
}
module.exports = { ensureColumns, LIVE_PROJECTION_COLUMNS, LIVE_PROJECTION_SCHEMA_SQL };

10
server/db/sqlite.js
View File

@ -1,6 +1,11 @@
const Database = require('better-sqlite3');
const path = require('path');
const fs = require('fs');
const {
ensureColumns,
LIVE_PROJECTION_COLUMNS,
LIVE_PROJECTION_SCHEMA_SQL
} = require('./liveProjectionSchema');
const dataDir = path.join(__dirname, '..', '..', 'data');
const dbPath = path.join(dataDir, 'content.db');
@ -73,6 +78,9 @@ function initSchema() {
CREATE INDEX IF NOT EXISTS idx_items_category ON playlist_items(source_id, category_id);
`);
ensureColumns(db, 'playlist_items', LIVE_PROJECTION_COLUMNS);
db.exec(LIVE_PROJECTION_SCHEMA_SQL);
// EPG Programs
// Optimized for range queries
db.exec(`
@ -247,6 +255,8 @@ const layaLanguageCache = {
module.exports = {
getDb,
initSchema,
ensureColumns,
LIVE_PROJECTION_COLUMNS,
favorites,
layaLanguageCache
};

1
server/index.js
View File

@ -181,6 +181,7 @@ app.use('/api/subtitle', require('./routes/subtitle'));
app.use('/api/settings', require('./routes/settings'));
app.use('/api/history', require('./routes/history'));
app.use('/api/laya', require('./routes/laya'));
app.use('/api/live', require('./routes/live'));
// Version endpoint
app.get('/api/version', (req, res) => {

46
server/routes/live.js Normal file
View File

@ -0,0 +1,46 @@
const express = require('express');
const { requireAuth } = require('../auth');
const liveProjection = require('../services/liveProjectionService');
const router = express.Router();
function serializeState(state) {
return {
status: state.status,
stale: state.status !== 'success',
startedAt: state.started_at,
completedAt: state.completed_at,
inputHash: state.input_hash,
projectedCount: state.projected_count,
error: state.error
};
}
router.get('/channels', requireAuth, async (req, res) => {
const sourceId = req.query.sourceId === undefined ? null : Number(req.query.sourceId);
if (sourceId !== null && (!Number.isInteger(sourceId) || sourceId <= 0)) {
return res.status(400).json({ error: 'sourceId must be a positive integer' });
}
if (req.query.includeHidden !== undefined && !['true', 'false'].includes(req.query.includeHidden)) {
return res.status(400).json({ error: 'includeHidden must be true or false' });
}
const includeHidden = req.query.includeHidden === 'true';
try {
let state = liveProjection.projectionState();
if (state.status !== 'success') state = liveProjection.rebuild();
const channels = liveProjection.getChannels({ sourceId, includeHidden, userId: req.user.id });
return res.json({ projection: serializeState(state), channels });
} catch (error) {
console.error('[LiveProjection] Failed to serve projected channels:', error.message);
const state = liveProjection.projectionState();
return res.status(503).json({
projection: serializeState(state),
channels: [],
error: 'Live channel projection is unavailable'
});
}
});
module.exports = router;
module.exports.serializeState = serializeState;

3
server/routes/sources.js
View File

@ -5,6 +5,7 @@ const { getDb } = require('../db/sqlite');
const xtreamApi = require('../services/xtreamApi');
const syncService = require('../services/syncService');
const m3uParser = require('../services/m3uParser');
const liveProjection = require('../services/liveProjectionService');
// Get all sources
router.get('/', async (req, res) => {
@ -132,6 +133,7 @@ router.delete('/:id', async (req, res) => {
// Delete source config and related hidden items (favorites handled by db.js)
await sources.delete(sourceId);
await liveProjection.queueRebuild();
res.json({ success: true });
} catch (err) {
@ -278,4 +280,3 @@ router.post('/sync-all', async (req, res) => {
});
module.exports = router;

251
server/services/liveProjectionService.js Normal file
View File

@ -0,0 +1,251 @@
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
};

3
server/services/syncService.js
View File

@ -3,6 +3,7 @@ const { sources, settings } = require('../db'); // For source config and setting
const xtreamApi = require('./xtreamApi');
const m3uParser = require('./m3uParser');
const epgParser = require('./epgParser');
const liveProjection = require('./liveProjectionService');
// Sync tracking
const activeSyncs = new Set(); // sourceId
@ -183,6 +184,7 @@ class SyncService {
console.log(`[Sync] Fetching Live Streams for ${source.name}`);
const liveStreams = await api.getLiveStreams();
await this.saveStreams(source.id, 'live', liveStreams);
await liveProjection.queueRebuild();
// 3. VOD Categories
console.log(`[Sync] Fetching VOD Categories for ${source.name}`);
@ -559,6 +561,7 @@ class SyncService {
}));
await this.saveCategories(source.id, 'live', categories);
await liveProjection.queueRebuild();
console.log(`[Sync] M3U sync complete for ${source.name}`);
}