openvibers/openvibe.media Commits

Media's own watch pages in OpenVibe.Search (WS-O task 10, Contracts 0.45.0): media.index_document.upserted for exactly the pages the sitemap lists (one WATCH_WHERE rule shared by both: public, finished, playable, made by a person, not Live's, no sandbox), a tombstone once one stops qualifying, sent only when it changed, every 10 minutes through the outbox; Live's VODs and clips stay Live's documents

OP OpenVibers committed 25 Sept 2026 · bc7b1a50fd · 7 files, +176 −13


Changes

ddiff --git a/package-lock.json b/package-lock.json
iindex 4dc81c9..68b8e60 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -16,7 +16,7 @@
"dotenv": "^16.4.5",
"express": "^4.19.2",
"multer": "^1.4.5-lts.1",
-- "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",
"sharp": "^0.33.4"
@@@ -1919,9 +1919,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",
ddiff --git a/package.json b/package.json
iindex 48cc091..9efbba8 100644
---- a/package.json
++++ b/package.json
@@@ -11,7 +11,7 @@
},
"license": "MIT",
"dependencies": {
-- "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",
"@aws-sdk/client-s3": "^3.600.0",
"@aws-sdk/lib-storage": "^3.600.0",
ddiff --git a/server/events.js b/server/events.js
iindex f432b3f..50efc59 100644
---- a/server/events.js
++++ b/server/events.js
@@@ -148,6 +148,17 @@ function enqueue(eventType, appId, subject, data, priority) {
return env;
}
++/**
++ * Queue a Search document or tombstone (media.index_document.*, ./public/search-documents.js) as is:
++ * no tenant field, low priority. Same rule as record(): inside the transaction that records the push.
++ */
++function recordIndexDocument(eventType, subject, payload) {
++ if (!outbox) return null;
++ const env = outbox.enqueue({ event_type: eventType, actor: { type: 'service', id: 'media' }, subject, visibility: 'internal', priority: 'low', payload });
++ stats.queued++;
++ return env;
++}
++
/**
* Operator scripts (scripts/media-jobs.js) change job state in the same database as the running
* service. When the service's outbox table exists (the outbox is on there), the script writes its
@@@ -240,4 +251,4 @@ function _reset() {
stats.queued = 0; stats.lastError = null;
}
--module.exports = { init, initWriter, record, recordJob, recordObjectChanges, drainObjectChanges, discardObjectChanges, objectChangeEvent, kick, status, TYPES, JOB_TRANSITIONS, _reset };
++module.exports = { init, initWriter, record, recordIndexDocument, recordJob, recordObjectChanges, drainObjectChanges, discardObjectChanges, objectChangeEvent, kick, status, TYPES, JOB_TRANSITIONS, _reset };
ddiff --git a/server/index.js b/server/index.js
iindex 4adf1a4..ef8dd62 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -276,6 +276,8 @@ const server = app.listen(config.port, config.host, () => {
console.log(`[Media] RTP ingest pool: udp ${config.rtp.portMin}-${config.rtp.portMax} (127.0.0.1)`);
// Durable events (roadmap Wave 3): webhook outcomes also go to OpenVibe.Events. Off without EVENTS_URL.
try { require('./events').init(); } catch (err) { console.warn('[Events] not started:', err.message); }
++ // Media's own watch pages in OpenVibe.Search (WS-O task 10): media.index_document.* through the same outbox.
++ try { require('./public/search-documents').start(); } catch (err) { console.warn('[Search] documents not started:', err.message); }
});
// A drill whose port is taken stops instead of running unready.
ddiff --git a/server/public/crawl.js b/server/public/crawl.js
iindex 54f5749..48cb04c 100644
---- a/server/public/crawl.js
++++ b/server/public/crawl.js
@@@ -28,6 +28,19 @@ const iso = (v) => {
};
const abs = (u) => (!u ? null : (/^https?:\/\//i.test(u) ? u : `${config.publicUrl}${u.startsWith('/') ? '' : '/'}${u}`));
++/**
++ * The watch pages Media is the canonical home of, as SQL: public, finished, playable, made by a person,
++ * owned by an app other than Live. The sitemap and the Search documents (./search-documents.js) both
++ * use it; pages.watchIndexable and the sandbox check still apply to each row.
++ */
++const WATCH_WHERE = {
++ vod: `COALESCE(is_recording, 0) = 0 AND COALESCE(clips_only, 0) = 0 AND is_public = 1 AND COALESCE(visibility, 'public') = 'public'
++ AND quarantined_at IS NULL AND COALESCE(health_status, 'ok') NOT IN ('corrupt', 'zero_byte', 'missing_file', 'needs_review')
++ AND file_path IS NOT NULL AND app_id != 'live' AND ${readiness.playableSql('vods.object_id')}`,
++ clip: `COALESCE(status, 'ready') = 'ready' AND COALESCE(is_public, 1) = 1 AND COALESCE(visibility, 'public') = 'public'
++ AND COALESCE(auto_generated, 0) = 0 AND COALESCE(file_path, '') != '' AND app_id != 'live' AND ${readiness.playableSql('clips.object_id')}`,
++};
++
function sandboxApps() {
return new Set(db.all("SELECT app_id FROM apps WHERE env = 'sandbox'").map(r => r.app_id));
}
@@@ -37,18 +50,13 @@ function sitemapEntries() {
const sandbox = sandboxApps();
const out = [];
const vods = db.all(`SELECT id, app_id, visibility, is_public, thumbnail_url, created_at FROM vods
-- WHERE COALESCE(is_recording, 0) = 0 AND COALESCE(clips_only, 0) = 0 AND is_public = 1 AND COALESCE(visibility, 'public') = 'public'
-- AND quarantined_at IS NULL AND COALESCE(health_status, 'ok') NOT IN ('corrupt', 'zero_byte', 'missing_file', 'needs_review')
-- AND file_path IS NOT NULL AND app_id != 'live' AND ${readiness.playableSql('vods.object_id')}
-- ORDER BY created_at DESC LIMIT ?`, [MAX_URLS]);
++ WHERE ${WATCH_WHERE.vod} ORDER BY created_at DESC LIMIT ?`, [MAX_URLS]);
for (const v of vods) {
if (sandbox.has(v.app_id) || !pages.watchIndexable('vod', v)) continue;
out.push({ loc: `${config.publicUrl}/v/${v.id}`, lastmod: iso(v.created_at), images: v.thumbnail_url ? [abs(v.thumbnail_url)] : [] });
}
const clips = db.all(`SELECT id, app_id, visibility, is_public, auto_generated, thumbnail_url, created_at FROM clips
-- WHERE COALESCE(status, 'ready') = 'ready' AND COALESCE(is_public, 1) = 1 AND COALESCE(visibility, 'public') = 'public'
-- AND COALESCE(auto_generated, 0) = 0 AND COALESCE(file_path, '') != '' AND app_id != 'live' AND ${readiness.playableSql('clips.object_id')}
-- ORDER BY created_at DESC LIMIT ?`, [MAX_URLS]);
++ WHERE ${WATCH_WHERE.clip} ORDER BY created_at DESC LIMIT ?`, [MAX_URLS]);
for (const c of clips) {
if (sandbox.has(c.app_id) || !pages.watchIndexable('clip', c)) continue;
out.push({ loc: `${config.publicUrl}/c/${c.id}`, lastmod: iso(c.created_at), images: c.thumbnail_url ? [abs(c.thumbnail_url)] : [] });
@@@ -112,5 +120,7 @@ router.get('/sitemap.xml', (req, res) => {
module.exports = router;
module.exports.sitemapEntries = sitemapEntries;
++module.exports.WATCH_WHERE = WATCH_WHERE;
++module.exports.sandboxApps = sandboxApps;
module.exports.robotsTxt = robotsTxt;
module.exports.llmsTxt = llmsTxt;
ddiff --git a/server/public/search-documents.js b/server/public/search-documents.js
nnew file mode 100644
iindex 0000000..470e0b9
---- /dev/null
++++ b/server/public/search-documents.js
@@@ -0,0 +1,118 @@
++'use strict';
++/**
++ * Media's own watch pages in OpenVibe.Search (roadmap WS-O task 10; Contracts 0.45.0
++ * media.index_document.upserted|deleted): one document per VOD or clip whose canonical page is here
++ * (/v/<id>, /c/<id>): exactly the pages the sitemap lists (./crawl.js WATCH_WHERE, pages.watchIndexable,
++ * no sandbox tenant). Live's VODs and clips are not among them: their canonical pages are Live's, and
++ * Live sends their documents (live.index_document type vod and clip).
++ *
++ * sync() every 10 minutes: every such page, sent when it changed, and a tombstone for each page sent
++ * before that no longer qualifies (deleted, private, unplayable, quarantined)
++ *
++ * search_doc_pushes keeps a hash and the revision, which grows by one with every document or tombstone;
++ * the event is queued in the transaction that records the push. Off while the outbox is off.
++ */
++const crypto = require('crypto');
++const config = require('../config');
++const db = require('../db/database');
++const events = require('../events');
++const crawl = require('./crawl');
++const pages = require('./pages');
++
++const SYNC_MS = 10 * 60 * 1000;
++const stats = { sent: 0, tombstones: 0, unchanged: 0, lastError: null, lastSyncAt: null };
++let timer = null;
++
++function ensureSchema() {
++ db.getDb().exec(`CREATE TABLE IF NOT EXISTS search_doc_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)
++ )`);
++}
++
++const iso = (v) => {
++ if (!v) return null;
++ const d = new Date(/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}/.test(String(v)) ? `${String(v).replace(' ', 'T')}Z` : v);
++ return Number.isNaN(d.getTime()) ? null : d.toISOString();
++};
++const clean = (s, n) => String(s || '').replace(/\s+/g, ' ').trim().slice(0, n);
++const hashOf = (doc) => crypto.createHash('sha256').update(JSON.stringify(doc)).digest('hex').slice(0, 32);
++
++function documentFor(kind, row) {
++ const doc = {
++ owner: 'media', type: kind, id: String(row.id), deleted: false, visibility: 'public',
++ canonical_url: `${config.publicUrl}/${kind === 'vod' ? 'v' : 'c'}/${row.id}`,
++ title: clean(row.title, 500) || (kind === 'vod' ? 'Video' : 'Clip'),
++ summary: clean(row.description, 4000) || `A ${kind === 'vod' ? 'video' : 'clip'} on OpenVibe.Media.`,
++ facets: { app: String(row.app_id).slice(0, 200), duration_seconds: Math.round(Number(row.duration_seconds) || 0) },
++ authorship: 'human', publication_state: 'published', published_at: iso(row.created_at),
++ indexability: { decision: 'index' },
++ };
++ const body = clean(row.description, 40000);
++ if (body) doc.body = body;
++ return doc;
++}
++
++/** Send one document or tombstone when it changed. → 'sent' | 'tombstone' | 'unchanged' */
++function send(kind, id, doc, now) {
++ const d = db.getDb();
++ const prev = d.prepare('SELECT hash, revision, deleted FROM search_doc_pushes WHERE kind = ? AND media_id = ?').get(kind, id);
++ if (!doc && (!prev || prev.deleted)) { stats.unchanged++; return 'unchanged'; }
++ const hash = doc ? hashOf(doc) : 'deleted';
++ if (prev && prev.hash === hash) { stats.unchanged++; return 'unchanged'; }
++ const revision = (prev ? prev.revision : 0) + 1;
++ const sid = String(id);
++ d.transaction(() => {
++ if (doc) events.recordIndexDocument('media.index_document.upserted', { type: kind, id: sid, revision }, { ...doc, revision, updated_at: new Date(now).toISOString() });
++ else events.recordIndexDocument('media.index_document.deleted', { type: kind, id: sid, revision }, { type: kind, id: sid, revision });
++ d.prepare(`INSERT INTO search_doc_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, id, hash, revision, doc ? 0 : 1);
++ })();
++ if (doc) { stats.sent++; return 'sent'; }
++ stats.tombstones++;
++ return 'tombstone';
++}
++
++function sync({ now = Date.now() } = {}) {
++ if (!events.status().enabled) return 0;
++ ensureSchema();
++ const sandbox = crawl.sandboxApps();
++ let n = 0;
++ for (const kind of ['vod', 'clip']) {
++ const table = kind === 'vod' ? 'vods' : 'clips';
++ const rows = db.all(`SELECT id, app_id, title, description, duration_seconds, visibility, is_public, ${kind === 'clip' ? 'auto_generated, ' : ''}created_at
++ FROM ${table} WHERE ${crawl.WATCH_WHERE[kind]}`).filter((r) => !sandbox.has(r.app_id) && pages.watchIndexable(kind, r));
++ const seen = new Set();
++ for (const r of rows) {
++ seen.add(r.id);
++ try { send(kind, r.id, documentFor(kind, r), now); n++; } catch (err) { stats.lastError = err.message; }
++ }
++ for (const p of db.all('SELECT media_id FROM search_doc_pushes WHERE kind = ? AND deleted = 0', [kind])) {
++ if (!seen.has(p.media_id)) { try { send(kind, p.media_id, null, now); } catch (err) { stats.lastError = err.message; } }
++ }
++ }
++ stats.lastSyncAt = new Date(now).toISOString();
++ return n;
++}
++
++function start() {
++ if (timer || !events.status().enabled || process.env.MEDIA_SEARCH_DOCUMENTS === 'off') return false;
++ ensureSchema();
++ const run = () => { try { sync(); } catch (err) { stats.lastError = err.message; } };
++ const first = setTimeout(run, 60 * 1000);
++ if (first.unref) first.unref();
++ timer = setInterval(run, SYNC_MS);
++ if (timer.unref) timer.unref();
++ return true;
++}
++
++function stop() { if (timer) clearInterval(timer); timer = null; }
++function status() { return { ...stats }; }
++
++module.exports = { start, stop, sync, documentFor, ensureSchema, status };
ddiff --git a/test/crawl.test.js b/test/crawl.test.js
iindex 24e818d..63c31d4 100644
---- a/test/crawl.test.js
++++ b/test/crawl.test.js
@@@ -116,6 +116,28 @@ const get = (p, headers = {}) => new Promise((resolve, reject) => {
assert.ok(r.body.includes('content="noindex, follow"'));
console.log('✅ watch pages: VideoObject when public; AI clips noindex, attributed to OpenVibe AI and their source VOD');
++ // ── Search documents (search-documents.js): exactly the sitemap's watch pages; a page that stops
++ // qualifying gets one tombstone; nothing is sent twice ──
++ const events = require('../server/events');
++ events.init({ eventsUrl: 'http://127.0.0.1:9', clientSecret: 's', intervalMs: 3_600_000, fetchImpl: async () => { throw new Error('offline'); } });
++ const search = require('../server/public/search-documents');
++ const queued = () => raw.prepare('SELECT envelope FROM event_outbox ORDER BY id').all().map((x) => JSON.parse(x.envelope)).filter((e) => /^media\.index_document\./.test(e.event_type));
++ const { validate } = require('openvibe-contracts');
++ search.sync();
++ const sent = queued();
++ assert.deepStrictEqual(sent.map((e) => e.payload.canonical_url).sort(), require('../server/public/crawl').sitemapEntries().map((e) => e.loc).sort(), 'the same pages as the sitemap');
++ for (const e of sent) { const v = validate(`${e.event_type}@1`, e.payload); assert.ok(v.valid, JSON.stringify(v.errors)); assert.strictEqual(e.priority, 'low'); assert.ok(!('app_id' in e.payload)); }
++ search.sync();
++ assert.strictEqual(queued().length, sent.length, 'unchanged: nothing sent again');
++ raw.prepare("UPDATE vods SET visibility = 'private', is_public = 0 WHERE id = 1").run();
++ search.sync();
++ const tomb = queued().at(-1);
++ assert.strictEqual(tomb.event_type, 'media.index_document.deleted');
++ assert.deepStrictEqual(tomb.payload, { type: 'vod', id: '1', revision: 2 });
++ assert.ok(validate('media.index_document.deleted@1', tomb.payload).valid);
++ events._reset && events._reset();
++ console.log('✅ search documents: the sitemap\'s pages, sent once, a tombstone when one goes private');
++
server.close();
db.close();
fs.rmSync(tmp, { recursive: true, force: true });