const crypto = require('crypto'); const { pairKey, scoreCandidate, identity } = require('./liveGroupingScoring'); const { detectLanguage } = require('./channelLanguage'); const { matchesSearch } = require('./flexibleChannelSearch'); const iptvOrgCatalog = require('./iptvOrgCatalogService'); const epgMatching = require('./epgMatchingService'); const MAX_SYNCHRONOUS_ENRICHMENTS = 250; 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; const BEIN_SUFFIX_TOKENS = new Set(['fr', 'en', 'thai', 'tod', 'appel', 'app']); const BEIN_TIMESHIFT_TOKENS = new Set(['2h', '4h', '8h', '12h']); 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) { const tokens = words(name).filter(token => !QUALITY_TOKENS.has(token)); const brandIndex = tokens.findIndex((token, index) => token === 'bein' && tokens[index + 1] === 'sports'); if (brandIndex >= 0) { const branded = tokens.slice(brandIndex); const numberedBaseLength = branded[2] === 'max' && /^\d+$/.test(branded[3]) ? 4 : (/^\d+$/.test(branded[2]) ? 3 : 0); if (numberedBaseLength) { const suffix = branded.slice(numberedBaseLength); const removable = suffix.every(token => BEIN_SUFFIX_TOKENS.has(token) || BEIN_TIMESHIFT_TOKENS.has(token) ); if (removable) return branded.slice(0, numberedBaseLength).join(' '); } return branded.join(' ').trim(); } return tokens.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, options = {}) { const candidateBuckets = 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 }; const language = detectLanguage({ ...projected, data: parseData(projected.data) }); projected.languageCode = language.code; projected.languageConfidence = language.confidence; projected.languageScores = language.scores; projected.languageEvidence = language.evidence; const itemIdentity = identity(fingerprint); const candidateKey = itemIdentity.family && itemIdentity.number ? `${itemIdentity.family}:${itemIdentity.number}` : fingerprint; if (!candidateBuckets.has(candidateKey)) candidateBuckets.set(candidateKey, []); const groups = candidateBuckets.get(candidateKey); const matchingGroup = groups.find(group => group.some(member => scoreCandidate(member, projected, { mode: options.mode || 'algo', layaHint: options.identityHints?.get(pairKey(member, projected)) }).matched)); if (matchingGroup) matchingGroup.push(projected); else groups.push([projected]); } return Array.from(candidateBuckets.values()).flat().map(variants => { const fingerprint = variants.map(variant => variant.fingerprint).sort((a, b) => a.length - b.length || a.localeCompare(b))[0]; 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 safeIdentityDescriptor(item) { return { name: String(item.name || '').slice(0, 300), group: String(item.category_name || item.groupTitle || '').slice(0, 200), source: String(item.source_name || '').slice(0, 120), tvgId: String(item.tvg_id || '').slice(0, 160) }; } function identityCandidatePairs(items) { const buckets = new Map(); for (const item of items) { const fingerprint = normalizeChannelName(item.name); const itemIdentity = identity(fingerprint); if (!itemIdentity.family || !itemIdentity.number) continue; const key = `${itemIdentity.family}:${itemIdentity.number}`; if (!buckets.has(key)) buckets.set(key, []); buckets.get(key).push({ ...item, fingerprint }); } const pairs = []; for (const bucket of buckets.values()) { for (let left = 0; left < bucket.length; left++) { for (let right = left + 1; right < bucket.length; right++) { if (bucket[left].fingerprint === bucket[right].fingerprint) continue; if (scoreCandidate(bucket[left], bucket[right], { mode: 'algo' }).blocker) continue; pairs.push({ id: pairKey(bucket[left], bucket[right]), left: safeIdentityDescriptor(bucket[left]), right: safeIdentityDescriptor(bucket[right]) }); if (pairs.length >= 256) return pairs; } } } return pairs; } async function collectIdentityHints(items, mode, layaClient, options = {}) { if (mode === 'algo' || !layaClient) return new Map(); const pairs = identityCandidatePairs(items); const total = Math.ceil(pairs.length / 64); let processed = 0; const hints = new Map(); options.onProgress?.({ processed, total }); for (let index = 0; index < pairs.length; index += 64) { const results = await layaClient.identityHintBatch(pairs.slice(index, index + 64)); results.forEach(result => hints.set(result.id, result)); processed += 1; options.onProgress?.({ processed, total }); } return hints; } async function collectOptionalIdentityHints(items, mode, layaClient, options = {}) { if (mode === 'algo') return new Map(); try { return await collectIdentityHints(items, mode, layaClient, options); } catch (error) { console.warn('[LiveProjection] Laya identity hints unavailable; continuing deterministically'); return new Map(); } } 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, phase: 'idle', processed_count: 0, total_count: 0, laya_processed_count: 0, laya_total_count: 0, error: null }; } function updateProjectionProgress(database, phase, progress = {}) { database.prepare(` UPDATE catalog_projection_state SET phase = ?, laya_processed_count = COALESCE(?, laya_processed_count), laya_total_count = COALESCE(?, laya_total_count) WHERE projection = 'live' `).run(phase, progress.layaProcessed ?? null, progress.layaTotal ?? null); } function markProjectionBuilding(database = null) { database ||= defaultDatabase(); const total = database.prepare("SELECT COUNT(*) AS count FROM playlist_items WHERE type = 'live'").get().count; const startedAt = Date.now(); database.prepare(` INSERT INTO catalog_projection_state (projection, status, started_at, completed_at, input_hash, projected_count, phase, processed_count, total_count, laya_processed_count, laya_total_count, error) VALUES ('live', 'building', ?, NULL, NULL, 0, 'preparing', 0, ?, 0, 0, NULL) ON CONFLICT(projection) DO UPDATE SET status = 'building', started_at = excluded.started_at, completed_at = NULL, phase = 'preparing', processed_count = 0, total_count = excluded.total_count, laya_processed_count = 0, laya_total_count = 0, error = NULL `).run(startedAt, total); return projectionState(database); } function loadProjectionItems(database) { return database.prepare(` SELECT p.id, p.source_id, p.item_id, p.name, p.category_id, p.data, c.name AS category_name FROM playlist_items p LEFT JOIN categories c ON c.source_id = p.source_id AND c.type = 'live' AND c.category_id = p.category_id WHERE p.type = 'live' ORDER BY p.id `).all().map(item => { const data = parseData(item.data); return { ...item, tvg_id: data.tvgId || data.tvg_id || null }; }); } function rebuild(database = null, options = {}) { database ||= defaultDatabase(); const items = options.items || loadProjectionItems(database); if (!options.stateAlreadyBuilding) markProjectionBuilding(database); updateProjectionProgress(database, 'projecting'); const groups = projectCatalog(items, { mode: options.mode, identityHints: options.identityHints }); const inputHash = catalogInputHash(items); updateProjectionProgress(database, 'writing'); 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, language_code = NULL, language_confidence = NULL, language_scores = NULL, language_evidence = NULL, language_detector = NULL, language_input_hash = 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 = ?, language_code = ?, language_confidence = ?, language_scores = ?, language_evidence = ?, language_detector = 'algorithm', language_input_hash = ? 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.languageCode, variant.languageConfidence, JSON.stringify(variant.languageScores), JSON.stringify(variant.languageEvidence), stableHash(`${variant.name || ''}\0${variant.category_name || ''}\0${variant.tvg_id || ''}`), variant.id); } } database.prepare(` UPDATE catalog_projection_state SET status = 'success', phase = 'complete', completed_at = ?, input_hash = ?, projected_count = ?, processed_count = ?, total_count = ?, error = NULL WHERE projection = 'live' `).run(Date.now(), inputHash, groups.length, items.length, items.length); }); apply(); return projectionState(database); } catch (error) { const diagnosticCode = /^SQLITE_[A-Z_]+$/.test(error.code || '') ? error.code : 'INTERNAL'; console.error(`[LiveProjection] Rebuild transaction failed (${diagnosticCode}, ${error.name || 'Error'})`); database.prepare(` UPDATE catalog_projection_state SET status = 'error', phase = 'failed', completed_at = ?, error = ? WHERE projection = 'live' `).run(Date.now(), `Live catalog projection failed [${diagnosticCode}]`); throw error; } } async function rebuildWithSignals(database = null, options = {}) { database ||= defaultDatabase(); const items = loadProjectionItems(database); const configured = options.mode || (await require('../db').settings.get()).languageDetectionMode || 'algo'; const mode = ['algo', 'hybrid', 'laya'].includes(configured) ? configured : 'algo'; let identityHints = new Map(); if (mode !== 'algo') { const layaClient = options.layaClient || new (require('./layaService').LayaService)(); identityHints = await collectOptionalIdentityHints(items, mode, layaClient, { onProgress: ({ processed, total }) => updateProjectionProgress(database, 'laya_identity', { layaProcessed: processed, layaTotal: total }) }); } return rebuild(database, { ...options, items, mode, identityHints }); } let activeRebuild = null; let activeState = null; function startRebuild(database = null, options = {}) { if (activeRebuild) return { alreadyRunning: true, state: activeState, promise: activeRebuild }; database ||= defaultDatabase(); const mark = options.mark || markProjectionBuilding; const runner = options.runner || rebuildWithSignals; const schedule = options.schedule || setImmediate; activeState = mark(database); activeRebuild = new Promise((resolve, reject) => { schedule(() => { Promise.resolve() .then(() => runner(database, { stateAlreadyBuilding: true })) .then(resolve, reject); }); }).finally(() => { activeRebuild = null; activeState = null; }); return { alreadyRunning: false, state: activeState, promise: activeRebuild }; } function queueRebuild(database = null) { return startRebuild(database).promise; } function parseData(value) { try { return value ? JSON.parse(value) : {}; } catch { return {}; } } function getChannelPage({ sourceId = null, includeHidden = false, userId = null, q = '', languages = [], epgSources = [] }, database = null, dependencies = {}) { database ||= defaultDatabase(); const standardCatalog = dependencies.iptvOrgCatalog || iptvOrgCatalog; const epgMatcher = dependencies.epgMatching || epgMatching; const params = []; let sourceFilter = ''; if (sourceId !== null) { sourceFilter = 'AND p.source_id = ?'; params.push(sourceId); } const rows = database.prepare(` SELECT p.*, 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 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); const categoryRows = sourceId === null ? database.prepare("SELECT source_id, category_id, name, is_hidden FROM categories WHERE type = 'live'").all() : database.prepare("SELECT source_id, category_id, name, is_hidden FROM categories WHERE type = 'live' AND source_id = ?").all(sourceId); const categories = new Map(categoryRows.map(category => [`${category.source_id}\0${category.category_id}`, category])); for (const row of rows) { const category = categories.get(`${row.source_id}\0${row.category_id}`); row.category_name = category?.name || null; row.category_hidden = category?.is_hidden || 0; } const serialized = serializeChannels(rows, includeHidden); const availableLanguages = [...new Map(serialized.map(channel => [channel.language.code, channel.language])).values()] .map(language => ({ code: language.code, short: language.short, label: language.label })) .sort((left, right) => left.label.localeCompare(right.label)); const filtered = serialized.filter(channel => (!languages.length || languages.includes(channel.language.code)) && matchesSearch(channel, q) ); // Catalog loading must stay bounded: both enrichers perform synchronous SQLite work, // and EPG matching is especially expensive against large XMLTV catalogs. Enrich only // a genuinely targeted, bounded search; base projection data remains usable otherwise. const shouldEnrich = Boolean(String(q).trim()) && filtered.length <= MAX_SYNCHRONOUS_ENRICHMENTS; const channels = !shouldEnrich ? filtered : filtered.map(channel => { const match = standardCatalog.findMatch(database, channel); if (match) { channel.standardName = match.name; channel.standard = { provider: 'iptv-org', channelId: match.channelId, country: match.country, confidence: match.confidence }; } const epgMatch = epgMatcher.findBestMatch(database, channel, epgSources); if (epgMatch) channel.epg = { ...channel.epg, ...epgMatch, language: channel.language.code }; return channel; }); return { channels, availableLanguages }; } function getChannels(options, database = null) { return getChannelPage(options, database).channels; } 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) }; const languageCode = row.language_code || 'und'; const languageKey = `${row.logical_channel_id}:${languageCode}`; if (!groups.has(languageKey)) { groups.set(languageKey, { id: languageKey, logicalChannelId: row.logical_channel_id, fingerprint: row.grouping_fingerprint, displayName: row.display_name, standardName: row.display_name, standard: { provider: 'local', channelId: null, confidence: 0 }, language: { code: languageCode, short: languageCode === 'und' ? 'UND' : languageCode.toUpperCase(), label: languageCode === 'und' ? 'Unknown' : new Intl.DisplayNames(['en'], { type: 'language' }).of(languageCode), confidence: Number(row.language_confidence || 0) }, name: row.display_name, variants: [], isFavorite: false, isHidden: true }); } const group = groups.get(languageKey); group.variants.push(variant); group.isFavorite ||= variant.isFavorite; group.isHidden &&= variant.isHidden; if (!group.preferredVariant || Number(variant.variantRank) < Number(group.preferredVariant.variantRank)) 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; const epgChannelId = group.preferredVariant?.tvgId || group.preferredVariant?.epg_channel_id || null; group.epg = { sourceId: null, channelId: epgChannelId, language: group.language.code, matchConfidence: epgChannelId ? 100 : 0 }; return group; }); } module.exports = { catalogInputHash, collectIdentityHints, collectOptionalIdentityHints, displayNameForFingerprint, getChannelPage, getChannels, identityCandidatePairs, normalizeChannelName, markProjectionBuilding, MAX_SYNCHRONOUS_ENRICHMENTS, matchesSearch, projectCatalog, projectionState, qualityForName, queueRebuild, rebuild, rebuildWithSignals, safeIdentityDescriptor, serializeChannels, startRebuild };