NodecastV2/server/services/syncService.js

const { getDb } = require('../db/sqlite');
const { sources, settings } = require('../db'); // For source config and settings
const xtreamApi = require('./xtreamApi');
const m3uParser = require('./m3uParser');
const epgParser = require('./epgParser');
const { randomUUID } = require('node:crypto');

// Sync tracking
const activeSyncs = new Set(); // sourceId
const activeEpgSyncs = new Set();
const epgSourceGenerations = new Map();

class SyncService {
    constructor() {
        this.lastSyncTime = null; // Track when global sync last completed
        this._syncTimer = null;   // Server-side sync timer
        this._currentInterval = null;
    }

    invalidateEpgSource(sourceId) {
        const key = String(sourceId);
        epgSourceGenerations.set(key, (epgSourceGenerations.get(key) || 0) + 1);
    }

    /**
     * Get when the last global sync completed
     */
    getLastSyncTime() {
        return this.lastSyncTime;
    }

    /**
     * Start the server-side sync timer based on settings
     * Should be called once on server startup after initial sync
     */
    async startSyncTimer() {
        // Get interval from settings
        const currentSettings = await settings.get();
        const intervalHours = parseInt(currentSettings.epgRefreshInterval) || 24;

        // If interval is 0, don't start timer (manual only mode)
        if (intervalHours <= 0) {
            console.log('[Sync] Auto-sync disabled (manual only mode)');
            this.stopSyncTimer();
            this._currentInterval = 0;
            return;
        }

        const intervalMs = intervalHours * 60 * 60 * 1000;

        // Don't restart if interval hasn't changed and timer exists
        if (this._currentInterval === intervalHours && this._syncTimer) {
            console.log(`[Sync] Timer already running for ${intervalHours} hours, not restarting`);
            return;
        }

        // Clear existing timer
        this.stopSyncTimer();

        const nextSyncTime = new Date(Date.now() + intervalMs);
        console.log(`[Sync] Starting server-side sync timer: every ${intervalHours} hours`);
        console.log(`[Sync] Next scheduled sync at: ${nextSyncTime.toLocaleString()}`);

        this._syncTimer = setInterval(async () => {
            console.log('[Sync] Scheduled sync triggered');
            await this.syncAll();
            // Log next sync time
            const next = new Date(Date.now() + intervalMs);
            console.log(`[Sync] Next scheduled sync at: ${next.toLocaleString()}`);
        }, intervalMs);

        this._currentInterval = intervalHours;
    }

    /**
     * Stop the server-side sync timer
     */
    stopSyncTimer() {
        if (this._syncTimer) {
            clearInterval(this._syncTimer);
            this._syncTimer = null;
        }
    }

    /**
     * Restart the sync timer with updated settings
     * Called when sync interval setting changes
     */
    async restartSyncTimer() {
        await this.startSyncTimer();
    }

    /**
     * Sync all enabled sources
     */
    async syncAll() {
        console.log('[Sync] Starting global sync...');
        try {
            const allSources = await sources.getAll();
            for (const source of allSources) {
                if (source.enabled) {
                    // Run sequentially to not overload
                    await this.syncSource(source.id);
                }
            }
            this.lastSyncTime = new Date();
            console.log('[Sync] Global sync completed at', this.lastSyncTime.toISOString());
        } catch (err) {
            console.error('[Sync] Global sync failed:', err);
        }
    }

    /**
     * Start sync for a source
     */
    async syncSource(sourceId) {
        if (activeSyncs.has(sourceId)) {
            console.log(`[Sync] Source ${sourceId} is already syncing`);
            return;
        }

        activeSyncs.add(sourceId);

        try {
            const db = getDb();
            const source = await sources.getById(sourceId);

            if (!source) {
                throw new Error(`Source ${sourceId} not found`);
            }

            console.log(`[Sync] Starting sync for source ${source.name} (ID: ${sourceId})`);

            if (!source.enabled) {
                console.log(`[Sync] Skipping disabled source ${source.name}`);
                activeSyncs.delete(sourceId);
                return;
            }

            // Update status
            this.updateSyncStatus(sourceId, 'all', 'syncing');

            if (source.type === 'xtream') {
                await this.syncXtream(source);
            } else if (source.type === 'm3u') {
                await this.syncM3u(source);
            } else if (source.type === 'epg') {
                await this.syncEpg(source);
            }

            this.updateSyncStatus(sourceId, 'all', 'success');
            console.log(`[Sync] Completed sync for source ${source.name}`);

        } catch (err) {
            console.error(`[Sync] Failed sync for source ${sourceId}:`, err);
            this.updateSyncStatus(sourceId, 'all', 'error', err.message);
        } finally {
            activeSyncs.delete(sourceId);
        }
    }

