import { randomUUID } from 'node:crypto'; import { db as defaultDb } from './db.js'; import * as catalog from './video-catalog.js'; import * as innertube from './innertube.js'; import * as searchCache from './search-cache.js'; import { audit } from './audit-log.js'; const MAX_SEEDS = 200, MAX_PLAN = 500, MAX_ERRORS = 50; const clean = value => String(value || '').replace(/[\u0000-\u001f\u007f]/g, ' ').trim().slice(0, 200); const fold = value => String(value || '').normalize('NFD').replace(/[\u0300-\u036f]/g, '').toLowerCase().replace(/\s+/g, ' ').trim(); const parse = value => { try { return JSON.parse(value || '{}'); } catch { return {}; } }; const sleepDefault = ms => new Promise(resolve => setTimeout(resolve, ms)); export const DEFAULT_PRESETS = [ 'worship', 'praise', 'hymns', 'gospel', 'pagsamba', 'papuri', 'mga awit ng papuri', 'ebanghelyo', 'dayeg', 'pagdayeg', 'mga alawiton', 'maayong balita', 'adoración', 'alabanza', 'himnos', 'evangelio', 'Hillsong Worship', 'Elevation Worship', 'Bethel Music', 'Planetshakers', 'Don Moen', 'Sandi Patty', ]; export const DEFAULT_MODIFIERS = ['lyrics', 'live', 'acoustic', 'instrumental', 'karaoke']; export function expandQueryPlan({ seeds = [], modifiers = [], maxQueries = 80 } = {}) { const seedList = (Array.isArray(seeds) ? seeds : String(seeds).split(/\r?\n/)).map(clean).filter(Boolean).slice(0, MAX_SEEDS); const mods = (Array.isArray(modifiers) ? modifiers : String(modifiers).split(/[,\r\n]+/)).map(clean).filter(Boolean).slice(0, 30); const cap = Math.max(1, Math.min(MAX_PLAN, Number(maxQueries) || 1)); const out = [], seen = new Set(); for (const seed of seedList) for (const modifier of ['', ...mods]) { const query = clean(modifier ? `${seed} ${modifier}` : seed), key = fold(query); if (key && !seen.has(key)) { seen.add(key); out.push(query); if (out.length >= cap) return out; } } return out; } function publicJob(row, indexSize = 0) { const state = parse(row.state); const done = Number(state.queriesDone) || 0, total = JSON.parse(row.plan || '[]').length; const elapsed = Math.max(0, Date.now() - Number(row.created_at)); const etaMs = done ? Math.max(0, Math.round(elapsed / done * (total - done))) : null; return { id: row.id, status: row.status, plan: JSON.parse(row.plan || '[]'), options: parse(row.options), queriesDone: done, queriesTotal: total, current: state.current || '', newVideos: Number(state.newVideos) || 0, totalIndexSize: indexSize, errors: state.errors || [], log: state.log || [], etaMs, createdAt: Number(row.created_at), updatedAt: Number(row.updated_at), error: row.error || null }; } export function createDeepSearchService({ db = defaultDb, searchDeep = innertube.searchDeep, related = innertube.related, browseUploads = innertube.browseUploads, ingest = catalog.ingest, putSearch = searchCache.put, sleep = sleepDefault, random = Math.random, now = Date.now, autoStart = true } = {}) { // Accept the db.js module namespace as well as the client itself. if (typeof db?.execute !== 'function' && typeof db?.db?.execute === 'function') db = db.db; let worker = null; let timer = null; const log = (state, message) => { state.log ||= []; state.log.push({ at: now(), message: String(message).slice(0, 240) }); if (state.log.length > 100) state.log.splice(0, state.log.length - 100); }; async function indexSize() { return Number((await db.execute('SELECT COUNT(*) AS n FROM video_meta')).rows[0]?.n || 0); } async function getJob(id) { return (await db.execute({ sql: 'SELECT * FROM deep_search_jobs WHERE id=?', args: [id] })).rows[0] || null; } async function checkpoint(job, state, status = 'running', error = null) { const r = await db.execute({ sql: "UPDATE deep_search_jobs SET state=?,status=?,error=?,updated_at=? WHERE id=? AND status='running'", args: [JSON.stringify(state), status, error, now(), job.id] }); if (!r.rowsAffected) { const stop = new Error('Deep search paused or cancelled.'); stop.name = 'DeepSearchStopped'; throw stop; } } async function reserveDaily(cap) { const day = new Date(now()).toISOString().slice(0, 10); const r = await db.execute({ sql: `INSERT INTO deep_search_daily(day,queries) VALUES(?,1) ON CONFLICT(day) DO UPDATE SET queries=queries+1 WHERE queries c?.id).map(c => [c.id, c])).values()]; if (unique.length) { const validIds = unique.filter(c => /^[\w-]{11}$/.test(c.id)).map(c => c.id); const existing = validIds.length ? (await db.execute({ sql: `SELECT id FROM video_meta WHERE id IN (${validIds.map(() => '?').join(',')})`, args: validIds })).rows : []; const present = new Set(existing.map(row => row.id)); await ingest(unique, 'deep'); state.newVideos += validIds.filter(id => !present.has(id)).length; } await putSearch(query, (cacheCards || unique).slice(0, 200)); } async function requireDailyBudget() { const used = await reserveDaily(Math.max(1, Number(options.dailyCap) || 1000)); if (used !== null) return; state.dailyPausedDay = new Date(now()).toISOString().slice(0, 10); log(state, 'Daily query cap reached; job will resume tomorrow.'); await checkpoint(job, state, 'paused'); const stop = new Error('Daily cap reached.'); stop.name = 'DeepSearchStopped'; throw stop; } try { for (let qi = state.queriesDone; qi < plan.length; qi++) { const query = plan[qi]; if (state.current !== query) state.currentIds = []; state.current = query; await checkpoint(job, state); await requireDailyBudget(); await checkpoint(job, state); let cards, lastError; for (let attempt = 0; attempt < 3; attempt++) { try { cards = await searchDeep(query, { limit: Math.max(1, Math.min(200, Number(options.resultsPerQuery) || 50)), maxPages: 14, timeoutMs: 30000, pageDelayMs: Math.max(0, Number(options.delayMs) || 0), jitterMs: Math.max(0, Number(options.jitterMs) || 0), onPage: async pageCards => { await checkpoint(job, state); const saved = new Set(state.currentIds || []); const fresh = pageCards.filter(card => !saved.has(card.id)); await saveCards(query, fresh, pageCards); state.currentIds = pageCards.map(card => card.id); await checkpoint(job, state); } }); await checkpoint(job, state); lastError = null; break; } catch (error) { if (error.name === 'DeepSearchStopped') throw error; lastError = error; if (attempt < 2) await sleep(Math.min(60000, (Number(options.delayMs) || 10000) * 2 ** attempt + Math.floor(random() * 3000))); } } if (lastError) { state.errors.push({ query, error: String(lastError.message).slice(0, 200) }); if (state.errors.length > MAX_ERRORS) state.errors.shift(); log(state, `Search failed: ${query}`); } else { const saved = new Set(state.currentIds || []); const remaining = (cards || []).filter(card => !saved.has(card.id)); await saveCards(query, remaining, cards || []); state.currentIds = (cards || []).map(card => card.id); const depth = Math.max(0, Math.min(2, Number(options.relatedDepth) || 0)); if (depth) { let frontier = (cards || []).slice(0, Math.max(1, Math.min(10, Number(options.relatedTop) || 3))).map(c => ({ id: c.id, depth: 1 })); for (let level = 1; level <= depth && frontier.length; level++) { const next = []; for (const item of frontier) { if (state.relatedSeen.includes(item.id)) continue; state.relatedSeen.push(item.id); try { await sleep(Math.max(0, Number(options.delayMs) || 10000) + Math.floor(random() * (Number(options.jitterMs) || 3000))); await checkpoint(job, state); await requireDailyBudget(); const recs = await related(item.id); await checkpoint(job, state); await saveCards(`${query} related ${item.id}`, recs); if (level < depth) next.push(...recs.slice(0, 2).map(c => ({ id: c.id, depth: level + 1 }))); } catch (error) { if (error.name === 'DeepSearchStopped') throw error; state.errors.push({ query: `related:${item.id}`, error: String(error.message).slice(0, 200) }); } } frontier = next; } } log(state, `Saved ${cards?.length || 0} results for “${query}”.`); } state.queriesDone = qi + 1; state.current = ''; state.currentIds = []; await checkpoint(job, state); if (qi + 1 < plan.length) await sleep(Math.max(0, Number(options.delayMs) || 10000) + Math.floor(random() * (Number(options.jitterMs) || 3000))); } for (const source of [...(options.channels || []), ...(options.playlists || [])].slice(0, 50)) { state.current = `browse ${source}`; await checkpoint(job, state); try { await requireDailyBudget(); await checkpoint(job, state); const cards = await browseUploads(source, { limit: Math.max(1, Math.min(200, Number(options.resultsPerQuery) || 50)), pageDelayMs: options.delayMs, jitterMs: options.jitterMs }); await checkpoint(job, state); await saveCards(state.current, cards); log(state, `Browsed ${cards.length} videos from ${source}.`); } catch (error) { if (error.name === 'DeepSearchStopped') throw error; state.errors.push({ query: state.current, error: String(error.message).slice(0, 200) }); } await checkpoint(job, state); } state.current = ''; log(state, 'Deep search complete.'); await checkpoint(job, state, 'complete'); } catch (error) { if (error.name !== 'DeepSearchStopped') { state.errors.push({ query: state.current, error: String(error.message).slice(0, 200) }); await db.execute({ sql: "UPDATE deep_search_jobs SET state=?,status='failed',error=?,updated_at=? WHERE id=? AND status='running'", args: [JSON.stringify(state), String(error.message).slice(0, 300), now(), job.id] }); } } } async function runNext() { if (worker) return worker; worker = (async () => { // Recover jobs interrupted by process restart; the current query is safe to repeat. await db.execute({ sql: "UPDATE deep_search_jobs SET status='queued',updated_at=? WHERE status='running'", args: [now()] }); const today = new Date(now()).toISOString().slice(0, 10); const paused = (await db.execute("SELECT id,state FROM deep_search_jobs WHERE status='paused'")).rows; for (const row of paused) if (parse(row.state).dailyPausedDay && parse(row.state).dailyPausedDay < today) await db.execute({ sql: "UPDATE deep_search_jobs SET status='queued',updated_at=? WHERE id=? AND status='paused'", args: [now(), row.id] }); const job = (await db.execute({ sql: "SELECT * FROM deep_search_jobs WHERE status='queued' ORDER BY created_at LIMIT 1" })).rows[0]; if (job) { await db.execute({ sql: "UPDATE deep_search_jobs SET status='running',updated_at=? WHERE id=? AND status='queued'", args: [now(), job.id] }); const current = await getJob(job.id); if (current?.status === 'running') await run(current); } return !!job; })().finally(() => { worker = null; }); return worker; } const kick = () => runNext().catch(error => console.error('[deep search]', error.message)); function start() { if (!timer) { timer = setInterval(kick, 15000); timer.unref?.(); } kick(); } if (autoStart) start(); async function estimate(body) { const plan = expandQueryPlan(body); const browseSources = [...(Array.isArray(body?.channels) ? body.channels : []), ...(Array.isArray(body?.playlists) ? body.playlists : [])].map(clean).filter(Boolean).slice(0, 50).length; const operations = plan.length + browseSources + (Number(body?.relatedDepth) > 0 ? Math.min(plan.length * (Number(body?.relatedTop) || 3) * Math.min(2, Number(body.relatedDepth)), 100) : 0); const today = new Date(now()).toISOString().slice(0, 10); const used = Number((await db.execute({ sql: 'SELECT queries FROM deep_search_daily WHERE day=?', args: [today] })).rows[0]?.queries || 0); return { plan, queries: plan.length, browseSources, estimatedOperations: operations, estimatedMinutes: Math.ceil(operations * Math.max(1, Number(body?.delaySeconds) || 10) / 60), indexSize: await indexSize(), dailyUsed: used, dailyRemaining: Math.max(0, (Number(body?.dailyCap) || 1000) - used) }; } function register(app, adminAuth) { app.post('/api/admin/deep-search/estimate', adminAuth, async c => { let body = {}; try { body = await c.req.json(); } catch {} return c.json({ ok: true, ...(await estimate(body)) }); }); app.get('/api/admin/deep-search/jobs', adminAuth, async c => c.json({ ok: true, indexSize: await indexSize(), jobs: (await db.execute('SELECT * FROM deep_search_jobs ORDER BY created_at DESC LIMIT 30')).rows.map(row => publicJob(row)) })); app.post('/api/admin/deep-search/start', adminAuth, async c => { let body; try { body = await c.req.json(); } catch { return c.json({ ok: false, error: 'Invalid JSON.' }, 400); } const plan = expandQueryPlan(body); if (!plan.length) return c.json({ ok: false, error: 'Add at least one seed term.' }, 400); const active = (await db.execute("SELECT id FROM deep_search_jobs WHERE status IN ('queued','running','paused') LIMIT 1")).rows[0]; if (active) return c.json({ ok: false, error: 'A deep-search job is already active.' }, 409); const options = { resultsPerQuery: Math.max(1, Math.min(200, Number(body.resultsPerQuery) || 50)), relatedDepth: Math.max(0, Math.min(2, Number(body.relatedDepth) || 0)), relatedTop: Math.max(1, Math.min(10, Number(body.relatedTop) || 3)), delayMs: Math.max(1000, Math.min(300000, (Number(body.delaySeconds) || 10) * 1000)), jitterMs: Math.max(0, Math.min(300000, (Number(body.jitterSeconds) || 3) * 1000)), dailyCap: Math.max(1, Math.min(10000, Number(body.dailyCap) || 1000)), channels: (Array.isArray(body.channels) ? body.channels : []).map(clean).filter(Boolean).slice(0, 25), playlists: (Array.isArray(body.playlists) ? body.playlists : []).map(clean).filter(Boolean).slice(0, 25) }; const id = randomUUID(), time = now(); await db.execute({ sql: "INSERT INTO deep_search_jobs(id,status,plan,options,state,created_at,updated_at) VALUES(?,'queued',?,?,?,?,?)", args: [id, JSON.stringify(plan), JSON.stringify(options), JSON.stringify({ queriesDone: 0, newVideos: 0, errors: [], log: [] }), time, time] }); await audit(db, 'admin', 'deep-search.start', id, { queries: plan.length }); if (autoStart) kick(); return c.json({ ok: true, id }, 202); }); for (const action of ['pause', 'resume', 'cancel']) app.post(`/api/admin/deep-search/:id/${action}`, adminAuth, async c => { const id = c.req.param('id'); const allowed = action === 'pause' ? ['queued', 'running'] : action === 'resume' ? ['paused'] : ['queued', 'running', 'paused']; const target = action === 'pause' ? 'paused' : action === 'resume' ? 'queued' : 'cancelled'; const marks = allowed.map(() => '?').join(','); const r = await db.execute({ sql: `UPDATE deep_search_jobs SET status=?,updated_at=? WHERE id=? AND status IN (${marks})`, args: [target, now(), id, ...allowed] }); if (r.rowsAffected) await audit(db, 'admin', `deep-search.${action}`, id); if (action === 'resume' && r.rowsAffected && autoStart) kick(); return c.json({ ok: !!r.rowsAffected, status: r.rowsAffected ? target : undefined }, r.rowsAffected ? 200 : 409); }); return { estimate, runNext, kick, start }; } return { estimate, runNext, kick, start, register }; } export function registerDeepSearchRoutes(app, options = {}) { const service = createDeepSearchService(options); service.register(app, options.adminAuth); return service; }