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 };