    /**
     * Update sync status in DB
     */
    updateSyncStatus(sourceId, type, status, error = null) {
        const db = getDb();
        const stmt = db.prepare(`
            INSERT INTO sync_status (source_id, type, last_sync, status, error)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(source_id, type) DO UPDATE SET
                last_sync = excluded.last_sync,
                status = excluded.status,
                error = excluded.error
        `);
        stmt.run(sourceId, type, Date.now(), status, error);
    }

    /**
     * Xtream Sync Logic
     */
    async syncXtream(source) {
        const api = xtreamApi.createFromSource(source);
        const db = getDb();

        // 1. Live Categories
        console.log(`[Sync] Fetching Live Categories for ${source.name}`);
        const liveCats = await api.getLiveCategories();
        await this.saveCategories(source.id, 'live', liveCats);

        // 2. Live Streams
        console.log(`[Sync] Fetching Live Streams for ${source.name}`);
        const liveStreams = await api.getLiveStreams();
        await this.saveStreams(source.id, 'live', liveStreams);

        // 3. VOD Categories
        console.log(`[Sync] Fetching VOD Categories for ${source.name}`);
        const vodCats = await api.getVodCategories();
        await this.saveCategories(source.id, 'movie', vodCats);

        // 4. VOD Streams
        console.log(`[Sync] Fetching VOD Streams for ${source.name}`);
        const vodStreams = await api.getVodStreams();
        await this.saveStreams(source.id, 'movie', vodStreams);

        // 5. Series Categories
        console.log(`[Sync] Fetching Series Categories for ${source.name}`);
        const seriesCats = await api.getSeriesCategories();
        await this.saveCategories(source.id, 'series', seriesCats);

        // 6. Series
        console.log(`[Sync] Fetching Series for ${source.name}`);
        const series = await api.getSeries();
        await this.saveStreams(source.id, 'series', series);

        // 7. EPG (Xmltv)
        // Try to fetch XMLTV if available
        console.log(`[Sync] Fetching EPG for ${source.name}`);
        try {
            const xmltvUrl = api.getXmltvUrl();
            await this.syncEpgFromUrl(source.id, xmltvUrl);
        } catch (e) {
            console.warn('[Sync] XMLTV fetch failed, skipping EPG sync for now:', e.message);
        }
    }

    /**
     * Batch save categories
     */
    async saveCategories(sourceId, type, categories) {
        if (!categories || categories.length === 0) return;
        console.log(`[Sync] Saving ${categories.length} ${type} categories for source ${sourceId}...`);
        const db = getDb();
        const stmt = db.prepare(`
            INSERT INTO categories (id, source_id, category_id, type, name, parent_id, data)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                name = excluded.name,
                data = excluded.data
        `);

        const insertBatch = db.transaction((batch) => {
            for (const cat of batch) {
                const catId = cat.category_id; // standard xtream field
                const name = cat.category_name;
                const id = `${sourceId}:${catId}`;
                stmt.run(id, sourceId, String(catId), type, name, cat.parent_id || null, JSON.stringify(cat));
            }
        });

        // Reduced batch size for better event loop interleaving
        const BATCH_SIZE = 100;
        for (let i = 0; i < categories.length; i += BATCH_SIZE) {
            insertBatch(categories.slice(i, i + BATCH_SIZE));
            // Yield to event loop between batches to allow other requests
            await new Promise(resolve => setImmediate(resolve));
        }

        console.log(`[Sync] Saved ${categories.length} ${type} categories`);
    }

    /**
     * Batch save streams (channels, vod, series)
     * Also purges stale entries that no longer exist in the source (unless skipPurge is true)
     * @param {number} sourceId - Source ID
     * @param {string} type - Type of items (live, movie, series)
     * @param {Array} items - Items to save
     * @param {Object} options - Options { skipPurge: boolean }
     * @returns {Set} Set of synced IDs (for external purge if skipPurge was true)
     */
    async saveStreams(sourceId, type, items, options = {}) {
        if (!items || items.length === 0) return new Set();
        const db = getDb();
        const { skipPurge = false } = options;

        // Collect all IDs we're syncing
        const syncedIds = new Set();

        const stmt = db.prepare(`
            INSERT INTO playlist_items (
                id, source_id, item_id, type, name, category_id, 
                stream_icon, stream_url, program, container_extension, 
                rating, year, added_at, data
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                name = excluded.name,
                category_id = excluded.category_id,
                stream_icon = excluded.stream_icon,
                stream_url = excluded.stream_url,
                program = excluded.program,
                container_extension = excluded.container_extension,
                data = excluded.data
        `);

        const insertBatch = db.transaction((batch) => {
            for (const item of batch) {
                // Map fields based on type
                let itemId, name, catId, icon, container;
                let rating = null, year = null, added = null;

                if (type === 'live') {
                    itemId = item.stream_id;
                    name = item.name || `Channel ${item.stream_id}`;
                    catId = item.category_id;
                    icon = item.stream_icon;
                    added = item.added;
                } else if (type === 'movie') {
                    itemId = item.stream_id;
                    name = item.name || `Movie ${item.stream_id}`;
                    catId = item.category_id;
                    icon = item.stream_icon; // or cover
                    container = item.container_extension;
                    rating = item.rating;
                    added = item.added;
                } else if (type === 'series') {
                    itemId = item.series_id;
                    name = item.name || `Series ${item.series_id}`;
                    catId = item.category_id;
                    icon = item.cover;
                    rating = item.rating;
                    year = item.releaseDate;
                    added = item.last_modified;
                }

                const id = `${sourceId}:${itemId}`;
                syncedIds.add(id);

                stmt.run(
                    id,
                    sourceId,
                    String(itemId),
                    type,
                    name,
                    String(catId),
                    icon,
                    null, // Direct URL not stored for Xtream usually, built on fly
                    item.program || null,
                    container,
                    rating,
                    year,
                    added,
                    JSON.stringify(item)
                );
            }
        });

        // Reduced batch size for better event loop interleaving
        const BATCH_SIZE = 100;
        for (let i = 0; i < items.length; i += BATCH_SIZE) {
            insertBatch(items.slice(i, i + BATCH_SIZE));
            // Yield to event loop between batches to allow other requests
            await new Promise(resolve => setImmediate(resolve));
        }

        // Purge stale entries (skip if doing batch sync like M3U)
        if (!skipPurge && syncedIds.size > 0) {
            await this.purgeStaleItems(sourceId, type, syncedIds);
        }

        console.log(`[Sync] Saved ${items.length} ${type} items`);
        return syncedIds;
    }

    /**
     * Purge stale items that are no longer in the source
     * @param {number} sourceId - Source ID
     * @param {string} type - Type of items (live, movie, series)
     * @param {Set} syncedIds - Set of IDs that should be kept
     */
    async purgeStaleItems(sourceId, type, syncedIds) {
        if (!syncedIds || syncedIds.size === 0) return;

        const db = getDb();
        db.exec('CREATE TEMP TABLE IF NOT EXISTS synced_ids (id TEXT PRIMARY KEY)');
        db.exec('DELETE FROM synced_ids');

        const insertTemp = db.prepare('INSERT OR IGNORE INTO synced_ids (id) VALUES (?)');
        const insertTempBatch = db.transaction((ids) => {
            for (const id of ids) {
                insertTemp.run(id);
            }
        });
        insertTempBatch([...syncedIds]);

        const deleteStmt = db.prepare(`
            DELETE FROM playlist_items 
            WHERE source_id = ? AND type = ? 
            AND id NOT IN (SELECT id FROM synced_ids)
        `);
        const deleted = deleteStmt.run(sourceId, type);

        if (deleted.changes > 0) {
            console.log(`[Sync] Purged ${deleted.changes} stale ${type} items`);
        }
    }


