openvibers/openvibe.live Commits
VODs and clips in OpenVibe.Search (WS-O task 10, Contracts 0.45.0): Live owns their canonical /vod and /clip pages, so it sends live.index_document type vod and clip (title, channel, category, Live's AI overview and transcript), read from Media, never the frozen tables; re-checked right after a change through /api/vods or /api/clips and when Media announces an item ready or failed, a 5-minute scan of the newest, and a daily refresh that removes what Media stopped listing (never after a partial listing); AI clips and NSFW streams are noindex; private, unlisted, deleted or failed items get one tombstone
Changes
@@@ -25,7 +25,7 @@"multer": "^1.4.5-lts.1","node-media-server": "^2.7.1","onnxruntime-node": "^1.20.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.45.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.9.1","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.12.0","sanitize-html": "^2.17.7",@@@ -2814,9 +2814,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.44.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.0",-- "integrity": "sha512-VMjnJSwU2dnk9KstCHQDKfY6UysSk2fWK1wdLp15Via49/hbeaXRTa4XIjO3d82fYizj0rJpi9GrK+AJRskjtQ==",++ "version": "0.45.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.45.0",++ "integrity": "sha512-6JJL9gHKd+gzowNzUls7l5nV0awR/KmEXxaxUaxzUPZilVFlDSM6yFpXMjFp+pDrUXDHYI5kfMevhpI0GEBFEA==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -43,7 +43,7 @@"multer": "^1.4.5-lts.1","node-media-server": "^2.7.1","onnxruntime-node": "^1.20.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.45.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.9.1","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.12.0","sanitize-html": "^2.17.7",nnew file mode 100644@@@ -0,0 +1,237 @@++'use strict';++/**++ * Live's VOD and clip pages in OpenVibe.Search (roadmap WS-O task 10; Contracts 0.45.0++ * live.index_document.* types vod and clip): Live owns the canonical /vod/<id> and /clip/<id> pages of++ * the Media items recorded from its streams (title, channel, AI overview, transcript), so Live sends++ * their Search documents through its transactional outbox (./stream-events.js), with a tombstone once++ * an item is deleted, made private or unlisted, or failed. Items come from Media (media-client), never++ * from the frozen local vods/clips tables; the AI overview and transcript are Live's vod_ai_state and++ * clip_ai_state.++ *++ * touch(kind, id) right after Live changes an item (the /api/vods and /api/clips routes) or Media++ * announces it ready or failed (server/media-proxy/outcomes.js)++ * scan() every 5 minutes: the newest public VODs and clips++ * refresh() daily: every public VOD and clip, and a tombstone for each item Search holds that++ * Media no longer lists as public (checked one by one before it is removed)++ *++ * A document is sent only when it changed (search_media_pushes keeps a hash and the revision, which++ * grows by one with every document or tombstone). AI clips and clips or VODs of NSFW streams are++ * indexed noindex, as the pages are. Off while the outbox is off (EVENTS_URL unset).++ */++const crypto = require('crypto');++const db = require('../db/database');++const streamEvents = require('./stream-events');++++const PUBLIC_BASE = (process.env.LIVE_PUBLIC_ORIGIN || 'https://openvibe.live').replace(/\/+$/, '');++const DAY_MS = 86400000;++const PAGE = 200;++const TOUCH_DELAY_MS = Number(process.env.LIVE_SEARCH_TOUCH_DELAY_MS) || 1500;++const KINDS = ['vod', 'clip'];++const stats = { sent: 0, tombstones: 0, unchanged: 0, lastError: null };++++let ready = false;++function ensureSchema() {++ if (ready) return;++ db.getDb().exec(`CREATE TABLE IF NOT EXISTS search_media_pushes (++ kind TEXT NOT NULL,++ media_id INTEGER NOT NULL,++ hash TEXT NOT NULL,++ revision INTEGER NOT NULL,++ deleted INTEGER NOT NULL DEFAULT 0,++ pushed_at DATETIME DEFAULT CURRENT_TIMESTAMP,++ PRIMARY KEY (kind, media_id)++ )`);++ ready = true;++}++++const iso = (v) => {++ if (!v) return null;++ const d = new Date(String(v).includes('T') ? v : `${String(v).replace(' ', 'T')}Z`);++ return Number.isNaN(d.getTime()) ? null : d.toISOString();++};++const clean = (s, n) => String(s || '').replace(/\s+/g, ' ').trim().slice(0, n);++const isAiClip = (c) => c.auto_generated === true || Number(c.auto_generated) === 1;++++function aiState(kind, id) {++ const st = kind === 'clip' ? db.getClipAiState(id) : db.getVodAiState(id);++ if (!st) return {};++ let transcript = '';++ try { transcript = (JSON.parse(st.ai_transcript_json || '[]') || []).map((s) => s && s.text).filter(Boolean).join(' '); } catch { /* unreadable: none */ }++ return { overview: st.ai_overview || st.ai_overview_short || '', short: st.ai_overview_short || '', transcript };++}++++/** Would Live's page for this item be public? (the same rule as its SSR meta and the Media sitemap) */++function isListable(kind, row) {++ if (!row || row.id == null) return false;++ const visibility = row.visibility || (row.is_public ? 'public' : 'private');++ if (visibility !== 'public' || row.is_public === false || Number(row.is_public) === 0) return false;++ if (kind === 'vod' && (row.is_recording || row.clips_only || row.status !== 'ready')) return false;++ if (kind === 'clip' && (row.status || 'ready') !== 'ready') return false;++ if (row.readiness && row.readiness.playable === false) return false;++ return true;++}++++/** The document for one Media item, or `{ deleted: true }` when Search must not hold it. */++function documentFor(kind, row) {++ if (!isListable(kind, row)) return { deleted: true };++ const d = db.getDb();++ const ownerId = kind === 'clip' ? (row.channel_user_id || row.user_id) : row.user_id;++ const owner = ownerId != null ? d.prepare('SELECT id, username, display_name, is_banned FROM users WHERE id = ?').get(Number(ownerId)) : null;++ if (owner && owner.is_banned) return { deleted: true };++ const stream = row.stream_id != null ? d.prepare('SELECT category, is_nsfw FROM streams WHERE id = ?').get(Number(row.stream_id)) : null;++ const ch = owner ? d.prepare('SELECT chat_language FROM channels WHERE user_id = ?').get(owner.id) : null;++ const name = owner ? (owner.display_name || owner.username) : null;++ const ai = aiState(kind, row.id);++ const aiClip = kind === 'clip' && isAiClip(row);++ const fallbackTitle = kind === 'vod' ? `${name ? `${name}'s ` : ''}stream VOD` : (aiClip ? 'AI clip' : 'Clip');++ const summary = clean(ai.short || ai.overview || row.description, 4000)++ || `${kind === 'vod' ? 'Recorded live stream' : aiClip ? 'AI clip of a live stream' : 'Clip of a live stream'}${name ? ` by ${name}` : ''} on OpenVibe.Live.`;++ const facets = { duration_seconds: Math.round(Number(row.duration_seconds ?? row.duration) || 0) };++ if (owner) facets.channel = owner.username;++ if (stream && stream.category) facets.category = String(stream.category).slice(0, 200);++ if (kind === 'clip') { facets.ai_clip = aiClip; if (row.vod_id != null) facets.vod_id = String(row.vod_id); }++ const reasons = [...(aiClip ? ['ai_unreviewed'] : []), ...(stream && stream.is_nsfw ? ['sensitive'] : [])];++ const doc = {++ owner: 'live', type: kind, id: String(row.id), deleted: false, visibility: 'public',++ canonical_url: `${PUBLIC_BASE}/${kind}/${row.id}`,++ title: clean(row.title || fallbackTitle, 500),++ summary,++ body: [clean(row.description, 4000), clean(ai.overview, 8000), ai.transcript].filter(Boolean).join('\n').slice(0, 40000),++ facets, authorship: aiClip ? 'ai_generated' : 'human', publication_state: 'published', published_at: iso(row.created_at),++ indexability: reasons.length ? { decision: 'noindex', reasons } : { decision: 'index' },++ };++ if (!doc.body) delete doc.body;++ if (ch && /^[a-z]{2,3}$/.test(String(ch.chat_language || ''))) doc.language = ch.chat_language;++ return doc;++}++++const hashOf = (doc) => crypto.createHash('sha256').update(JSON.stringify(doc)).digest('hex').slice(0, 32);++++/** Send one item's document or tombstone when it changed. `row` null: gone. → 'sent' | 'tombstone' | 'unchanged' | 'skipped' */++function publish(kind, id, row, { now = Date.now() } = {}) {++ ensureSchema();++ if (!streamEvents.status().enabled || !KINDS.includes(kind)) return 'skipped';++ const mediaId = Number(id);++ if (!Number.isSafeInteger(mediaId) || mediaId < 1) return 'skipped';++ const doc = row ? documentFor(kind, row) : { deleted: true };++ const d = db.getDb();++ const prev = d.prepare('SELECT hash, revision, deleted FROM search_media_pushes WHERE kind = ? AND media_id = ?').get(kind, mediaId);++ if (doc.deleted && (!prev || prev.deleted)) { stats.unchanged++; return 'unchanged'; } // never sent, or already gone++ const hash = doc.deleted ? 'deleted' : hashOf(doc);++ if (prev && prev.hash === hash) { stats.unchanged++; return 'unchanged'; }++ const revision = (prev ? prev.revision : 0) + 1;++ const sid = String(mediaId);++ d.transaction(() => {++ streamEvents.enqueue(doc.deleted++ ? { event_type: 'live.index_document.deleted', actor: { type: 'service', id: 'live' }, subject: { type: kind, id: sid, revision }, visibility: 'internal', priority: 'low', payload: { type: kind, id: sid, revision } }++ : { event_type: 'live.index_document.upserted', actor: { type: 'service', id: 'live' }, subject: { type: kind, id: sid, revision }, visibility: 'internal', priority: 'low',++ payload: { ...doc, revision, updated_at: new Date(now).toISOString() } });++ d.prepare(`INSERT INTO search_media_pushes (kind, media_id, hash, revision, deleted, pushed_at) VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)++ ON CONFLICT(kind, media_id) DO UPDATE SET hash = excluded.hash, revision = excluded.revision, deleted = excluded.deleted, pushed_at = excluded.pushed_at`)++ .run(kind, mediaId, hash, revision, doc.deleted ? 1 : 0);++ })();++ streamEvents.kick();++ if (doc.deleted) { stats.tombstones++; return 'tombstone'; }++ stats.sent++;++ return 'sent';++}++++let mediaImpl = null;++const mediaClient = () => mediaImpl || require('../media-client');++const fetchOne = (kind, id, media) => (kind === 'vod' ? media.getVod(id, { timeoutMs: 10000 }) : media.getClip(id, { timeoutMs: 10000 }));++++/** Re-read one item from Media and send what changed. A 404 is a deletion; any other failure changes nothing. */++async function touch(kind, id, { media = mediaClient(), now = Date.now() } = {}) {++ if (!streamEvents.status().enabled || !KINDS.includes(kind)) return 'skipped';++ let row;++ try {++ row = await fetchOne(kind, id, media);++ } catch (err) {++ if (err && err.status === 404) row = null;++ else { stats.lastError = `${kind} ${id}: ${err && err.message}`; return 'skipped'; }++ }++ return publish(kind, id, row && (row.vod || row.clip || row), { now });++}++++/** Fire-and-forget touch for route and event hooks: never throws, never delays the caller. */++function touchLater(kind, ids) {++ if (!streamEvents.status().enabled || process.env.LIVE_SEARCH_DOCUMENTS === 'off') return;++ const list = [...new Set((Array.isArray(ids) ? ids : [ids]).map((v) => parseInt(v, 10)).filter((v) => v > 0))].slice(0, 500);++ if (!list.length) return;++ setTimeout(async () => {++ for (const id of list) { try { await touch(kind, id); } catch (err) { stats.lastError = err.message; } }++ }, TOUCH_DELAY_MS).unref();++}++++async function listPage(kind, offset, media, limit = PAGE) {++ const r = kind === 'vod' ? await media.listVods({ limit, offset }, { timeoutMs: 20000 }) : await media.listClips({ limit, offset }, { timeoutMs: 20000 });++ return { rows: (r && (kind === 'vod' ? r.vods : r.clips)) || [], hasMore: !!(r && r.hasMore) };++}++++/** The newest public items (Media lists newest first). */++async function scan({ media = mediaClient(), now = Date.now(), limit = 50 } = {}) {++ if (!streamEvents.status().enabled) return 0;++ let n = 0;++ for (const kind of KINDS) {++ try {++ const { rows } = await listPage(kind, 0, media, limit);++ for (const row of rows) { publish(kind, row.id, row, { now }); n++; }++ } catch (err) { stats.lastError = `${kind} scan: ${err.message}`; }++ }++ return n;++}++++/** Every public item, then a check of each one Search holds that the listing no longer shows. */++async function refresh({ media = mediaClient(), now = Date.now(), maxPages = 100 } = {}) {++ if (!streamEvents.status().enabled) return 0;++ ensureSchema();++ let n = 0;++ for (const kind of KINDS) {++ const seen = new Set();++ let complete = false;++ try {++ for (let page = 0, offset = 0; page < maxPages; page++, offset += PAGE) {++ const { rows, hasMore } = await listPage(kind, offset, media);++ for (const row of rows) { seen.add(Number(row.id)); publish(kind, row.id, row, { now }); n++; }++ if (!hasMore || !rows.length) { complete = true; break; }++ }++ } catch (err) { stats.lastError = `${kind} refresh: ${err.message}`; }++ if (!complete) continue; // a partial listing never removes anything++ const held = db.getDb().prepare('SELECT media_id FROM search_media_pushes WHERE kind = ? AND deleted = 0').all(kind).map((r) => r.media_id);++ for (const id of held) if (!seen.has(id)) await touch(kind, id, { media, now });++ }++ return n;++}++++function init() {++ if (!streamEvents.status().enabled || process.env.LIVE_SEARCH_DOCUMENTS === 'off') return false;++ ensureSchema();++ const jobs = require('../utils/jobs');++ jobs.every('search-media-scan', 5 * 60 * 1000, () => scan(), { initialDelayMs: 3 * 60 * 1000, jitterMs: 15 * 1000 });++ jobs.every('search-media-refresh', DAY_MS, () => refresh(), { initialDelayMs: 6 * 60 * 1000, jitterMs: 60 * 1000 });++ return true;++}++++/**++ * Express middleware for /api/vods and /api/clips: after a successful change (not a GET), re-check the++ * items it named (the numeric :id in the path, or `ids` in a bulk body).++ */++function afterChange(kind) {++ return (req, res, next) => {++ if (req.method === 'GET' || req.method === 'HEAD') return next();++ res.on('finish', () => {++ if (res.statusCode < 200 || res.statusCode >= 300) return;++ const m = String(req.path || '').match(/^\/(?:clips\/)?(\d+)(?:\/|$)/);++ const ids = m ? [m[1]] : (req.body && Array.isArray(req.body.ids) ? req.body.ids : []);++ const k = kind === 'vod' && /^\/clips\//.test(String(req.path || '')) ? 'clip' : kind;++ touchLater(k, ids);++ });++ next();++ };++}++++function status() { return { ...stats }; }++++function _setMedia(m) { mediaImpl = m; }++++module.exports = { init, ensureSchema, isListable, documentFor, publish, touch, touchLater, scan, refresh, afterChange, status, _setMedia };@@@ -676,8 +676,10 @@ app.use('/api/funds', monetizationRoutes);app.use('/api/coins', coinsRoutes);app.use('/api/payments', require('./monetization/payments-routes'));app.use('/api/cosmetics', cosmeticsRoutes);--app.use('/api/vods', vodRoutes);--app.use('/api/clips', clipRoutes);++// Live's VOD and clip pages in OpenVibe.Search follow every change made here (events/search-media-documents.js).++const searchMedia = require('./events/search-media-documents');++app.use('/api/vods', searchMedia.afterChange('vod'), vodRoutes);++app.use('/api/clips', searchMedia.afterChange('clip'), clipRoutes);app.use('/api/chat-ai', require('./ai/chat-ai-routes'));app.use('/api/easter-egg', require('./ai/easter-egg-routes'));app.use('/api/arena', require('./arena/routes')); // streamer vs streamer (docs/arena.md)@@@ -1558,6 +1560,7 @@ async function start() {try { require('./events/stream-events').init(); } catch (err) { console.warn('[Events] not started:', err.message); }// Channels in OpenVibe.Search (WS-O task 10): live.index_document.* through the same outbox.try { require('./events/search-documents').init(); } catch (err) { console.warn('[Search] channel documents not started:', err.message); }++ try { require('./events/search-media-documents').init(); } catch (err) { console.warn('[Search] VOD and clip documents not started:', err.message); }// 8d. User modules on Network (Contracts 0.41.0): live.profile and live.stats, written when they// change (server/auth/module-summaries.js). Off without OV_OAUTH_CLIENT_SECRET.@@@ -206,6 +206,9 @@ function handle({ via, event, data, eventId = null, subject = null }) {}stats.applied[via]++;if (after) { try { after(); } catch (e) { console.warn('[MediaOutcome] post-commit step failed:', e.message); } }++ // A VOD or clip that became ready (or failed) changes what OpenVibe.Search should hold for its page.++ const kind = /^(vod|clip)\.(ready|failed)$/.exec(String(event));++ if (kind && data && data.id != null) { try { require('../events/search-media-documents').touchLater(kind[1], data.id); } catch { /* search is optional */ } }return { applied: true, duplicate: false, outcome: out.result };}nnew file mode 100644@@@ -0,0 +1,150 @@++'use strict';++// Live's VOD and clip pages in OpenVibe.Search (WS-O task 10, Contracts 0.45.0): a public, ready VOD goes++// through the outbox as live.index_document.upserted type vod (valid against the contract, canonical on++// Live, with Live's AI overview and transcript), again only when it changed; a recording VOD is never++// sent; a VOD made private gets one tombstone; an AI clip and a clip of an NSFW stream are noindex; a 404++// is a deletion; the daily refresh removes what Media stopped listing, but never after a partial listing;++// a change through /api/vods is re-checked right after it succeeds.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const { EventEmitter } = require('events');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-live-searchmedia-'));++process.env.DB_PATH = path.join(tmp, 'live.db');++process.env.LIVE_SEARCH_TOUCH_DELAY_MS = '10';++const quiet = console.log;++console.log = (...a) => { if (!/^\[/.test(String(a[0]))) quiet(...a); };++console.warn = () => {};++++const published = [];++const stub = http.createServer((req, res) => {++ let body = '';++ req.on('data', (c) => { body += c; });++ req.on('end', () => {++ res.setHeader('Content-Type', 'application/json');++ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));++ if (req.url === '/api/v1/events' && req.method === 'POST') {++ const parsed = JSON.parse(body);++ const list = parsed.events || [parsed];++ const { validate } = require('openvibe-contracts');++ const bad = list.find((e) => !validate('events.event-envelope@1', e).valid);++ if (bad) { res.statusCode = 422; return res.end(JSON.stringify({ code: 'events.invalid_envelope' })); }++ const results = list.map((e) => { published.push(e); return { event_id: e.event_id, seq: published.length, duplicate: false }; });++ return res.end(JSON.stringify(parsed.events ? { results } : results[0]));++ }++ res.statusCode = 404; res.end('{}');++ });++});++++// Media as Live's service sees it: listings hold public items only; get returns any of Live's.++const items = { vod: new Map(), clip: new Map() };++let listFails = false;++const notFound = () => Object.assign(new Error('not found'), { status: 404 });++const media = {++ async listVods({ limit, offset }) { if (listFails) throw new Error('Media unreachable'); const all = [...items.vod.values()].filter((v) => v.visibility === 'public' && !v.is_recording); return { vods: all.slice(offset, offset + limit), hasMore: offset + limit < all.length }; },++ async listClips({ limit, offset }) { if (listFails) throw new Error('Media unreachable'); const all = [...items.clip.values()].filter((c) => c.visibility === 'public'); return { clips: all.slice(offset, offset + limit), hasMore: offset + limit < all.length }; },++ async getVod(id) { const v = items.vod.get(Number(id)); if (!v) throw notFound(); return v; },++ async getClip(id) { const c = items.clip.get(Number(id)); if (!c) throw notFound(); return c; },++};++const vod = (id, extra = {}) => ({ id, app_id: 'live', user_id: 1, stream_id: 1, title: 'Building a forum', description: '', status: 'ready', visibility: 'public', is_public: true, is_recording: false, clips_only: false, duration_seconds: 5400, created_at: '2026-09-24 19:00:00', readiness: { playable: true }, ...extra });++++(async () => {++ await new Promise((r) => stub.listen(0, '127.0.0.1', r));++ const base = `http://127.0.0.1:${stub.address().port}`;++ process.env.OV_NETWORK_INTERNAL_URL = base;++ const { validate } = require('openvibe-contracts');++ const db = require('../server/db/database');++ db.initDb();++ const d = db.getDb();++ d.prepare("INSERT INTO users (id, username, display_name, password_hash) VALUES (1, 'alex', 'Alex', 'x'), (2, 'viewer', 'Viewer', 'x')").run();++ const s1 = db.createStream({ user_id: 1, title: 'Building a forum', category: 'tech', protocol: 'webrtc', is_nsfw: 0 }).lastInsertRowid;++ const s2 = db.createStream({ user_id: 1, title: 'Late night', category: 'irl', protocol: 'webrtc', is_nsfw: 1 }).lastInsertRowid;++ const streamEvents = require('../server/events/stream-events');++ const outbox = streamEvents.init({ eventsUrl: base, clientSecret: 's3cret', intervalMs: 50 });++ const docs = require('../server/events/search-media-documents');++ docs._setMedia(media);++ const index = () => published.filter((e) => /^live\.index_document\./.test(e.event_type));++ const valid = (ev) => { const v = validate(ev.event_type + '@1', ev.payload); assert.ok(v.valid, `${ev.event_type}: ${JSON.stringify(v.errors)}`); };++++ // A public, ready VOD with Live's AI overview and transcript.++ items.vod.set(912, vod(912, { stream_id: s1 }));++ db.setVodAiOverview(912, 'Alex builds the rating menu of the forum and fixes it on phones.');++ d.prepare('UPDATE vod_ai_state SET ai_transcript_json = ? WHERE vod_id = 912').run(JSON.stringify([{ text: 'okay so today' }, { text: 'we are building the forum' }]));++ items.vod.set(913, vod(913, { is_recording: true, status: 'recording' }));++ assert.strictEqual(await docs.scan(), 1);++ assert.strictEqual(await docs.touch('vod', 913), 'unchanged', 'a recording is never sent');++ await outbox.flush();++ let ev = index()[0];++ assert.strictEqual(ev.event_type, 'live.index_document.upserted'); valid(ev);++ assert.deepStrictEqual(ev.subject, { type: 'vod', id: '912', revision: 1 });++ assert.strictEqual(ev.payload.canonical_url, 'https://openvibe.live/vod/912');++ assert.ok(ev.payload.summary.startsWith('Alex builds the rating menu'));++ assert.ok(ev.payload.body.includes('we are building the forum'), 'the transcript is searchable');++ assert.deepStrictEqual(ev.payload.facets, { duration_seconds: 5400, channel: 'alex', category: 'tech' });++ assert.strictEqual(ev.payload.indexability.decision, 'index');++ assert.strictEqual(docs.publish('vod', 912, items.vod.get(912)), 'unchanged', 'nothing changed: nothing sent');++++ // An AI clip, and a person's clip of an NSFW stream.++ items.clip.set(3107, { id: 3107, app_id: 'live', vod_id: 912, stream_id: s1, channel_user_id: 1, user_id: null, title: 'Chat erupts', status: 'ready', visibility: 'public', is_public: true, auto_generated: true, duration_seconds: 30, created_at: '2026-09-24 21:00:00' });++ items.clip.set(3108, { id: 3108, app_id: 'live', vod_id: null, stream_id: s2, channel_user_id: 1, user_id: 2, title: '', status: 'ready', visibility: 'public', is_public: true, auto_generated: false, duration_seconds: 12, created_at: '2026-09-25 01:00:00' });++ assert.strictEqual(await docs.touch('clip', 3107), 'sent');++ assert.strictEqual(await docs.touch('clip', 3108), 'sent');++ await outbox.flush();++ const [ai, nsfw] = index().slice(1, 3);++ valid(ai); valid(nsfw);++ assert.strictEqual(ai.payload.authorship, 'ai_generated');++ assert.deepStrictEqual(ai.payload.indexability, { decision: 'noindex', reasons: ['ai_unreviewed'] });++ assert.strictEqual(ai.payload.facets.vod_id, '912'); assert.strictEqual(ai.payload.facets.ai_clip, true);++ assert.strictEqual(nsfw.payload.title, 'Clip'); assert.strictEqual(nsfw.payload.authorship, 'human');++ assert.deepStrictEqual(nsfw.payload.indexability, { decision: 'noindex', reasons: ['sensitive'] });++++ // Made private: one tombstone.++ items.vod.set(912, vod(912, { stream_id: s1, visibility: 'private', is_public: false }));++ assert.strictEqual(await docs.touch('vod', 912), 'tombstone');++ assert.strictEqual(await docs.touch('vod', 912), 'unchanged', 'a tombstone is sent once');++ await outbox.flush();++ ev = index()[3];++ assert.strictEqual(ev.event_type, 'live.index_document.deleted'); valid(ev);++ assert.deepStrictEqual(ev.payload, { type: 'vod', id: '912', revision: 2 });++++ // Deleted on Media: a 404 is a deletion. A failed read changes nothing.++ const realGet = media.getClip;++ media.getClip = async () => { throw Object.assign(new Error('Media timed out'), { status: 0 }); };++ assert.strictEqual(await docs.touch('clip', 3108), 'skipped', 'Media unreachable: nothing removed');++ media.getClip = realGet;++ items.clip.delete(3108);++ assert.strictEqual(await docs.touch('clip', 3108), 'tombstone');++++ // The daily refresh: a partial listing removes nothing; a complete one removes what Media stopped listing.++ items.clip.get(3107).visibility = 'unlisted';++ listFails = true;++ await docs.refresh();++ assert.strictEqual(d.prepare("SELECT deleted FROM search_media_pushes WHERE kind = 'clip' AND media_id = 3107").get().deleted, 0);++ listFails = false;++ await docs.refresh();++ assert.strictEqual(d.prepare("SELECT deleted FROM search_media_pushes WHERE kind = 'clip' AND media_id = 3107").get().deleted, 1, 'unlisted: removed');++++ // A successful change through /api/vods is re-checked; a failed one is not.++ items.vod.set(912, vod(912, { stream_id: s1 }));++ const call = (method, p, status, body) => {++ const res = new EventEmitter(); res.statusCode = status;++ docs.afterChange('vod')({ method, path: p, body }, res, () => {});++ res.emit('finish');++ };++ call('PUT', '/912', 403);++ await new Promise((r) => setTimeout(r, 60));++ assert.strictEqual(d.prepare("SELECT deleted FROM search_media_pushes WHERE kind = 'vod' AND media_id = 912").get().deleted, 1, 'a refused change is not followed');++ call('POST', '/bulk', 200, { ids: [912], action: 'public' });++ await new Promise((r) => setTimeout(r, 60));++ const row = d.prepare("SELECT deleted, revision FROM search_media_pushes WHERE kind = 'vod' AND media_id = 912").get();++ assert.deepStrictEqual(row, { deleted: 0, revision: 3 }, 'made public again: sent at the next revision');++++ outbox.stop && outbox.stop();++ stub.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ quiet('search media documents: all checks passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -25,7 +25,7 @@
@@ -25,7 +25,7 @@
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
"onnxruntime-node": "^1.20.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.45.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.9.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.9.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.12.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.12.0",
"sanitize-html": "^2.17.7",
"sanitize-html": "^2.17.7",
@@ -2814,9 +2814,9 @@
@@ -2814,9 +2814,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.44.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.0",
- "integrity": "sha512-VMjnJSwU2dnk9KstCHQDKfY6UysSk2fWK1wdLp15Via49/hbeaXRTa4XIjO3d82fYizj0rJpi9GrK+AJRskjtQ==",
+ "version": "0.45.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.45.0",
+ "integrity": "sha512-6JJL9gHKd+gzowNzUls7l5nV0awR/KmEXxaxUaxzUPZilVFlDSM6yFpXMjFp+pDrUXDHYI5kfMevhpI0GEBFEA==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -43,7 +43,7 @@
@@ -43,7 +43,7 @@
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
"onnxruntime-node": "^1.20.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.45.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.9.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.9.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.12.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.12.0",
"sanitize-html": "^2.17.7",
"sanitize-html": "^2.17.7",
new file mode 100644
new file mode 100644
@@ -0,0 +1,237 @@
@@ -0,0 +1,237 @@
+'use strict';
+/**
+ * Live's VOD and clip pages in OpenVibe.Search (roadmap WS-O task 10; Contracts 0.45.0
+ * live.index_document.* types vod and clip): Live owns the canonical /vod/<id> and /clip/<id> pages of
+ * the Media items recorded from its streams (title, channel, AI overview, transcript), so Live sends
+ * their Search documents through its transactional outbox (./stream-events.js), with a tombstone once
+ * an item is deleted, made private or unlisted, or failed. Items come from Media (media-client), never
+ * from the frozen local vods/clips tables; the AI overview and transcript are Live's vod_ai_state and
+ * clip_ai_state.
+ *
+ * touch(kind, id) right after Live changes an item (the /api/vods and /api/clips routes) or Media
+ * announces it ready or failed (server/media-proxy/outcomes.js)
+ * scan() every 5 minutes: the newest public VODs and clips
+ * refresh() daily: every public VOD and clip, and a tombstone for each item Search holds that
+ * Media no longer lists as public (checked one by one before it is removed)
+ *
+ * A document is sent only when it changed (search_media_pushes keeps a hash and the revision, which
+ * grows by one with every document or tombstone). AI clips and clips or VODs of NSFW streams are
+ * indexed noindex, as the pages are. Off while the outbox is off (EVENTS_URL unset).
+ */
+const crypto = require('crypto');
+const db = require('../db/database');
+const streamEvents = require('./stream-events');
+
+const PUBLIC_BASE = (process.env.LIVE_PUBLIC_ORIGIN || 'https://openvibe.live').replace(/\/+$/, '');
+const DAY_MS = 86400000;
+const PAGE = 200;
+const TOUCH_DELAY_MS = Number(process.env.LIVE_SEARCH_TOUCH_DELAY_MS) || 1500;
+const KINDS = ['vod', 'clip'];
+const stats = { sent: 0, tombstones: 0, unchanged: 0, lastError: null };
+
+let ready = false;
+function ensureSchema() {
+ if (ready) return;
+ db.getDb().exec(`CREATE TABLE IF NOT EXISTS search_media_pushes (
+ kind TEXT NOT NULL,
+ media_id INTEGER NOT NULL,
+ hash TEXT NOT NULL,
+ revision INTEGER NOT NULL,
+ deleted INTEGER NOT NULL DEFAULT 0,
+ pushed_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+ PRIMARY KEY (kind, media_id)
+ )`);
+ ready = true;
+}
+
+const iso = (v) => {
+ if (!v) return null;
+ const d = new Date(String(v).includes('T') ? v : `${String(v).replace(' ', 'T')}Z`);
+ return Number.isNaN(d.getTime()) ? null : d.toISOString();
+};
+const clean = (s, n) => String(s || '').replace(/\s+/g, ' ').trim().slice(0, n);
+const isAiClip = (c) => c.auto_generated === true || Number(c.auto_generated) === 1;
+
+function aiState(kind, id) {
+ const st = kind === 'clip' ? db.getClipAiState(id) : db.getVodAiState(id);
+ if (!st) return {};
+ let transcript = '';
+ try { transcript = (JSON.parse(st.ai_transcript_json || '[]') || []).map((s) => s && s.text).filter(Boolean).join(' '); } catch { /* unreadable: none */ }
+ return { overview: st.ai_overview || st.ai_overview_short || '', short: st.ai_overview_short || '', transcript };
+}
+
+/** Would Live's page for this item be public? (the same rule as its SSR meta and the Media sitemap) */
+function isListable(kind, row) {
+ if (!row || row.id == null) return false;
+ const visibility = row.visibility || (row.is_public ? 'public' : 'private');
+ if (visibility !== 'public' || row.is_public === false || Number(row.is_public) === 0) return false;
+ if (kind === 'vod' && (row.is_recording || row.clips_only || row.status !== 'ready')) return false;
+ if (kind === 'clip' && (row.status || 'ready') !== 'ready') return false;
+ if (row.readiness && row.readiness.playable === false) return false;
+ return true;
+}
+
+/** The document for one Media item, or `{ deleted: true }` when Search must not hold it. */
+function documentFor(kind, row) {
+ if (!isListable(kind, row)) return { deleted: true };
+ const d = db.getDb();
+ const ownerId = kind === 'clip' ? (row.channel_user_id || row.user_id) : row.user_id;
+ const owner = ownerId != null ? d.prepare('SELECT id, username, display_name, is_banned FROM users WHERE id = ?').get(Number(ownerId)) : null;
+ if (owner && owner.is_banned) return { deleted: true };
+ const stream = row.stream_id != null ? d.prepare('SELECT category, is_nsfw FROM streams WHERE id = ?').get(Number(row.stream_id)) : null;
+ const ch = owner ? d.prepare('SELECT chat_language FROM channels WHERE user_id = ?').get(owner.id) : null;
+ const name = owner ? (owner.display_name || owner.username) : null;
+ const ai = aiState(kind, row.id);
+ const aiClip = kind === 'clip' && isAiClip(row);
+ const fallbackTitle = kind === 'vod' ? `${name ? `${name}'s ` : ''}stream VOD` : (aiClip ? 'AI clip' : 'Clip');
+ const summary = clean(ai.short || ai.overview || row.description, 4000)
+ || `${kind === 'vod' ? 'Recorded live stream' : aiClip ? 'AI clip of a live stream' : 'Clip of a live stream'}${name ? ` by ${name}` : ''} on OpenVibe.Live.`;
+ const facets = { duration_seconds: Math.round(Number(row.duration_seconds ?? row.duration) || 0) };
+ if (owner) facets.channel = owner.username;
+ if (stream && stream.category) facets.category = String(stream.category).slice(0, 200);
+ if (kind === 'clip') { facets.ai_clip = aiClip; if (row.vod_id != null) facets.vod_id = String(row.vod_id); }
+ const reasons = [...(aiClip ? ['ai_unreviewed'] : []), ...(stream && stream.is_nsfw ? ['sensitive'] : [])];
+ const doc = {
+ owner: 'live', type: kind, id: String(row.id), deleted: false, visibility: 'public',
+ canonical_url: `${PUBLIC_BASE}/${kind}/${row.id}`,
+ title: clean(row.title || fallbackTitle, 500),
+ summary,
+ body: [clean(row.description, 4000), clean(ai.overview, 8000), ai.transcript].filter(Boolean).join('\n').slice(0, 40000),
+ facets, authorship: aiClip ? 'ai_generated' : 'human', publication_state: 'published', published_at: iso(row.created_at),
+ indexability: reasons.length ? { decision: 'noindex', reasons } : { decision: 'index' },
+ };
+ if (!doc.body) delete doc.body;
+ if (ch && /^[a-z]{2,3}$/.test(String(ch.chat_language || ''))) doc.language = ch.chat_language;
+ return doc;
+}
+
+const hashOf = (doc) => crypto.createHash('sha256').update(JSON.stringify(doc)).digest('hex').slice(0, 32);
+
+/** Send one item's document or tombstone when it changed. `row` null: gone. → 'sent' | 'tombstone' | 'unchanged' | 'skipped' */
+function publish(kind, id, row, { now = Date.now() } = {}) {
+ ensureSchema();
+ if (!streamEvents.status().enabled || !KINDS.includes(kind)) return 'skipped';
+ const mediaId = Number(id);
+ if (!Number.isSafeInteger(mediaId) || mediaId < 1) return 'skipped';
+ const doc = row ? documentFor(kind, row) : { deleted: true };
+ const d = db.getDb();
+ const prev = d.prepare('SELECT hash, revision, deleted FROM search_media_pushes WHERE kind = ? AND media_id = ?').get(kind, mediaId);
+ if (doc.deleted && (!prev || prev.deleted)) { stats.unchanged++; return 'unchanged'; } // never sent, or already gone
+ const hash = doc.deleted ? 'deleted' : hashOf(doc);
+ if (prev && prev.hash === hash) { stats.unchanged++; return 'unchanged'; }
+ const revision = (prev ? prev.revision : 0) + 1;
+ const sid = String(mediaId);
+ d.transaction(() => {
+ streamEvents.enqueue(doc.deleted
+ ? { event_type: 'live.index_document.deleted', actor: { type: 'service', id: 'live' }, subject: { type: kind, id: sid, revision }, visibility: 'internal', priority: 'low', payload: { type: kind, id: sid, revision } }
+ : { event_type: 'live.index_document.upserted', actor: { type: 'service', id: 'live' }, subject: { type: kind, id: sid, revision }, visibility: 'internal', priority: 'low',
+ payload: { ...doc, revision, updated_at: new Date(now).toISOString() } });
+ d.prepare(`INSERT INTO search_media_pushes (kind, media_id, hash, revision, deleted, pushed_at) VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
+ ON CONFLICT(kind, media_id) DO UPDATE SET hash = excluded.hash, revision = excluded.revision, deleted = excluded.deleted, pushed_at = excluded.pushed_at`)
+ .run(kind, mediaId, hash, revision, doc.deleted ? 1 : 0);
+ })();
+ streamEvents.kick();
+ if (doc.deleted) { stats.tombstones++; return 'tombstone'; }
+ stats.sent++;
+ return 'sent';
+}
+
+let mediaImpl = null;
+const mediaClient = () => mediaImpl || require('../media-client');
+const fetchOne = (kind, id, media) => (kind === 'vod' ? media.getVod(id, { timeoutMs: 10000 }) : media.getClip(id, { timeoutMs: 10000 }));
+
+/** Re-read one item from Media and send what changed. A 404 is a deletion; any other failure changes nothing. */
+async function touch(kind, id, { media = mediaClient(), now = Date.now() } = {}) {
+ if (!streamEvents.status().enabled || !KINDS.includes(kind)) return 'skipped';
+ let row;
+ try {
+ row = await fetchOne(kind, id, media);
+ } catch (err) {
+ if (err && err.status === 404) row = null;
+ else { stats.lastError = `${kind} ${id}: ${err && err.message}`; return 'skipped'; }
+ }
+ return publish(kind, id, row && (row.vod || row.clip || row), { now });
+}
+
+/** Fire-and-forget touch for route and event hooks: never throws, never delays the caller. */
+function touchLater(kind, ids) {
+ if (!streamEvents.status().enabled || process.env.LIVE_SEARCH_DOCUMENTS === 'off') return;
+ const list = [...new Set((Array.isArray(ids) ? ids : [ids]).map((v) => parseInt(v, 10)).filter((v) => v > 0))].slice(0, 500);
+ if (!list.length) return;
+ setTimeout(async () => {
+ for (const id of list) { try { await touch(kind, id); } catch (err) { stats.lastError = err.message; } }
+ }, TOUCH_DELAY_MS).unref();
+}
+
+async function listPage(kind, offset, media, limit = PAGE) {
+ const r = kind === 'vod' ? await media.listVods({ limit, offset }, { timeoutMs: 20000 }) : await media.listClips({ limit, offset }, { timeoutMs: 20000 });
+ return { rows: (r && (kind === 'vod' ? r.vods : r.clips)) || [], hasMore: !!(r && r.hasMore) };
+}
+
+/** The newest public items (Media lists newest first). */
+async function scan({ media = mediaClient(), now = Date.now(), limit = 50 } = {}) {
+ if (!streamEvents.status().enabled) return 0;
+ let n = 0;
+ for (const kind of KINDS) {
+ try {
+ const { rows } = await listPage(kind, 0, media, limit);
+ for (const row of rows) { publish(kind, row.id, row, { now }); n++; }
+ } catch (err) { stats.lastError = `${kind} scan: ${err.message}`; }
+ }
+ return n;
+}
+
+/** Every public item, then a check of each one Search holds that the listing no longer shows. */
+async function refresh({ media = mediaClient(), now = Date.now(), maxPages = 100 } = {}) {
+ if (!streamEvents.status().enabled) return 0;
+ ensureSchema();
+ let n = 0;
+ for (const kind of KINDS) {
+ const seen = new Set();
+ let complete = false;
+ try {
+ for (let page = 0, offset = 0; page < maxPages; page++, offset += PAGE) {
+ const { rows, hasMore } = await listPage(kind, offset, media);
+ for (const row of rows) { seen.add(Number(row.id)); publish(kind, row.id, row, { now }); n++; }
+ if (!hasMore || !rows.length) { complete = true; break; }
+ }
+ } catch (err) { stats.lastError = `${kind} refresh: ${err.message}`; }
+ if (!complete) continue; // a partial listing never removes anything
+ const held = db.getDb().prepare('SELECT media_id FROM search_media_pushes WHERE kind = ? AND deleted = 0').all(kind).map((r) => r.media_id);
+ for (const id of held) if (!seen.has(id)) await touch(kind, id, { media, now });
+ }
+ return n;
+}
+
+function init() {
+ if (!streamEvents.status().enabled || process.env.LIVE_SEARCH_DOCUMENTS === 'off') return false;
+ ensureSchema();
+ const jobs = require('../utils/jobs');
+ jobs.every('search-media-scan', 5 * 60 * 1000, () => scan(), { initialDelayMs: 3 * 60 * 1000, jitterMs: 15 * 1000 });
+ jobs.every('search-media-refresh', DAY_MS, () => refresh(), { initialDelayMs: 6 * 60 * 1000, jitterMs: 60 * 1000 });
+ return true;
+}
+
+/**
+ * Express middleware for /api/vods and /api/clips: after a successful change (not a GET), re-check the
+ * items it named (the numeric :id in the path, or `ids` in a bulk body).
+ */
+function afterChange(kind) {
+ return (req, res, next) => {
+ if (req.method === 'GET' || req.method === 'HEAD') return next();
+ res.on('finish', () => {
+ if (res.statusCode < 200 || res.statusCode >= 300) return;
+ const m = String(req.path || '').match(/^\/(?:clips\/)?(\d+)(?:\/|$)/);
+ const ids = m ? [m[1]] : (req.body && Array.isArray(req.body.ids) ? req.body.ids : []);
+ const k = kind === 'vod' && /^\/clips\//.test(String(req.path || '')) ? 'clip' : kind;
+ touchLater(k, ids);
+ });
+ next();
+ };
+}
+
+function status() { return { ...stats }; }
+
+function _setMedia(m) { mediaImpl = m; }
+
+module.exports = { init, ensureSchema, isListable, documentFor, publish, touch, touchLater, scan, refresh, afterChange, status, _setMedia };
@@ -676,8 +676,10 @@ app.use('/api/funds', monetizationRoutes);
@@ -676,8 +676,10 @@ app.use('/api/funds', monetizationRoutes);
app.use('/api/coins', coinsRoutes);
app.use('/api/coins', coinsRoutes);
app.use('/api/payments', require('./monetization/payments-routes'));
app.use('/api/payments', require('./monetization/payments-routes'));
app.use('/api/cosmetics', cosmeticsRoutes);
app.use('/api/cosmetics', cosmeticsRoutes);
-app.use('/api/vods', vodRoutes);
-app.use('/api/clips', clipRoutes);
+// Live's VOD and clip pages in OpenVibe.Search follow every change made here (events/search-media-documents.js).
+const searchMedia = require('./events/search-media-documents');
+app.use('/api/vods', searchMedia.afterChange('vod'), vodRoutes);
+app.use('/api/clips', searchMedia.afterChange('clip'), clipRoutes);
app.use('/api/chat-ai', require('./ai/chat-ai-routes'));
app.use('/api/chat-ai', require('./ai/chat-ai-routes'));
app.use('/api/easter-egg', require('./ai/easter-egg-routes'));
app.use('/api/easter-egg', require('./ai/easter-egg-routes'));
app.use('/api/arena', require('./arena/routes')); // streamer vs streamer (docs/arena.md)
app.use('/api/arena', require('./arena/routes')); // streamer vs streamer (docs/arena.md)
@@ -1558,6 +1560,7 @@ async function start() {
@@ -1558,6 +1560,7 @@ async function start() {
try { require('./events/stream-events').init(); } catch (err) { console.warn('[Events] not started:', err.message); }
try { require('./events/stream-events').init(); } catch (err) { console.warn('[Events] not started:', err.message); }
// Channels in OpenVibe.Search (WS-O task 10): live.index_document.* through the same outbox.
// Channels in OpenVibe.Search (WS-O task 10): live.index_document.* through the same outbox.
try { require('./events/search-documents').init(); } catch (err) { console.warn('[Search] channel documents not started:', err.message); }
try { require('./events/search-documents').init(); } catch (err) { console.warn('[Search] channel documents not started:', err.message); }
+ try { require('./events/search-media-documents').init(); } catch (err) { console.warn('[Search] VOD and clip documents not started:', err.message); }
// 8d. User modules on Network (Contracts 0.41.0): live.profile and live.stats, written when they
// 8d. User modules on Network (Contracts 0.41.0): live.profile and live.stats, written when they
// change (server/auth/module-summaries.js). Off without OV_OAUTH_CLIENT_SECRET.
// change (server/auth/module-summaries.js). Off without OV_OAUTH_CLIENT_SECRET.
@@ -206,6 +206,9 @@ function handle({ via, event, data, eventId = null, subject = null }) {
@@ -206,6 +206,9 @@ function handle({ via, event, data, eventId = null, subject = null }) {
}
}
stats.applied[via]++;
stats.applied[via]++;
if (after) { try { after(); } catch (e) { console.warn('[MediaOutcome] post-commit step failed:', e.message); } }
if (after) { try { after(); } catch (e) { console.warn('[MediaOutcome] post-commit step failed:', e.message); } }
+ // A VOD or clip that became ready (or failed) changes what OpenVibe.Search should hold for its page.
+ const kind = /^(vod|clip)\.(ready|failed)$/.exec(String(event));
+ if (kind && data && data.id != null) { try { require('../events/search-media-documents').touchLater(kind[1], data.id); } catch { /* search is optional */ } }
return { applied: true, duplicate: false, outcome: out.result };
return { applied: true, duplicate: false, outcome: out.result };
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,150 @@
@@ -0,0 +1,150 @@
+'use strict';
+// Live's VOD and clip pages in OpenVibe.Search (WS-O task 10, Contracts 0.45.0): a public, ready VOD goes
+// through the outbox as live.index_document.upserted type vod (valid against the contract, canonical on
+// Live, with Live's AI overview and transcript), again only when it changed; a recording VOD is never
+// sent; a VOD made private gets one tombstone; an AI clip and a clip of an NSFW stream are noindex; a 404
+// is a deletion; the daily refresh removes what Media stopped listing, but never after a partial listing;
+// a change through /api/vods is re-checked right after it succeeds.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const { EventEmitter } = require('events');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-live-searchmedia-'));
+process.env.DB_PATH = path.join(tmp, 'live.db');
+process.env.LIVE_SEARCH_TOUCH_DELAY_MS = '10';
+const quiet = console.log;
+console.log = (...a) => { if (!/^\[/.test(String(a[0]))) quiet(...a); };
+console.warn = () => {};
+
+const published = [];
+const stub = http.createServer((req, res) => {
+ let body = '';
+ req.on('data', (c) => { body += c; });
+ req.on('end', () => {
+ res.setHeader('Content-Type', 'application/json');
+ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));
+ if (req.url === '/api/v1/events' && req.method === 'POST') {
+ const parsed = JSON.parse(body);
+ const list = parsed.events || [parsed];
+ const { validate } = require('openvibe-contracts');
+ const bad = list.find((e) => !validate('events.event-envelope@1', e).valid);
+ if (bad) { res.statusCode = 422; return res.end(JSON.stringify({ code: 'events.invalid_envelope' })); }
+ const results = list.map((e) => { published.push(e); return { event_id: e.event_id, seq: published.length, duplicate: false }; });
+ return res.end(JSON.stringify(parsed.events ? { results } : results[0]));
+ }
+ res.statusCode = 404; res.end('{}');
+ });
+});
+
+// Media as Live's service sees it: listings hold public items only; get returns any of Live's.
+const items = { vod: new Map(), clip: new Map() };
+let listFails = false;
+const notFound = () => Object.assign(new Error('not found'), { status: 404 });
+const media = {
+ async listVods({ limit, offset }) { if (listFails) throw new Error('Media unreachable'); const all = [...items.vod.values()].filter((v) => v.visibility === 'public' && !v.is_recording); return { vods: all.slice(offset, offset + limit), hasMore: offset + limit < all.length }; },
+ async listClips({ limit, offset }) { if (listFails) throw new Error('Media unreachable'); const all = [...items.clip.values()].filter((c) => c.visibility === 'public'); return { clips: all.slice(offset, offset + limit), hasMore: offset + limit < all.length }; },
+ async getVod(id) { const v = items.vod.get(Number(id)); if (!v) throw notFound(); return v; },
+ async getClip(id) { const c = items.clip.get(Number(id)); if (!c) throw notFound(); return c; },
+};
+const vod = (id, extra = {}) => ({ id, app_id: 'live', user_id: 1, stream_id: 1, title: 'Building a forum', description: '', status: 'ready', visibility: 'public', is_public: true, is_recording: false, clips_only: false, duration_seconds: 5400, created_at: '2026-09-24 19:00:00', readiness: { playable: true }, ...extra });
+
+(async () => {
+ await new Promise((r) => stub.listen(0, '127.0.0.1', r));
+ const base = `http://127.0.0.1:${stub.address().port}`;
+ process.env.OV_NETWORK_INTERNAL_URL = base;
+ const { validate } = require('openvibe-contracts');
+ const db = require('../server/db/database');
+ db.initDb();
+ const d = db.getDb();
+ d.prepare("INSERT INTO users (id, username, display_name, password_hash) VALUES (1, 'alex', 'Alex', 'x'), (2, 'viewer', 'Viewer', 'x')").run();
+ const s1 = db.createStream({ user_id: 1, title: 'Building a forum', category: 'tech', protocol: 'webrtc', is_nsfw: 0 }).lastInsertRowid;
+ const s2 = db.createStream({ user_id: 1, title: 'Late night', category: 'irl', protocol: 'webrtc', is_nsfw: 1 }).lastInsertRowid;
+ const streamEvents = require('../server/events/stream-events');
+ const outbox = streamEvents.init({ eventsUrl: base, clientSecret: 's3cret', intervalMs: 50 });
+ const docs = require('../server/events/search-media-documents');
+ docs._setMedia(media);
+ const index = () => published.filter((e) => /^live\.index_document\./.test(e.event_type));
+ const valid = (ev) => { const v = validate(ev.event_type + '@1', ev.payload); assert.ok(v.valid, `${ev.event_type}: ${JSON.stringify(v.errors)}`); };
+
+ // A public, ready VOD with Live's AI overview and transcript.
+ items.vod.set(912, vod(912, { stream_id: s1 }));
+ db.setVodAiOverview(912, 'Alex builds the rating menu of the forum and fixes it on phones.');
+ d.prepare('UPDATE vod_ai_state SET ai_transcript_json = ? WHERE vod_id = 912').run(JSON.stringify([{ text: 'okay so today' }, { text: 'we are building the forum' }]));
+ items.vod.set(913, vod(913, { is_recording: true, status: 'recording' }));
+ assert.strictEqual(await docs.scan(), 1);
+ assert.strictEqual(await docs.touch('vod', 913), 'unchanged', 'a recording is never sent');
+ await outbox.flush();
+ let ev = index()[0];
+ assert.strictEqual(ev.event_type, 'live.index_document.upserted'); valid(ev);
+ assert.deepStrictEqual(ev.subject, { type: 'vod', id: '912', revision: 1 });
+ assert.strictEqual(ev.payload.canonical_url, 'https://openvibe.live/vod/912');
+ assert.ok(ev.payload.summary.startsWith('Alex builds the rating menu'));
+ assert.ok(ev.payload.body.includes('we are building the forum'), 'the transcript is searchable');
+ assert.deepStrictEqual(ev.payload.facets, { duration_seconds: 5400, channel: 'alex', category: 'tech' });
+ assert.strictEqual(ev.payload.indexability.decision, 'index');
+ assert.strictEqual(docs.publish('vod', 912, items.vod.get(912)), 'unchanged', 'nothing changed: nothing sent');
+
+ // An AI clip, and a person's clip of an NSFW stream.
+ items.clip.set(3107, { id: 3107, app_id: 'live', vod_id: 912, stream_id: s1, channel_user_id: 1, user_id: null, title: 'Chat erupts', status: 'ready', visibility: 'public', is_public: true, auto_generated: true, duration_seconds: 30, created_at: '2026-09-24 21:00:00' });
+ items.clip.set(3108, { id: 3108, app_id: 'live', vod_id: null, stream_id: s2, channel_user_id: 1, user_id: 2, title: '', status: 'ready', visibility: 'public', is_public: true, auto_generated: false, duration_seconds: 12, created_at: '2026-09-25 01:00:00' });
+ assert.strictEqual(await docs.touch('clip', 3107), 'sent');
+ assert.strictEqual(await docs.touch('clip', 3108), 'sent');
+ await outbox.flush();
+ const [ai, nsfw] = index().slice(1, 3);
+ valid(ai); valid(nsfw);
+ assert.strictEqual(ai.payload.authorship, 'ai_generated');
+ assert.deepStrictEqual(ai.payload.indexability, { decision: 'noindex', reasons: ['ai_unreviewed'] });
+ assert.strictEqual(ai.payload.facets.vod_id, '912'); assert.strictEqual(ai.payload.facets.ai_clip, true);
+ assert.strictEqual(nsfw.payload.title, 'Clip'); assert.strictEqual(nsfw.payload.authorship, 'human');
+ assert.deepStrictEqual(nsfw.payload.indexability, { decision: 'noindex', reasons: ['sensitive'] });
+
+ // Made private: one tombstone.
+ items.vod.set(912, vod(912, { stream_id: s1, visibility: 'private', is_public: false }));
+ assert.strictEqual(await docs.touch('vod', 912), 'tombstone');
+ assert.strictEqual(await docs.touch('vod', 912), 'unchanged', 'a tombstone is sent once');
+ await outbox.flush();
+ ev = index()[3];
+ assert.strictEqual(ev.event_type, 'live.index_document.deleted'); valid(ev);
+ assert.deepStrictEqual(ev.payload, { type: 'vod', id: '912', revision: 2 });
+
+ // Deleted on Media: a 404 is a deletion. A failed read changes nothing.
+ const realGet = media.getClip;
+ media.getClip = async () => { throw Object.assign(new Error('Media timed out'), { status: 0 }); };
+ assert.strictEqual(await docs.touch('clip', 3108), 'skipped', 'Media unreachable: nothing removed');
+ media.getClip = realGet;
+ items.clip.delete(3108);
+ assert.strictEqual(await docs.touch('clip', 3108), 'tombstone');
+
+ // The daily refresh: a partial listing removes nothing; a complete one removes what Media stopped listing.
+ items.clip.get(3107).visibility = 'unlisted';
+ listFails = true;
+ await docs.refresh();
+ assert.strictEqual(d.prepare("SELECT deleted FROM search_media_pushes WHERE kind = 'clip' AND media_id = 3107").get().deleted, 0);
+ listFails = false;
+ await docs.refresh();
+ assert.strictEqual(d.prepare("SELECT deleted FROM search_media_pushes WHERE kind = 'clip' AND media_id = 3107").get().deleted, 1, 'unlisted: removed');
+
+ // A successful change through /api/vods is re-checked; a failed one is not.
+ items.vod.set(912, vod(912, { stream_id: s1 }));
+ const call = (method, p, status, body) => {
+ const res = new EventEmitter(); res.statusCode = status;
+ docs.afterChange('vod')({ method, path: p, body }, res, () => {});
+ res.emit('finish');
+ };
+ call('PUT', '/912', 403);
+ await new Promise((r) => setTimeout(r, 60));
+ assert.strictEqual(d.prepare("SELECT deleted FROM search_media_pushes WHERE kind = 'vod' AND media_id = 912").get().deleted, 1, 'a refused change is not followed');
+ call('POST', '/bulk', 200, { ids: [912], action: 'public' });
+ await new Promise((r) => setTimeout(r, 60));
+ const row = d.prepare("SELECT deleted, revision FROM search_media_pushes WHERE kind = 'vod' AND media_id = 912").get();
+ assert.deepStrictEqual(row, { deleted: 0, revision: 3 }, 'made public again: sent at the next revision');
+
+ outbox.stop && outbox.stop();
+ stub.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ quiet('search media documents: all checks passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });