Nodecast/server/services/layaService.js

const crypto = require('crypto');
const fs = require('fs/promises');
const path = require('path');

const MODEL_DIR = path.resolve(__dirname, '..', '..', 'data', 'models', 'laya');
const MAX_BATCH_ITEMS = 64;
const MAX_BATCH_BYTES = 64 * 1024;
const MAX_TEXT_LENGTH = 4096;
const ALLOWED_MODELS = new Set(['english', 'multilingual', 'typed-decisions']);
const DOWNLOAD_TIMEOUT_MS = 30 * 60 * 1000;
const BATCH_TIMEOUT_MS = 2 * 60 * 1000;

class LayaRequestError extends Error {
    constructor(message, statusCode = 503) {
        super(message);
        this.name = 'LayaRequestError';
        this.statusCode = statusCode;
    }
}

function validateBatchPayload(payload) {
    if (!payload || typeof payload !== 'object' || Array.isArray(payload)) {
        throw new LayaRequestError('Request body must be an object', 400);
    }
    if (!Array.isArray(payload.items) || payload.items.length === 0) {
        throw new LayaRequestError('items must be a non-empty array', 400);
    }
    if (Object.keys(payload).some(key => !['items', 'model'].includes(key))) {
        throw new LayaRequestError('Request body contains unsupported fields', 400);
    }
    if (payload.items.length > MAX_BATCH_ITEMS || Buffer.byteLength(JSON.stringify(payload)) > MAX_BATCH_BYTES) {
        throw new LayaRequestError('Batch payload is too large', 413);
    }
    payload.items.forEach((item, index) => {
        if (!item || typeof item !== 'object' || typeof item.text !== 'string' || !item.text.trim()) {
            throw new LayaRequestError(`items[${index}].text must be a non-empty string`, 400);
        }
        if (item.text.length > MAX_TEXT_LENGTH) {
            throw new LayaRequestError(`items[${index}].text is too long`, 413);
        }
        if (Object.keys(item).some(key => !['id', 'text'].includes(key))) {
            throw new LayaRequestError(`items[${index}] contains unsupported fields`, 400);
        }
        if (item.id !== undefined && !['string', 'number'].includes(typeof item.id)) {
            throw new LayaRequestError(`items[${index}].id must be a string or number`, 400);
        }
    });
    if (payload.model !== undefined && !ALLOWED_MODELS.has(payload.model)) {
        throw new LayaRequestError('Unsupported Laya model', 400);
    }
    return payload;
}

async function clearDirectoryContents(directory) {
    const root = path.resolve(directory);
    await fs.mkdir(root, { recursive: true });
    const entries = await fs.readdir(root);
    for (const entry of entries) {
        const target = path.resolve(root, entry);
        if (path.dirname(target) !== root) {
            throw new Error('Refusing to delete outside the Laya model directory');
        }
        await fs.rm(target, { recursive: true, force: true });
    }
}

async function directorySize(directory) {
    let total = 0;
    let entries;
    try {
        entries = await fs.readdir(directory, { withFileTypes: true });
    } catch (error) {
        if (error.code === 'ENOENT') return 0;
        throw error;
    }
    for (const entry of entries) {
        const target = path.join(directory, entry.name);
        if (entry.isDirectory()) total += await directorySize(target);
        else if (entry.isFile()) total += (await fs.stat(target)).size;
    }
    return total;
}

function safeError(error) {
    const statusCode = error instanceof LayaRequestError ? error.statusCode : 500;
    const message = statusCode === 500 ? 'Model download failed' : error.message;
    return {
        message,
        detail: statusCode === 503
            ? 'The optional Laya runtime is unavailable.'
            : `Laya model operation failed (HTTP ${statusCode}).`
    };
}

function initialDownloadState() {
    return {
        phase: 'idle',
        downloading: false,
        progressPercent: null,
        bytesDownloaded: null,
        totalBytes: null,
        error: null,
        lastError: null
    };
}

function sanitizeIdentityText(value, maxLength) {
    return String(value || '')
        .replace(/https?:\/\/\S+/gi, '[redacted]')
        .replace(/\b(?:password|passwd|token|username|user)\s*[=:]\s*\S+/gi, '[redacted]')
        .slice(0, maxLength);
}

function sanitizeIdentityDescriptor(value) {
    return {
        name: sanitizeIdentityText(value?.name, 300),
        group: sanitizeIdentityText(value?.group, 200),
        source: sanitizeIdentityText(value?.source, 120),
        tvgId: sanitizeIdentityText(value?.tvgId, 160)
    };
}