    /**
     * Sync EPG from URL (Streaming - Memory Efficient)
     * Processes EPG files in batches to avoid OOM on large EPG data
     */
    async syncEpgFromUrl(sourceId, url) {
        console.log(`[Sync] Fetching EPG from: ${new URL(url).origin}`);

        const sourceKey = String(sourceId);
        if (activeEpgSyncs.has(sourceKey)) {
            throw new Error(`EPG refresh already in progress for source ${sourceId}`);
        }
        activeEpgSyncs.add(sourceKey);
        const sourceGeneration = epgSourceGenerations.get(sourceKey) || 0;
        let cleanupRun = null;
        let refreshError = null;

        try {

            // Temporary memory logging for verification
            const logMemory = () => {
                const used = process.memoryUsage();
                console.log(`[Sync] Memory: ${Math.round(used.heapUsed / 1024 / 1024)}MB heap`);
            };

            logMemory();

            const db = getDb();
            const runId = randomUUID();
            let totalChannels = 0;
            let totalProgrammes = 0;
            let skippedProgrammes = 0;
            let batchCount = 0;
            let sawFinalBatch = false;

            db.exec(`
                CREATE TEMP TABLE IF NOT EXISTS epg_programs_staging (
                    run_id TEXT NOT NULL,
                    source_id INTEGER NOT NULL,
                    channel_id TEXT NOT NULL,
                    start_time INTEGER NOT NULL,
                    end_time INTEGER NOT NULL,
                    title TEXT,
                    description TEXT,
                    data JSON
                );
                CREATE INDEX IF NOT EXISTS temp.idx_epg_programs_staging_run
                    ON epg_programs_staging(run_id, source_id);

                CREATE TEMP TABLE IF NOT EXISTS epg_channels_staging (
                    run_id TEXT NOT NULL,
                    source_id INTEGER NOT NULL,
                    id TEXT NOT NULL,
                    item_id TEXT NOT NULL,
                    name TEXT,
                    stream_icon TEXT,
                    data JSON,
                    PRIMARY KEY (run_id, id)
                );
            `);

            const deleteStagedProgrammes = db.prepare(
                'DELETE FROM temp.epg_programs_staging WHERE run_id = ?'
            );
            const deleteStagedChannels = db.prepare(
                'DELETE FROM temp.epg_channels_staging WHERE run_id = ?'
            );
            cleanupRun = () => {
                deleteStagedProgrammes.run(runId);
                deleteStagedChannels.run(runId);
            };

            // A prior interrupted attempt in this process must never be reused.
            db.prepare('DELETE FROM temp.epg_programs_staging WHERE source_id = ?').run(sourceId);
            db.prepare('DELETE FROM temp.epg_channels_staging WHERE source_id = ?').run(sourceId);

            const programmeStmt = db.prepare(`
                INSERT INTO temp.epg_programs_staging (
                    run_id, source_id, channel_id, start_time, end_time, title, description, data
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            `);
            const channelStmt = db.prepare(`
                INSERT INTO temp.epg_channels_staging (
                    run_id, source_id, id, item_id, name, stream_icon, data
                ) VALUES (?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(run_id, id) DO UPDATE SET
                    item_id = excluded.item_id,
                    name = excluded.name,
                    stream_icon = excluded.stream_icon,
                    data = excluded.data
            `);

            const stageProgrammes = db.transaction((programmes) => {
                for (const programme of programmes) {
                    programmeStmt.run(
                        runId,
                        sourceId,
                        programme.channelId,
                        programme.start ? programme.start.getTime() : 0,
                        programme.stop ? programme.stop.getTime() : 0,
                        programme.title,
                        programme.description || programme.desc,
                        JSON.stringify(programme)
                    );
                }
            });
            const stageChannels = db.transaction((channels) => {
                for (const channel of channels) {
                    // EPG channel IDs often overlap Xtream stream IDs. Keep the
                    // staged row in the same dedicated namespace as the active row.
                    const id = `${sourceId}:epg_channel:${channel.id}`;
                    channelStmt.run(
                        runId,
                        sourceId,
                        id,
                        channel.id,
                        channel.name,
                        channel.icon || null,
                        JSON.stringify(channel)
                    );
                }
            });

            const activateStagedGuide = db.transaction(() => {
                db.prepare('DELETE FROM epg_programs WHERE source_id = ?').run(sourceId);
                db.prepare(`
                    INSERT INTO epg_programs (
                        channel_id, source_id, start_time, end_time, title, description, data
                    )
                    SELECT channel_id, source_id, start_time, end_time, title, description, data
                    FROM temp.epg_programs_staging
                    WHERE run_id = ? AND source_id = ?
                `).run(runId, sourceId);

                db.prepare(`
                    DELETE FROM playlist_items
                    WHERE source_id = ? AND type = 'epg_channel'
                `).run(sourceId);
                db.prepare(`
                    INSERT INTO playlist_items (
                        id, source_id, item_id, type, name, stream_icon,
                        stream_url, category_id, data
                    )
                    SELECT id, source_id, item_id, 'epg_channel', name, stream_icon,
                           NULL, NULL, data
                    FROM temp.epg_channels_staging
                    WHERE run_id = ? AND source_id = ?
                `).run(runId, sourceId);

                cleanupRun();
            });

            const sourceAtStart = await sources.getById(sourceId);

            // Stream into isolated staging tables. The active guide remains
            // readable throughout download and parsing.
            for await (const batch of epgParser.fetchAndParseStreaming(url)) {
                batchCount++;
                skippedProgrammes += batch.skippedProgrammes || 0;

                if (batch.channels !== null && batch.channels !== undefined) {
                    stageChannels(batch.channels);
                    totalChannels += batch.channels.length;
                }

                if (batch.programmes.length > 0) {
                    stageProgrammes(batch.programmes);
                    totalProgrammes += batch.programmes.length;
                }

                if (batch.isLast) sawFinalBatch = true;

                if (batchCount % 10 === 0) {
                    console.log(`[Sync] Staged ${totalProgrammes} programmes so far...`);
                    logMemory();
                }

                // Keep requests responsive between SQLite batches.
                await new Promise(resolve => setImmediate(resolve));
            }

            if (!sawFinalBatch) {
                throw new Error('EPG refresh ended before the XMLTV document completed');
            }

            // If a real configured source disappeared while its download was
            // running, do not recreate any of its guide data.
            const sourceStillExists = !sourceAtStart || await sources.getById(sourceId);
            const sourceWasInvalidated = (epgSourceGenerations.get(sourceKey) || 0) !== sourceGeneration;
            if (sourceWasInvalidated || !sourceStillExists) {
                throw new Error(`EPG source ${sourceId} was removed during refresh`);
            }

            console.log(`[Sync] EPG Parsed: ${totalChannels} channels, ${totalProgrammes} programmes`);
            if (skippedProgrammes > 0) {
                console.warn(`[Sync] Skipped ${skippedProgrammes} programme entries with invalid XMLTV timestamps`);
            }
            logMemory();

            // One short transaction makes programmes and channel mapping
            // visible together. A valid empty feed intentionally clears both.
            activateStagedGuide();
            console.log(`[Sync] Activated ${totalChannels} EPG channels and ${totalProgrammes} programmes`);
        } catch (error) {
            refreshError = error;
            throw error;
        } finally {
            try {
                if (cleanupRun) cleanupRun();
            } catch (cleanupError) {
                console.warn('[Sync] Failed to clean EPG staging data:', cleanupError.message);
                if (!refreshError) throw cleanupError;
            } finally {
                activeEpgSyncs.delete(sourceKey);
            }
        }
    }

