Nodecast/server/services/liveProjectionService.js

const crypto = require('crypto');
const { pairKey, scoreCandidate, identity } = require('./liveGroupingScoring');
const { detectLanguage } = require('./channelLanguage');
const { matchesSearch } = require('./flexibleChannelSearch');
const iptvOrgCatalog = require('./iptvOrgCatalogService');

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 getChannels({ sourceId = null, includeHidden = false, userId = null, q = '', languages = [] }, 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).map(channel => {
        const match = iptvOrgCatalog.findMatch(database, channel);
        if (match) {
            channel.standardName = match.name;
            channel.standard = { provider: 'iptv-org', channelId: match.channelId, country: match.country, confidence: match.confidence };
        }
        return channel;
    }).filter(channel =>
        (!languages.length || languages.includes(channel.language.code)) && matchesSearch(channel, q)
    );
}

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,
    getChannels,
    identityCandidatePairs,
    normalizeChannelName,
    markProjectionBuilding,
    matchesSearch,
    projectCatalog,
    projectionState,
    qualityForName,
    queueRebuild,
    rebuild,
    rebuildWithSignals,
    safeIdentityDescriptor,
    serializeChannels,
    startRebuild
};