class LayaService {
    constructor(options = {}) {
        this.baseUrl = options.baseUrl || process.env.LAYA_BASE_URL || 'http://laya:8000';
        this.modelDir = path.resolve(options.modelDir || MODEL_DIR);
        this.fetch = options.fetch || globalThis.fetch;
        this.cache = options.cache === undefined
            ? require('../db/sqlite').layaLanguageCache
            : options.cache;
        this.timeoutMs = options.timeoutMs || 3000;
        this.batchTimeoutMs = options.batchTimeoutMs
            || Number(process.env.LAYA_BATCH_TIMEOUT_MS)
            || BATCH_TIMEOUT_MS;
        this.downloadTimeoutMs = options.downloadTimeoutMs || DOWNLOAD_TIMEOUT_MS;
        this.progressIntervalMs = options.progressIntervalMs || 500;
        this.downloadState = initialDownloadState();
        this.downloadPromise = null;
    }

    async request(endpoint, options = {}) {
        if (typeof this.fetch !== 'function') throw new LayaRequestError('Laya runtime client is unavailable');
        const controller = new AbortController();
        const { timeoutMs = this.timeoutMs, ...fetchOptions } = options;
        const timeout = setTimeout(() => controller.abort(), timeoutMs);
        const headers = { 'content-type': 'application/json' };
        if (process.env.LAYA_API_KEY) headers.authorization = `Bearer ${process.env.LAYA_API_KEY}`;
        try {
            const response = await this.fetch(`${this.baseUrl}${endpoint}`, {
                ...fetchOptions,
                headers: { ...headers, ...fetchOptions.headers },
                signal: controller.signal
            });
            const data = await response.json().catch(() => null);
            if (!response.ok) throw new LayaRequestError(`Laya runtime returned HTTP ${response.status}`, 502);
            return data;
        } catch (error) {
            if (error instanceof LayaRequestError) throw error;
            if (error?.name === 'AbortError') {
                throw new LayaRequestError(`Laya runtime request timed out after ${timeoutMs} ms`, 504);
            }
            throw new LayaRequestError('Laya runtime is unavailable');
        } finally {
            clearTimeout(timeout);
        }
    }

    async status() {
        await fs.mkdir(this.modelDir, { recursive: true });
        const sizeBytes = await directorySize(this.modelDir);
        const modelPresent = sizeBytes > 0;
        if (this.downloadState.downloading) this.downloadState.bytesDownloaded = sizeBytes;
        try {
            const runtime = await this.request('/health', { method: 'GET' });
            if (!this.downloadState.downloading && this.downloadState.phase !== 'failed' && modelPresent) {
                this.downloadState.phase = 'ready';
            }
            return {
                ...this.downloadState,
                runtimeAvailable: true,
                modelPresent,
                ready: modelPresent,
                sizeBytes,
                runtime,
                runtimeError: null
            };
        } catch (error) {
            return {
                ...this.downloadState,
                runtimeAvailable: false,
                modelPresent,
                ready: false,
                sizeBytes,
                runtimeError: {
                    message: error.message,
                    detail: 'Start the Laya sidecar and verify LAYA_BASE_URL.'
                }
            };
        }
    }

    startModelDownload(model = 'multilingual') {
        if (!ALLOWED_MODELS.has(model)) throw new LayaRequestError('Unsupported Laya model', 400);
        if (this.downloadState.downloading) throw new LayaRequestError('A Laya model download is already running', 409);
        this.downloadState = {
            ...initialDownloadState(),
            phase: 'preparing',
            downloading: true,
            bytesDownloaded: 0
        };
        this.downloadPromise = this.downloadModel(model).catch(() => undefined);
        return { ...this.downloadState };
    }

    async downloadModel(model = 'multilingual') {
        if (!ALLOWED_MODELS.has(model)) throw new LayaRequestError('Unsupported Laya model', 400);
        await fs.mkdir(this.modelDir, { recursive: true });
        this.downloadState.phase = 'downloading';
        const progressTimer = setInterval(() => {
            directorySize(this.modelDir)
                .then(size => { this.downloadState.bytesDownloaded = size; })
                .catch(() => {});
        }, this.progressIntervalMs);
        progressTimer.unref?.();
        try {
            await this.request('/v1/systemone', {
                method: 'POST',
                timeoutMs: this.downloadTimeoutMs,
                body: JSON.stringify({
                    state: 'Bonjour',
                    model,
                    questions: {
                        language: {
                            type: 'choice',
                            instructions: 'Identify the language of this text.',
                            criteria: { fr: 'French', en: 'English', es: 'Spanish', other: 'Another language' }
                        }
                    }
                })
            });
            this.downloadState.phase = 'verifying';
            this.downloadState.bytesDownloaded = await directorySize(this.modelDir);
            const status = await this.status();
            this.downloadState = {
                ...this.downloadState,
                phase: 'ready',
                downloading: false,
                error: null,
                lastError: null
            };
            return { ...status, ...this.downloadState };
        } catch (error) {
            const visibleError = safeError(error);
            this.downloadState = {
                ...this.downloadState,
                phase: 'failed',
                downloading: false,
                error: visibleError,
                lastError: visibleError,
                bytesDownloaded: await directorySize(this.modelDir)
            };
            throw error;
        } finally {
            clearInterval(progressTimer);
        }
    }

