openvibers/openvibe.mediahub Commits

Video: public uploads are in OpenVibe.Search (owner media-hub, type video) (#77)

OP OpenVibers committed 29 days ago · b94ed9496e · 8 files, +273 −2

Every public, ready, active upload is a search.index-document@1 written through Search's owner API with
MediaHub's service token (search.document.write; Events cannot carry a hyphenated event type), the
title, description, AI overview and transcript as its text, /u/<id> as its address. A minute after boot,
hourly and 30 s after an edit, delete or report; re-sent one revision up only when its hash changed
(video_search_state, migration 0013); private, unlisted, hidden or removed uploads are tombstoned.
Off without MEDIAHUB_SEARCH_URL.

Changes

ddiff --git a/README.md b/README.md
iindex 73b57ab..b60fa14 100644
---- a/README.md
++++ b/README.md
@@@ -84,6 +84,10 @@ On other services it calls:
- OpenVibe.Network `identity.subject.resolve` — to turn the usernames a share names into canonical account subjects.
- OpenVibe.Events `events.subscription.manage` — to create its account export/deletion subscriptions at boot.
++- OpenVibe.Search `search.document.write` — OpenVibe.Video's public, ready uploads as Search documents (owner
++ `media-hub`, type `video`, `/u/<id>`) through the owner API ([server/video/search-index.js](server/video/search-index.js),
++ `MEDIAHUB_SEARCH_URL`; a minute after boot, hourly, and 30 s after an edit; a private, hidden or removed upload is
++ tombstoned; `video_search_state`, migration 0013).
## Why v1 is narrow
ddiff --git a/migrations/0013_video_search.sql b/migrations/0013_video_search.sql
nnew file mode 100644
iindex 0000000..55a017b
---- /dev/null
++++ b/migrations/0013_video_search.sql
@@@ -0,0 +1,9 @@
++-- phase: expand
++-- OpenVibe.Video uploads in OpenVibe.Search (server/video/search-index.js): what was last sent for each upload, so a
++-- sweep re-sends a document only when its content changed. Search's own listing gives the stored revision.
++CREATE TABLE video_search_state (
++ id text COLLATE "C" PRIMARY KEY, -- the upload's vup_ id (the Search document id)
++ hash text COLLATE "C" NOT NULL, -- sha256 of the document last sent
++ revision integer NOT NULL,
++ updated_at text COLLATE "C" NOT NULL
++);
ddiff --git a/server/app.js b/server/app.js
iindex 695afce..4ce22c9 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -77,7 +77,8 @@ async function createApp(opts = {}) {
const videoAi = opts.videoAi || require('./video/ai-edit').createAiEdit({ s, media, config, fetchImpl, log });
const syncKeys = createSyncKeys({ s, config });
const drops = createDrops({ s, media, config, uploads, log });
-- const ctx = { config, s, store, keys, sso, principal, media, identity, drive, uploads, account, playlists, pics, videoUploads, videoAi, syncKeys, drops, fetchImpl, log };
++ const videoSearch = require('./video/search-index').createVideoSearchIndex({ s, config, videoAi, fetchImpl, log });
++ const ctx = { config, s, store, keys, sso, principal, media, identity, drive, uploads, account, playlists, pics, videoUploads, videoAi, syncKeys, drops, videoSearch, fetchImpl, log };
// IndexNow for openvibe.video: off without a key; a malformed key warns and stays off rather than stopping the site.
ctx.indexnow = opts.indexnow || (() => {
const warn = (...a) => (log.warn || console.warn)(...a);
ddiff --git a/server/config.js b/server/config.js
iindex 58c9860..f0027b4 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -147,6 +147,8 @@ function load(env = process.env) {
// OpenVibe.Network: SSO (OAuth2 authorization server with PKCE) and its JWKS.
networkUrl,
networkInternalUrl: trim(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000'),
++ // OpenVibe.Search's owner API (server/video/search-index.js): public uploads become Search documents. Unset: off.
++ search: { url: trim(env.MEDIAHUB_SEARCH_URL || ''), videoOff: env.MEDIAHUB_VIDEO_SEARCH === 'off' },
networkIssuer: trim(env.OV_NETWORK_ISSUER || networkUrl),
// The audience this service's app, agent and service tokens carry.
audience: env.MEDIAHUB_AUDIENCE || 'openvibe.download',
ddiff --git a/server/index.js b/server/index.js
iindex 8a8b757..e9ac61e 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -63,7 +63,9 @@ async function start() {
aiTimer.unref();
timers.push(aiTimer);
}
-- createLifecycle({ server, ctx, timers, extra: [() => videoIndexNow.stop(), () => { if (subscriptions) subscriptions.stop(); }] });
++ // Public uploads in OpenVibe.Search (server/video/search-index.js; off without MEDIAHUB_SEARCH_URL).
++ ctx.videoSearch.start();
++ createLifecycle({ server, ctx, timers, extra: [() => videoIndexNow.stop(), () => ctx.videoSearch.stop(), () => { if (subscriptions) subscriptions.stop(); }] });
return { server, ctx };
}
ddiff --git a/server/video/search-index.js b/server/video/search-index.js
nnew file mode 100644
iindex 0000000..575d9c5
---- /dev/null
++++ b/server/video/search-index.js
@@@ -0,0 +1,148 @@
++'use strict';
++/**
++ * OpenVibe.Video uploads in OpenVibe.Search: every public, ready upload is one document (search.index-document@1,
++ * owner media-hub, type video, id = the vup_ id, /u/<id>), written through Search's owner API with MediaHub's service
++ * token (search.document.write on openvibe.search; Events cannot carry it, an event type has no hyphen). An upload
++ * that is made private or unlisted, hidden after reports, removed or deleted is tombstoned.
++ *
++ * Runs a minute after boot and every hour, and 30 seconds after a change (kick()). Search's listing gives the stored
++ * revisions; a document is re-sent one revision up only when its content hash differs from the last one sent
++ * (video_search_state, migration 0013), so a quiet hour costs one listing.
++ *
++ * Off unless MEDIAHUB_SEARCH_URL and the OAuth client secret are set (MEDIAHUB_VIDEO_SEARCH=off turns it off).
++ */
++const crypto = require('crypto');
++const { serviceAuth } = require('openvibe-contracts');
++
++const OWNER = 'media-hub';
++const TYPE = 'video';
++const EVERY_MS = 60 * 60 * 1000;
++const KICK_MS = 30 * 1000;
++
++const hashOf = (doc) => crypto.createHash('sha256').update(JSON.stringify(doc)).digest('hex');
++
++/** An upload row (and its AI edit, when done) → the document body; identity and revision are added by the caller. */
++function documentFor(row, { videoUrl, ai = null }) {
++ const channel = row.owner_username || null;
++ const overview = ai && ai.status === 'done' && ai.overview ? String(ai.overview) : '';
++ const transcript = ai && ai.status === 'done' && Array.isArray(ai.segments) ? ai.segments.map((x) => x.text).join(' ') : '';
++ return {
++ visibility: 'public',
++ canonical_url: `${videoUrl}/u/${row.id}`,
++ title: String(row.title || 'Video').slice(0, 500),
++ summary: String(row.description || overview || (channel ? `A video by @${channel} on OpenVibe.Video.` : 'A video on OpenVibe.Video.')).slice(0, 4000),
++ body: [row.title, row.description, overview, transcript.slice(0, 20000)].filter(Boolean).join('\n').slice(0, 48000),
++ facets: { kind: 'upload', ...(channel ? { channel } : {}), ...(row.duration_seconds ? { minutes: Math.max(1, Math.round(Number(row.duration_seconds) / 60)) } : {}) },
++ authorship: 'human',
++ publication_state: 'published',
++ published_at: row.ready_at || row.created_at || null,
++ updated_at: row.updated_at || null,
++ indexability: { decision: 'index', reasons: [] },
++ };
++}
++
++function createVideoSearchIndex({ s, config, videoAi = null, fetchImpl = globalThis.fetch, log = console, tokens = null }) {
++ const url = String((config.search && config.search.url) || '').replace(/\/+$/, '');
++ const enabled = Boolean(url && (tokens || config.oauth.clientSecret) && !(config.search && config.search.videoOff));
++ const auth = tokens || (enabled ? serviceAuth.createTokenClient({
++ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
++ audience: 'openvibe.search', scope: 'search.document.write', fetchImpl,
++ }) : null);
++ const videoUrl = String(config.video.url).replace(/\/+$/, '');
++ const stats = { runs: 0, upserted: 0, deleted: 0, unchanged: 0, failed: 0, last_run_at: null, last_error: null };
++
++ async function call(method, p, body) {
++ const res = await fetchImpl(`${url}${p}`, {
++ method,
++ headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await auth.authHeaders()) },
++ body: body ? JSON.stringify(body) : undefined,
++ signal: AbortSignal.timeout(15000),
++ });
++ const json = await res.json().catch(() => ({}));
++ if (res.status === 401 && auth.invalidate) auth.invalidate();
++ return { status: res.status, ok: res.ok, body: json };
++ }
++
++ async function remoteDocs() {
++ const out = new Map();
++ let after = 0;
++ for (let i = 0; i < 100; i++) {
++ const r = await call('GET', `/api/v1/owners/${OWNER}/documents?type=${TYPE}&limit=1000&after=${after}`);
++ if (!r.ok) throw new Error(`listing: ${r.status} ${r.body.code || ''}`);
++ for (const d of r.body.documents || []) out.set(d.id, d);
++ if (!r.body.next_after) break;
++ after = r.body.next_after;
++ }
++ return out;
++ }
++
++ async function run() {
++ if (!enabled) return null;
++ stats.runs++;
++ stats.last_run_at = new Date(s.now()).toISOString();
++ const listed = await s.db.many("SELECT * FROM video_uploads WHERE visibility = 'public' AND status = 'ready' AND state = 'active' ORDER BY id");
++ const sent = new Map((await s.db.many('SELECT id, hash FROM video_search_state')).map((r) => [r.id, r.hash]));
++ const remote = await remoteDocs();
++ const keep = new Set();
++ for (const row of listed) {
++ keep.add(row.id);
++ const ai = videoAi && videoAi.forUpload ? await videoAi.forUpload(row).catch(() => null) : null;
++ const doc = documentFor(row, { videoUrl, ai });
++ const h = hashOf(doc);
++ const r = remote.get(row.id);
++ if (r && !r.deleted && sent.get(row.id) === h) { stats.unchanged++; continue; }
++ const revision = r ? Number(r.revision) + 1 : 1;
++ const res = await call('PUT', `/api/v1/documents/${OWNER}/${TYPE}/${row.id}`, { ...doc, revision });
++ if (res.ok) {
++ await s.db.exec(`INSERT INTO video_search_state (id, hash, revision, updated_at) VALUES ($1, $2, $3, $4)
++ ON CONFLICT (id) DO UPDATE SET hash = excluded.hash, revision = excluded.revision, updated_at = excluded.updated_at`,
++ [row.id, h, revision, s.iso()]);
++ stats.upserted++;
++ } else {
++ stats.failed++;
++ stats.last_error = `${row.id}: ${res.status} ${res.body.code || ''}`;
++ }
++ }
++ for (const [id, r] of remote) {
++ if (keep.has(id) || r.deleted) continue;
++ const res = await call('DELETE', `/api/v1/documents/${OWNER}/${TYPE}/${encodeURIComponent(id)}?revision=${Number(r.revision) + 1}`);
++ if (res.ok) {
++ await s.db.exec('DELETE FROM video_search_state WHERE id = $1', [id]);
++ stats.deleted++;
++ } else {
++ stats.failed++;
++ stats.last_error = `${id}: ${res.status} ${res.body.code || ''}`;
++ }
++ }
++ return { ...stats };
++ }
++
++ const quietly = () => run().catch((err) => { stats.failed++; stats.last_error = err.message; log.warn('[OpenVibe.Video] search index run failed:', err.message); });
++ let timer = null;
++ let first = null;
++ let kicked = null;
++ return {
++ enabled,
++ run,
++ stats: () => ({ ...stats }),
++ /** A change happened (an upload edited, deleted or reported): run soon, once for a burst of changes. */
++ kick() {
++ if (!enabled || kicked) return;
++ kicked = setTimeout(() => { kicked = null; quietly(); }, KICK_MS);
++ if (kicked.unref) kicked.unref();
++ },
++ start({ firstDelayMs = 60_000 } = {}) {
++ if (!enabled || timer) return;
++ first = setTimeout(quietly, firstDelayMs);
++ if (first.unref) first.unref();
++ timer = setInterval(quietly, EVERY_MS);
++ if (timer.unref) timer.unref();
++ },
++ stop() {
++ clearTimeout(first); clearInterval(timer); clearTimeout(kicked);
++ timer = null; first = null; kicked = null;
++ },
++ };
++}
++
++module.exports = { createVideoSearchIndex, documentFor };
ddiff --git a/server/video/upload-routes.js b/server/video/upload-routes.js
iindex 0009fb6..5ce6e72 100644
---- a/server/video/upload-routes.js
++++ b/server/video/upload-routes.js
@@@ -277,6 +277,7 @@ ${!own && ready ? html`<details class="ovv-panel ovv-report"><summary>Report thi
if (!(err instanceof UploadError)) throw err;
return page(req, res, { title: 'Not saved', body: html`<h1>Not saved</h1><p>${err.detail}</p><p><a href="/u/${row.id}">Back</a></p>` }, err.status);
}
++ if (ctx.videoSearch) ctx.videoSearch.kick();
return res.redirect(303, `/u/${row.id}?saved=1`);
});
@@@ -287,6 +288,7 @@ ${!own && ready ? html`<details class="ovv-panel ovv-report"><summary>Report thi
if (!sameSite(req, res, `/u/${row.id}`)) return undefined;
if ((req.body || {}).confirm !== '1') return res.redirect(303, `/u/${row.id}`);
await up.remove(row);
++ if (ctx.videoSearch) ctx.videoSearch.kick();
return res.redirect(303, '/me/videos');
});
@@@ -301,6 +303,7 @@ ${!own && ready ? html`<details class="ovv-panel ovv-report"><summary>Report thi
if (!(err instanceof UploadError)) throw err;
return page(req, res, { title: 'Not sent', body: html`<h1>Not sent</h1><p>${err.detail}</p><p><a href="/u/${row.id}">Back</a></p>` }, err.status);
}
++ if (ctx.videoSearch) ctx.videoSearch.kick();
return res.redirect(303, `/u/${row.id}?reported=1`);
});
ddiff --git a/test/video-search-index.test.js b/test/video-search-index.test.js
nnew file mode 100644
iindex 0000000..68f2d5c
---- /dev/null
++++ b/test/video-search-index.test.js
@@@ -0,0 +1,102 @@
++'use strict';
++/**
++ * OpenVibe.Video uploads in OpenVibe.Search (server/video/search-index.js), against a stand-in Search owner API:
++ * - a public, ready, active upload is a valid search.index-document@1 (owner media-hub, type video, /u/<id>);
++ * - private, unlisted, unfinished or hidden uploads are not sent; an unchanged one is not sent again;
++ * - an edit is re-sent one revision up; an upload made private (or removed) is tombstoned.
++ */
++const assert = require('assert');
++const http = require('http');
++const { validate } = require('openvibe-contracts');
++const { boot, check, done } = require('./helpers/boot');
++const { createVideoSearchIndex } = require('../server/video/search-index');
++
++function startSearch() {
++ const docs = new Map();
++ const calls = [];
++ const server = http.createServer((req, res) => {
++ let raw = '';
++ req.on('data', (c) => { raw += c; });
++ req.on('end', () => {
++ const u = new URL(req.url, 'http://search.test');
++ calls.push(`${req.method} ${u.pathname}`);
++ const send = (status, body) => { res.writeHead(status, { 'content-type': 'application/json' }); res.end(JSON.stringify(body)); };
++ if (req.headers.authorization !== 'Bearer svc-search') return send(401, { code: 'token.invalid' });
++ if (req.method === 'GET' && u.pathname === '/api/v1/owners/media-hub/documents') {
++ return send(200, { documents: [...docs.values()].map((d) => ({ type: d.type, id: d.id, revision: d.revision, deleted: Boolean(d.deleted) })), next_after: null });
++ }
++ const m = /^\/api\/v1\/documents\/media-hub\/video\/([^/]+)$/.exec(u.pathname);
++ if (m && req.method === 'PUT') {
++ const doc = { ...JSON.parse(raw), owner: 'media-hub', type: 'video', id: decodeURIComponent(m[1]) };
++ const v = validate('search.index-document@1', doc);
++ if (!v.valid) return send(422, { code: 'search.bad_document', errors: v.errors });
++ docs.set(doc.id, doc);
++ return send(200, { outcome: 'applied', revision: doc.revision });
++ }
++ if (m && req.method === 'DELETE') {
++ const id = decodeURIComponent(m[1]);
++ docs.set(id, { type: 'video', id, revision: Number(u.searchParams.get('revision')), deleted: true });
++ return send(200, { outcome: 'applied' });
++ }
++ return send(404, { code: 'route.not_found' });
++ });
++ });
++ return new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve({ docs, calls, url: `http://127.0.0.1:${server.address().port}`, close: () => new Promise((r) => server.close(r)) })));
++}
++
++(async () => {
++ const search = await startSearch();
++ const t = await boot({ env: {} });
++ const s = t.ctx.s;
++ const insert = (id, o = {}) => s.db.exec(
++ `INSERT INTO video_uploads (id, owner, owner_username, title, description, visibility, content_type, size_bytes, part_size, parts_expected, status, state, created_at, updated_at, ready_at, duration_seconds)
++ VALUES ($1, 'user:usr_01J0000000000000000000000K', 'kim', $2, $3, $4, 'video/mp4', 100, 100, 1, $5, $6, '2026-10-15T10:00:00.000Z', '2026-10-15T10:00:00.000Z', '2026-10-15T10:05:00.000Z', 125)`,
++ [id, o.title || 'Trip to the coast', o.description || 'Waves and wind.', o.visibility || 'public', o.status || 'ready', o.state || 'active']);
++ const ids = { pub: 'vup_01J0000000000000000000000A', priv: 'vup_01J0000000000000000000000B', proc: 'vup_01J0000000000000000000000C', hidden: 'vup_01J0000000000000000000000D' };
++ await insert(ids.pub);
++ await insert(ids.priv, { visibility: 'private' });
++ await insert(ids.proc, { status: 'processing' });
++ await insert(ids.hidden, { state: 'hidden' });
++ const index = createVideoSearchIndex({
++ s, config: { ...t.config, search: { url: search.url } }, videoAi: null,
++ tokens: { authHeaders: async () => ({ authorization: 'Bearer svc-search' }) },
++ log: { warn() {} },
++ });
++
++ await check('only the public, ready, active upload is sent, as a valid document', async () => {
++ assert.ok(index.enabled);
++ const r = await index.run();
++ assert.strictEqual(r.upserted, 1, JSON.stringify(r));
++ assert.deepStrictEqual([...search.docs.keys()], [ids.pub]);
++ const doc = search.docs.get(ids.pub);
++ assert.strictEqual(doc.canonical_url, `https://openvibe.video/u/${ids.pub}`);
++ assert.strictEqual(doc.title, 'Trip to the coast');
++ assert.deepStrictEqual(doc.facets, { kind: 'upload', channel: 'kim', minutes: 2 });
++ assert.strictEqual(doc.revision, 1);
++ });
++
++ await check('unchanged is not sent again; an edit goes one revision up; made private, it is tombstoned', async () => {
++ const before = search.calls.filter((c) => c.startsWith('PUT')).length;
++ await index.run();
++ assert.strictEqual(search.calls.filter((c) => c.startsWith('PUT')).length, before, 'nothing changed, nothing sent');
++ await s.db.exec("UPDATE video_uploads SET title = 'Coast trip' WHERE id = $1", [ids.pub]);
++ await index.run();
++ assert.strictEqual(search.docs.get(ids.pub).revision, 2);
++ assert.strictEqual(search.docs.get(ids.pub).title, 'Coast trip');
++ await s.db.exec("UPDATE video_uploads SET visibility = 'private' WHERE id = $1", [ids.pub]);
++ const r = await index.run();
++ assert.strictEqual(r.deleted, 1);
++ assert.deepStrictEqual(search.docs.get(ids.pub), { type: 'video', id: ids.pub, revision: 3, deleted: true });
++ assert.strictEqual(Number(await s.db.value('SELECT count(*) FROM video_search_state')), 0);
++ });
++
++ await check('off without a Search URL', async () => {
++ const off = createVideoSearchIndex({ s, config: { ...t.config, search: { url: '' } } });
++ assert.strictEqual(off.enabled, false);
++ assert.strictEqual(await off.run(), null);
++ });
++
++ await t.close();
++ await search.close();
++ done();
++})();