Merge branch 'fix/t35-laya-stream-performance' into develop (#35)

server/routes/live.js

@@ -43,7 +43,7 @@ router.get('/channels', requireAuth, async (req, res) => {
     try {
         let state = liveProjection.projectionState();
         if (state.status !== 'success' && state.status !== 'building') {
-            state = await liveProjection.startRebuild().promise;
+            state = await liveProjection.startRebuild(null, { mode: 'algo' }).promise;
         }
         const epgSources = await sources.getByType('epg');
         const { channels, availableLanguages } = liveProjection.getChannelPage({

server/services/liveProjectionService.js

@@ -323,14 +323,17 @@ 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 {
-    const runner = options.runner || rebuildWithSignals;
+        mark = markProjectionBuilding,
-    const schedule = options.schedule || setImmediate;
+        runner = rebuildWithSignals,
+        schedule = setImmediate,
+        ...rebuildOptions
+    } = options;
     activeState = mark(database);
     activeRebuild = new Promise((resolve, reject) => {
         schedule(() => {
             Promise.resolve()
-                .then(() => runner(database, { stateAlreadyBuilding: true }))
+                .then(() => runner(database, { ...rebuildOptions, stateAlreadyBuilding: true }))
                 .then(resolve, reject);
         });
     }).finally(() => {
@@ -341,7 +344,7 @@ function startRebuild(database = null, options = {}) {
 }
 
 function queueRebuild(database = null) {
-    return startRebuild(database).promise;
+    return startRebuild(database, { mode: 'algo' }).promise;
 }
 
 function parseData(value) {

test/liveProjectionService.test.js

@@ -121,6 +121,28 @@ test('manual rebuild scheduling coalesces concurrent requests', async () => {
     assert.equal(runs, 1);
 });
 
+test('startRebuild passes business options to the runner without infrastructure options', async () => {
+    let runnerOptions;
+    const result = startRebuild({}, {
+        mode: 'algo',
+        layaClient: { identityHintBatch: async () => [] },
+        mark: () => ({ status: 'building' }),
+        runner: async (_database, options) => {
+            runnerOptions = options;
+            return { status: 'success' };
+        },
+        schedule: callback => callback()
+    });
+
+    await result.promise;
+    assert.equal(runnerOptions.mode, 'algo');
+    assert.equal(typeof runnerOptions.layaClient.identityHintBatch, 'function');
+    assert.equal(runnerOptions.stateAlreadyBuilding, true);
+    assert.equal('mark' in runnerOptions, false);
+    assert.equal('runner' in runnerOptions, false);
+    assert.equal('schedule' in runnerOptions, false);
+});
+
 test('Laya rebuild progress advances after a successful identity batch while analysis is still running', async () => {
     const db = createProjectionDb();
     for (let index = 0; index < 12; index++) {

test/t35LayaPerfRegression.test.js

@@ -0,0 +1,157 @@
+const test = require('node:test');
+const assert = require('node:assert/strict');
+const fs = require('node:fs');
+const path = require('node:path');
+const Database = require('better-sqlite3');
+const {
+    ensureColumns,
+    LIVE_PROJECTION_COLUMNS,
+    LIVE_PROJECTION_SCHEMA_SQL
+} = require('../server/db/liveProjectionSchema');
+const appDb = require('../server/db');
+const layaModule = require('../server/services/layaService');
+const liveProjection = require('../server/services/liveProjectionService');
+
+const liveRouteSource = fs.readFileSync(path.join(__dirname, '..', 'server', 'routes', 'live.js'), 'utf8');
+
+function createDb() {
+    const db = new Database(':memory:');
+    db.exec(`
+        CREATE TABLE categories (
+            id TEXT PRIMARY KEY,
+            source_id INTEGER NOT NULL,
+            category_id TEXT NOT NULL,
+            type TEXT NOT NULL,
+            name TEXT NOT NULL,
+            parent_id TEXT,
+            is_hidden INTEGER DEFAULT 0,
+            data JSON
+        );
+        CREATE TABLE playlist_items (
+            id TEXT PRIMARY KEY,
+            source_id INTEGER NOT NULL,
+            item_id TEXT NOT NULL,
+            type TEXT NOT NULL,
+            name TEXT NOT NULL,
+            category_id TEXT,
+            parent_id TEXT,
+            stream_icon TEXT,
+            stream_url TEXT,
+            container_extension TEXT,
+            rating REAL,
+            year TEXT,
+            added_at TEXT,
+            is_hidden INTEGER DEFAULT 0,
+            is_favorite INTEGER DEFAULT 0,
+            data JSON
+        );
+        CREATE TABLE favorites (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            user_id INTEGER NOT NULL,
+            source_id INTEGER NOT NULL,
+            item_id TEXT NOT NULL,
+            item_type TEXT NOT NULL
+        );
+    `);
+    ensureColumns(db, 'playlist_items', LIVE_PROJECTION_COLUMNS);
+    db.exec(LIVE_PROJECTION_SCHEMA_SQL);
+    return db;
+}
+
+function seedChannel(db, sourceId, itemId, name) {
+    db.prepare(`
+        INSERT INTO playlist_items
+            (id, source_id, item_id, type, name, category_id, stream_icon, stream_url, data)
+        VALUES (?, ?, ?, 'live', ?, 'sports', '', ?, '{}')
+    `).run(`${sourceId}:${itemId}`, sourceId, itemId, name, `https://streams.test/${itemId}.m3u8`);
+}
+
+async function simulateLiveListOpen(database) {
+    let state = liveProjection.projectionState(database);
+    if (state.status !== 'success' && state.status !== 'building') {
+        state = await liveProjection.startRebuild(database, {
+            mode: 'algo',
+            schedule: callback => callback()
+        }).promise;
+    }
+    return liveProjection.getChannelPage({ includeHidden: false, userId: 7 }, database, {
+        iptvOrgCatalog: { findMatch: () => null },
+        epgMatching: { findBestMatch: () => null }
+    });
+}
+
+test('opening the live list with a missing projection must not start a global Laya identity batch', async () => {
+    assert.match(
+        liveRouteSource,
+        /liveProjection\.startRebuild\(null, \{ mode: 'algo' \}\)/,
+        'the live-list repair path must explicitly request the algorithmic rebuild mode'
+    );
+    const db = createDb();
+    for (let index = 0; index < 12; index++) {
+        seedChannel(db, 1, `existing-${index}`, `beIN Sports 1 regional ${index}`);
+    }
+
+    let layaCalls = 0;
+    let layaPairCount = 0;
+    const originalSettingsGet = appDb.settings.get;
+    const OriginalLayaService = layaModule.LayaService;
+    appDb.settings.get = async () => ({ ...appDb.getDefaultSettings(), languageDetectionMode: 'laya' });
+    layaModule.LayaService = class {
+        async identityHintBatch(pairs) {
+            layaCalls += 1;
+            layaPairCount += pairs.length;
+            return pairs.map(pair => ({
+                id: pair.id,
+                sameIdentityProbability: 0.9,
+                confidence: 0.9,
+                family: 'bein sports',
+                number: '1'
+            }));
+        }
+    };
+
+    try {
+        const page = await simulateLiveListOpen(db);
+        assert.ok(page.channels.length > 0);
+        assert.equal(layaCalls, 0, `unexpected Laya identity batches while opening live list; batches=${layaCalls}, pairs=${layaPairCount}`);
+    } finally {
+        appDb.settings.get = originalSettingsGet;
+        layaModule.LayaService = OriginalLayaService;
+        db.close();
+    }
+});
+
+test('background live sync rebuilds must not start Laya identity batches in configured Laya mode', async () => {
+    const db = createDb();
+    for (let index = 0; index < 12; index++) {
+        seedChannel(db, 1, `existing-${index}`, `beIN Sports 1 regional ${index}`);
+    }
+
+    let layaCalls = 0;
+    let layaPairCount = 0;
+    const originalSettingsGet = appDb.settings.get;
+    const OriginalLayaService = layaModule.LayaService;
+    appDb.settings.get = async () => ({ ...appDb.getDefaultSettings(), languageDetectionMode: 'laya' });
+    layaModule.LayaService = class {
+        async identityHintBatch(pairs) {
+            layaCalls += 1;
+            layaPairCount += pairs.length;
+            return pairs.map(pair => ({
+                id: pair.id,
+                sameIdentityProbability: 0.9,
+                confidence: 0.9,
+                family: 'bein sports',
+                number: '1'
+            }));
+        }
+    };
+
+    try {
+        await liveProjection.queueRebuild(db);
+        assert.equal(layaCalls, 0, `unexpected Laya identity batches during background sync rebuild; batches=${layaCalls}, pairs=${layaPairCount}`);
+    } finally {
+        appDb.settings.get = originalSettingsGet;
+        layaModule.LayaService = OriginalLayaService;
+        db.close();
+    }
+});