Add resumable admin deep search

This commit is contained in:
Jonathan Sykes
2026-10-14 01:47:35 +08:00
parent ca10fe6486
commit 58d0e6f140
12 changed files with 600 additions and 22 deletions

122
server/deep-search.test.js Normal file
View File

@@ -0,0 +1,122 @@
import { expect, test, beforeAll } from 'bun:test';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { Hono } from 'hono';
const root = mkdtempSync(join(tmpdir(), 'ytp-deep-search-'));
process.env.DB_PATH = join(root, 'test.db');
const { initDb, db } = await import('./db.js');
const catalog = await import('./video-catalog.js');
const searchCache = await import('./search-cache.js');
const { createDeepSearchService, expandQueryPlan } = await import('./deep-search.js');
const auth = async (_c, next) => next();
const sample = { id: 'd33psrch001', title: 'Worship in Cebuano', channel: 'Hope Choir' };
beforeAll(async () => { await initDb(); });
function makeApp(service) {
const app = new Hono(); service.register(app, auth); return app;
}
async function request(app, path, method = 'GET', body) {
const response = await app.request(path, { method, headers: body ? { 'content-type': 'application/json' } : {}, body: body ? JSON.stringify(body) : undefined });
return { response, json: await response.json() };
}
const cleanDb = async () => {
await db.batch(['DELETE FROM deep_search_jobs', 'DELETE FROM deep_search_daily', 'DELETE FROM video_meta', 'DELETE FROM video_meta_sources', 'DELETE FROM search_cache']);
};
test.serial('query plan expands modifiers, folds accents and deduplicates', () => {
expect(expandQueryPlan({ seeds: ['Adoración', 'adoracion', ' Gospel '], modifiers: ['live', 'LIVE'], maxQueries: 30 }))
.toEqual(['Adoración', 'Adoración live', 'Gospel', 'Gospel live']);
expect(expandQueryPlan({ seeds: ['one', 'two'], modifiers: ['live'], maxQueries: 2 })).toHaveLength(2);
});
test.serial('estimate and background job ingest to catalog, source and search cache', async () => {
await cleanDb();
const service = createDeepSearchService({ db, ingest: catalog.ingest, putSearch: searchCache.put,
searchDeep: async () => [sample], sleep: async () => {}, autoStart: false });
const app = makeApp(service);
const estimate = await request(app, '/api/admin/deep-search/estimate', 'POST', { seeds: ['worship'], modifiers: ['live','lyrics'], maxQueries: 4 });
expect(estimate.json.plan).toEqual(['worship', 'worship live', 'worship lyrics']);
const created = await request(app, '/api/admin/deep-search/start', 'POST', { seeds: ['worship'], modifiers: [], maxQueries: 1, delaySeconds: 1 });
expect(created.response.status).toBe(202);
await service.runNext();
const jobs = await request(app, '/api/admin/deep-search/jobs');
expect(jobs.json.jobs[0].status).toBe('complete');
expect(jobs.json.jobs[0].newVideos).toBe(1);
const row = (await db.execute({ sql: 'SELECT source FROM video_meta_sources WHERE video_id=?', args: [sample.id] })).rows[0];
expect(row.source).toBe('deep');
expect((await searchCache.get('worship')).results[0].id).toBe(sample.id);
});
test.serial('pause during upstream call does not ingest; resume continues from the persisted query', async () => {
await cleanDb(); let begin, release, calls = 0;
const began = new Promise(resolve => { begin = resolve; });
const gate = new Promise(resolve => { release = resolve; });
const service = createDeepSearchService({ db, ingest: catalog.ingest, putSearch: searchCache.put, autoStart: false,
sleep: async () => {}, searchDeep: async () => { calls++; if (calls === 1) { begin(); await gate; } return [sample]; } });
const app = makeApp(service);
const created = await request(app, '/api/admin/deep-search/start', 'POST', { seeds: ['worship'], maxQueries: 1, delaySeconds: 1 });
const work = service.runNext(); await began;
await request(app, `/api/admin/deep-search/${created.json.id}/pause`, 'POST', {});
release(); await work;
expect((await db.execute({ sql: 'SELECT COUNT(*) AS n FROM video_meta WHERE id=?', args: [sample.id] })).rows[0].n).toBe(0);
let jobs = await request(app, '/api/admin/deep-search/jobs'); expect(jobs.json.jobs[0].status).toBe('paused');
await request(app, `/api/admin/deep-search/${created.json.id}/resume`, 'POST', {});
await service.runNext();
jobs = await request(app, '/api/admin/deep-search/jobs');
expect(jobs.json.jobs[0].status).toBe('complete'); expect(calls).toBe(2);
});
test.serial('cancelled in-flight job does not write its late result', async () => {
await cleanDb(); let begin, release;
const began = new Promise(resolve => { begin = resolve; }), gate = new Promise(resolve => { release = resolve; });
const service = createDeepSearchService({ db, autoStart: false, sleep: async () => {},
searchDeep: async () => { begin(); await gate; return [sample]; }, ingest: catalog.ingest, putSearch: searchCache.put });
const app = makeApp(service);
const created = await request(app, '/api/admin/deep-search/start', 'POST', { seeds: ['worship'], maxQueries: 1, delaySeconds: 1 });
const work = service.runNext(); await began;
await request(app, `/api/admin/deep-search/${created.json.id}/cancel`, 'POST', {}); release(); await work;
expect((await db.execute({ sql: 'SELECT COUNT(*) AS n FROM video_meta WHERE id=?', args: [sample.id] })).rows[0].n).toBe(0);
const jobs = await request(app, '/api/admin/deep-search/jobs'); expect(jobs.json.jobs[0].status).toBe('cancelled');
});
test.serial('page checkpoint survives a retry without double-counting its videos', async () => {
await cleanDb(); let calls = 0;
const service = createDeepSearchService({ db, autoStart: false, sleep: async () => {}, ingest: catalog.ingest, putSearch: searchCache.put,
searchDeep: async (_query, { onPage }) => {
calls++;
await onPage([sample]);
if (calls === 1) throw Error('connection reset after first page');
return [sample];
} });
const app = makeApp(service);
await request(app, '/api/admin/deep-search/start', 'POST', { seeds: ['worship'], maxQueries: 1, delaySeconds: 1 });
await service.runNext();
const row = (await db.execute({ sql: 'SELECT seen FROM video_meta WHERE id=?', args: [sample.id] })).rows[0];
expect(Number(row.seen)).toBe(1);
});
test.serial('daily cap pauses and worker resumes it on the next UTC day', async () => {
await cleanDb(); let time = Date.parse('2026-10-14T23:59:00Z');
const service = createDeepSearchService({ db, now: () => time, autoStart: false, sleep: async () => {},
searchDeep: async query => [{ ...sample, id: query === 'worship' ? 'd33psrch001' : 'd33psrch002' }],
ingest: catalog.ingest, putSearch: searchCache.put });
const app = makeApp(service);
const created = await request(app, '/api/admin/deep-search/start', 'POST', { seeds: ['worship','praise'], maxQueries: 2, delaySeconds: 1, dailyCap: 1 });
await service.runNext();
let jobs = await request(app, '/api/admin/deep-search/jobs'); expect(jobs.json.jobs[0].status).toBe('paused');
time += 120000; await service.runNext(); jobs = await request(app, '/api/admin/deep-search/jobs');
expect(jobs.json.jobs[0].status).toBe('complete'); expect(jobs.json.jobs[0].queriesDone).toBe(2);
});
test.serial('only one queued/running job is accepted', async () => {
await cleanDb();
const service = createDeepSearchService({ db, autoStart: false }); const app = makeApp(service);
const body = { seeds: ['worship'], maxQueries: 1 };
expect((await request(app, '/api/admin/deep-search/start', 'POST', body)).response.status).toBe(202);
expect((await request(app, '/api/admin/deep-search/start', 'POST', body)).response.status).toBe(409);
});
test.serial('cleanup test database', () => { rmSync(root, { recursive: true, force: true }); });