    async deleteModel() {
        if (this.downloadState.downloading) throw new LayaRequestError('Cannot delete a model while downloading', 409);
        this.downloadState = { ...initialDownloadState(), phase: 'deleting' };
        try {
            await clearDirectoryContents(this.modelDir);
            if (this.cache) this.cache.clear();
            this.downloadState = initialDownloadState();
            return {
                ...this.downloadState,
                deleted: true,
                modelPresent: false,
                runtimeAvailable: false,
                sizeBytes: 0
            };
        } catch (error) {
            const visibleError = safeError(error);
            this.downloadState = {
                ...this.downloadState,
                phase: 'failed',
                error: visibleError,
                lastError: visibleError
            };
            throw error;
        }
    }

    cacheKey(text, mode, model) {
        return crypto.createHash('sha256').update(`${mode}\0${model}\0${text}`).digest('hex');
    }

    async detectBatch(payload) {
        validateBatchPayload(payload);
        const model = payload.model || 'multilingual';
        const mode = 'laya';
        const results = new Array(payload.items.length);
        const misses = [];

        payload.items.forEach((item, index) => {
            const key = this.cacheKey(item.text, mode, model);
            const cached = this.cache ? this.cache.get(key) : null;
            if (cached) results[index] = { ...cached, cached: true };
            else misses.push({ item, index, key });
        });

        if (misses.length) {
            const response = await this.request('/v1/systemone/batch', {
                method: 'POST',
                timeoutMs: this.batchTimeoutMs,
                body: JSON.stringify({
                    states: misses.map(({ item }) => item.text),
                    model,
                    questions: {
                        language: {
                            type: 'choice',
                            instructions: 'Identify the language of this text.',
                            criteria: { fr: 'French', en: 'English', es: 'Spanish', other: 'Another language' }
                        }
                    }
                })
            });
            if (!response || !Array.isArray(response.results) || response.results.length !== misses.length) {
                throw new LayaRequestError('Laya runtime returned an invalid batch response', 502);
            }
            misses.forEach(({ item, index, key }, responseIndex) => {
                const answer = response.results[responseIndex]?.answers?.language;
                const result = {
                    id: item.id ?? index,
                    language: answer?.choice || 'unknown',
                    confidence: answer?.answer_confidence ?? answer?.confidence ?? 0,
                    cached: false
                };
                results[index] = result;
                if (this.cache) this.cache.set(key, mode, model, { ...result, cached: undefined });
            });
        }
        return { runtimeAvailable: true, model, results };
    }

    async identityHintBatch(pairs) {
        if (!Array.isArray(pairs) || pairs.length === 0 || pairs.length > MAX_BATCH_ITEMS) {
            throw new LayaRequestError(`Identity hint batch must contain 1 to ${MAX_BATCH_ITEMS} pairs`, 400);
        }
        const response = await this.request('/v1/systemone/batch', {
            method: 'POST',
            timeoutMs: this.batchTimeoutMs,
            body: JSON.stringify({
                states: pairs.map(pair => ({
                    left: sanitizeIdentityDescriptor(pair.left),
                    right: sanitizeIdentityDescriptor(pair.right)
                })),
                model: 'multilingual',
                questions: {
                    sameIdentity: {
                        type: 'noul',
                        instructions: 'Do these two TV channel descriptors identify the same numbered channel?'
                    }
                }
            })
        });
        if (!response || !Array.isArray(response.results) || response.results.length !== pairs.length) {
            throw new LayaRequestError('Laya runtime returned an invalid identity batch response', 502);
        }
        return pairs.map((pair, index) => {
            const answer = response.results[index]?.answers?.sameIdentity || {};
            return {
                id: pair.id,
                sameIdentityProbability: Number(answer.noul) || 0,
                confidence: Number(answer.answer_confidence ?? answer.confidence) || 0,
                family: answer.family || null,
                number: answer.number || null
            };
        });
    }
}

module.exports = {
    ALLOWED_MODELS,
    BATCH_TIMEOUT_MS,
    LayaRequestError,
    LayaService,
    MODEL_DIR,
    clearDirectoryContents,
    directorySize,
    initialDownloadState,
    sanitizeIdentityDescriptor,
    validateBatchPayload
};