    /**
     * M3U Sync Logic (Streaming - Memory Efficient)
     * Processes M3U files in batches to avoid OOM on large playlists
     */
    async syncM3u(source) {
        console.log(`[Sync] Fetching M3U playlist for ${source.name}`);

        // Temporary memory logging for verification
        const logMemory = () => {
            const used = process.memoryUsage();
            console.log(`[Sync] Memory: ${Math.round(used.heapUsed / 1024 / 1024)}MB heap`);
        };

        logMemory();

        const allGroups = new Set();
        const allSyncedIds = new Set(); // Collect IDs across all batches
        let totalChannels = 0;
        let batchCount = 0;

        // Stream and process in batches (default 500 channels per batch)
        for await (const batch of m3uParser.fetchAndParseStreaming(source.url)) {
            batchCount++;

            // Map M3U channel format to our schema
            const playlistItems = batch.channels.map(ch => ({
                stream_id: ch.id,
                name: ch.name,
                category_id: ch.groupTitle || 'Uncategorized',
                stream_icon: ch.tvgLogo,
                stream_url: ch.url,
                tvgId: ch.tvgId || null,
                program: ch.program || null,
            }));

            // Save this batch immediately (skip purge - we'll do it at the end)
            if (playlistItems.length > 0) {
                const batchIds = await this.saveStreams(source.id, 'live', playlistItems, { skipPurge: true });
                batchIds.forEach(id => allSyncedIds.add(id));
                totalChannels += playlistItems.length;
            }

            // Collect groups for category creation at the end
            batch.groups.forEach(g => allGroups.add(g));

            // Log progress every 10 batches
            if (batchCount % 10 === 0) {
                console.log(`[Sync] Processed ${totalChannels} channels so far...`);
                logMemory();
            }
        }

        console.log(`[Sync] M3U Parsed: ${totalChannels} channels, ${allGroups.size} groups`);
        logMemory();

        // Purge stale items after all batches are complete
        if (allSyncedIds.size > 0) {
            await this.purgeStaleItems(source.id, 'live', allSyncedIds);
        }

        // Save Categories (Groups) at the end
        const categories = Array.from(allGroups).map(name => ({
            category_id: name,
            category_name: name,
            parent_id: null
        }));

        await this.saveCategories(source.id, 'live', categories);
        console.log(`[Sync] M3U sync complete for ${source.name}`);
    }

    /**
     * EPG Source Sync Logic
     */
    async syncEpg(source) {
        console.log(`[Sync] Fetching standalone EPG for ${source.name}`);
        await this.syncEpgFromUrl(source.id, source.url);
    }
}

module.exports = new SyncService();