openvibers/openvibe.live Commits
Channels in OpenVibe.Search (WS-O task 10, Contracts 0.44.0): each streamer's channel is a Search document sent through the outbox as live.index_document.upserted (name, bio, recent stream titles, category, live, followers; NSFW channels noindex) only when it changed, with a tombstone once a channel is banned; a 5-minute scan of what changed and a daily refresh
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.43.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.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.43.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.43.0",-- "integrity": "sha512-zgyfoqWUzSr8lCEqY6q+XVliBldOUWSCacliBsBXRz12fK2d6ELrmHOUbZA6evobT/MnNpxs6rDvcmaVfb+nDQ==",++ "version": "0.44.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.0",++ "integrity": "sha512-VMjnJSwU2dnk9KstCHQDKfY6UysSk2fWK1wdLp15Via49/hbeaXRTa4XIjO3d82fYizj0rJpi9GrK+AJRskjtQ==","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.43.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.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,135 @@++'use strict';++/**++ * Live's channels in OpenVibe.Search (roadmap WS-O task 10; Contracts 0.44.0 live.index_document.*):++ * one document per streamer (a user with at least one stream), published through Live's transactional++ * outbox (./stream-events.js) as live.index_document.upserted, and a tombstone (.deleted) once a channel++ * is banned or gone. Search takes them through its '*.index_document.*' subscription.++ *++ * scan() every 5 minutes: channels whose stream started or ended, who gained a follower or whose++ * account changed since the previous scan++ * refresh() daily: every channel (follower counts and the last stream move)++ *++ * A document is sent only when what Search should hold changed (search_doc_pushes keeps a hash and the++ * revision, which grows by one with every document or tombstone). NSFW channels are indexed `noindex`++ * (Search can only make that stricter). 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 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_doc_pushes (++ user_id INTEGER PRIMARY KEY,++ hash TEXT NOT NULL,++ revision INTEGER NOT NULL,++ deleted INTEGER NOT NULL DEFAULT 0,++ pushed_at DATETIME DEFAULT CURRENT_TIMESTAMP++ )`);++ 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();++};++++/** The document for one account's channel, `{ deleted: true }` when it must leave Search, or null (never a channel). */++function documentFor(userId) {++ const d = db.getDb();++ const u = d.prepare('SELECT id, username, display_name, bio, is_banned FROM users WHERE id = ?').get(userId);++ if (!u) return { deleted: true };++ const streams = d.prepare('SELECT COUNT(*) AS n, MIN(started_at) AS first FROM streams WHERE user_id = ?').get(userId);++ if (!streams.n) return null;++ if (u.is_banned) return { deleted: true };++ const ch = d.prepare('SELECT description, category, chat_language FROM channels WHERE user_id = ?').get(userId) || {};++ const last = d.prepare('SELECT title, category, is_live, is_nsfw FROM streams WHERE user_id = ? ORDER BY id DESC LIMIT 1').get(userId) || {};++ const titles = d.prepare("SELECT DISTINCT title FROM streams WHERE user_id = ? AND title IS NOT NULL AND title != '' ORDER BY id DESC LIMIT 10").all(userId).map((r) => r.title);++ const followers = d.prepare('SELECT COUNT(*) AS n FROM follows WHERE streamer_id = ?').get(userId).n;++ const name = u.display_name || u.username;++ const about = String(ch.description || u.bio || '').trim();++ const lang = /^[a-z]{2,3}$/.test(String(ch.chat_language || '')) ? ch.chat_language : null;++ const doc = {++ owner: 'live', type: 'channel', id: String(u.id), deleted: false, visibility: 'public',++ canonical_url: `${PUBLIC_BASE}/@${encodeURIComponent(u.username)}`,++ title: `${name} (@${u.username})`.slice(0, 500),++ summary: (about || `${name} streams on OpenVibe.Live.`).slice(0, 4000),++ body: [about, ...titles].filter(Boolean).join('\n').slice(0, 8000),++ facets: { category: String(last.category || ch.category || 'irl').slice(0, 200), live: !!last.is_live, followers },++ authorship: 'human', publication_state: 'published', published_at: iso(streams.first),++ indexability: last.is_nsfw ? { decision: 'noindex', reasons: ['sensitive'] } : { decision: 'index' },++ };++ if (lang) doc.language = lang;++ return doc;++}++++const hashOf = (doc) => crypto.createHash('sha256').update(JSON.stringify(doc)).digest('hex').slice(0, 32);++++/** Send one channel's document or tombstone when it changed. → 'sent' | 'tombstone' | 'unchanged' | 'skipped' */++function publish(userId, { now = Date.now() } = {}) {++ ensureSchema();++ if (!streamEvents.status().enabled) return 'skipped';++ const doc = documentFor(userId);++ const d = db.getDb();++ const prev = d.prepare('SELECT hash, revision, deleted FROM search_doc_pushes WHERE user_id = ?').get(userId);++ if (!doc) return 'skipped';++ 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 id = String(userId);++ d.transaction(() => {++ streamEvents.enqueue(doc.deleted++ ? { event_type: 'live.index_document.deleted', actor: { type: 'service', id: 'live' }, subject: { type: 'channel', id, revision }, visibility: 'internal', priority: 'normal', payload: { type: 'channel', id, revision } }++ : { event_type: 'live.index_document.upserted', actor: { type: 'service', id: 'live' }, subject: { type: 'channel', id, revision }, visibility: 'internal', priority: 'normal',++ payload: { ...doc, revision, updated_at: new Date(now).toISOString() } });++ d.prepare(`INSERT INTO search_doc_pushes (user_id, hash, revision, deleted, pushed_at) VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)++ ON CONFLICT(user_id) DO UPDATE SET hash = excluded.hash, revision = excluded.revision, deleted = excluded.deleted, pushed_at = excluded.pushed_at`)++ .run(userId, hash, revision, doc.deleted ? 1 : 0);++ })();++ streamEvents.kick();++ if (doc.deleted) { stats.tombstones++; return 'tombstone'; }++ stats.sent++;++ return 'sent';++}++++let lastScan = null;++const sqlTime = (ms) => new Date(ms).toISOString().replace('T', ' ').slice(0, 19);++function scan({ now = Date.now() } = {}) {++ if (!streamEvents.status().enabled) return 0;++ const from = lastScan || sqlTime(now - 10 * 60 * 1000);++ const to = sqlTime(now);++ const ids = db.getDb().prepare(`SELECT user_id AS id FROM streams WHERE (started_at >= @from AND started_at < @to) OR (ended_at >= @from AND ended_at < @to)++ UNION SELECT streamer_id FROM follows WHERE created_at >= @from AND created_at < @to++ UNION SELECT id FROM users WHERE updated_at >= @from AND updated_at < @to++ UNION SELECT user_id FROM channels WHERE updated_at >= @from AND updated_at < @to`).all({ from, to }).map((r) => r.id);++ for (const id of ids) { try { publish(id, { now }); } catch (err) { stats.lastError = err.message; } }++ lastScan = to;++ return ids.length;++}++++function refresh({ now = Date.now() } = {}) {++ if (!streamEvents.status().enabled) return 0;++ const ids = db.getDb().prepare('SELECT DISTINCT user_id AS id FROM streams UNION SELECT user_id FROM search_doc_pushes').all().map((r) => r.id);++ for (const id of ids) { try { publish(id, { now }); } catch (err) { stats.lastError = err.message; } }++ return ids.length;++}++++function init() {++ if (!streamEvents.status().enabled || process.env.LIVE_SEARCH_DOCUMENTS === 'off') return false;++ ensureSchema();++ const jobs = require('../utils/jobs');++ jobs.every('search-documents-scan', 5 * 60 * 1000, () => scan(), { initialDelayMs: 2 * 60 * 1000, jitterMs: 15 * 1000 });++ jobs.every('search-documents-refresh', DAY_MS, () => refresh(), { initialDelayMs: 4 * 60 * 1000, jitterMs: 60 * 1000 });++ return true;++}++++function status() { return { ...stats }; }++++module.exports = { init, ensureSchema, documentFor, publish, scan, refresh, status };@@@ -1556,6 +1556,8 @@ async function start() {// 8c. Durable events (roadmap Wave 3): stream lifecycle goes to OpenVibe.Events through the// transactional outbox (server/events/stream-events.js). Off unless EVENTS_URL is set.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); }// 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.nnew file mode 100644@@@ -0,0 +1,89 @@++'use strict';++// Live's channels in OpenVibe.Search (WS-O task 10, Contracts 0.44.0): a streamer's channel document goes++// through the outbox as live.index_document.upserted (valid against the contract), again only when it++// changed with the revision up by one; an NSFW channel is noindex; a banned channel gets a tombstone once;++// someone who never streamed is not a channel; the scan finds a stream that just ended.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-live-searchdocs-'));++process.env.DB_PATH = path.join(tmp, 'live.db');++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 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('{}');++ });++});++++(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, bio) VALUES (1, 'alex', 'Alex', 'x', 'Builds things live.'), (2, 'viewer', 'Viewer', 'x', '')").run();++ const streamEvents = require('../server/events/stream-events');++ const outbox = streamEvents.init({ eventsUrl: base, clientSecret: 's3cret', intervalMs: 50 });++ const docs = require('../server/events/search-documents');++ const index = () => published.filter((e) => /^live\.index_document\./.test(e.event_type));++++ const s = db.createStream({ user_id: 1, title: 'Building a forum', category: 'tech', protocol: 'webrtc', is_nsfw: 0 });++ db.endStream(s.lastInsertRowid);++ assert.strictEqual(docs.publish(2), 'skipped', 'someone who never streamed is not a channel');++ assert.strictEqual(docs.publish(1), 'sent');++ await outbox.flush();++ let ev = index()[0];++ assert.strictEqual(ev.event_type, 'live.index_document.upserted'); assert.strictEqual(ev.source, 'live');++ assert.deepStrictEqual(ev.subject, { type: 'channel', id: '1', revision: 1 });++ const v = validate('live.index_document.upserted@1', ev.payload);++ assert.ok(v.valid, JSON.stringify(v.errors));++ assert.strictEqual(ev.payload.canonical_url, 'https://openvibe.live/@alex');++ assert.strictEqual(ev.payload.summary, 'Builds things live.');++ assert.ok(ev.payload.body.includes('Building a forum'));++ assert.strictEqual(docs.publish(1), 'unchanged', 'nothing changed: nothing sent');++++ d.prepare("INSERT INTO follows (follower_id, streamer_id) VALUES (2, 1)").run();++ assert.strictEqual(docs.publish(1), 'sent');++ await outbox.flush();++ assert.strictEqual(index()[1].payload.revision, 2); assert.strictEqual(index()[1].payload.facets.followers, 1);++++ const nsfw = db.createStream({ user_id: 1, title: 'Late night', protocol: 'webrtc', is_nsfw: 1 });++ db.endStream(nsfw.lastInsertRowid);++ assert.ok(docs.scan({ now: Date.now() + 1000 }) >= 1, 'the scan finds the stream that ended');++ await outbox.flush();++ assert.strictEqual(index()[2].payload.indexability.decision, 'noindex');++++ d.prepare('UPDATE users SET is_banned = 1 WHERE id = 1').run();++ assert.strictEqual(docs.publish(1), 'tombstone');++ await outbox.flush();++ ev = index()[3];++ assert.strictEqual(ev.event_type, 'live.index_document.deleted');++ assert.ok(validate('live.index_document.deleted@1', ev.payload).valid);++ assert.deepStrictEqual(ev.payload, { type: 'channel', id: '1', revision: 4 });++ assert.strictEqual(docs.publish(1), 'unchanged', 'a tombstone is sent once');++++ outbox.stop && outbox.stop();++ stub.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ quiet('search 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.43.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.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.43.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.43.0",
- "integrity": "sha512-zgyfoqWUzSr8lCEqY6q+XVliBldOUWSCacliBsBXRz12fK2d6ELrmHOUbZA6evobT/MnNpxs6rDvcmaVfb+nDQ==",
+ "version": "0.44.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.0",
+ "integrity": "sha512-VMjnJSwU2dnk9KstCHQDKfY6UysSk2fWK1wdLp15Via49/hbeaXRTa4XIjO3d82fYizj0rJpi9GrK+AJRskjtQ==",
"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.43.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.44.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,135 @@
@@ -0,0 +1,135 @@
+'use strict';
+/**
+ * Live's channels in OpenVibe.Search (roadmap WS-O task 10; Contracts 0.44.0 live.index_document.*):
+ * one document per streamer (a user with at least one stream), published through Live's transactional
+ * outbox (./stream-events.js) as live.index_document.upserted, and a tombstone (.deleted) once a channel
+ * is banned or gone. Search takes them through its '*.index_document.*' subscription.
+ *
+ * scan() every 5 minutes: channels whose stream started or ended, who gained a follower or whose
+ * account changed since the previous scan
+ * refresh() daily: every channel (follower counts and the last stream move)
+ *
+ * A document is sent only when what Search should hold changed (search_doc_pushes keeps a hash and the
+ * revision, which grows by one with every document or tombstone). NSFW channels are indexed `noindex`
+ * (Search can only make that stricter). 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 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_doc_pushes (
+ user_id INTEGER PRIMARY KEY,
+ hash TEXT NOT NULL,
+ revision INTEGER NOT NULL,
+ deleted INTEGER NOT NULL DEFAULT 0,
+ pushed_at DATETIME DEFAULT CURRENT_TIMESTAMP
+ )`);
+ 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();
+};
+
+/** The document for one account's channel, `{ deleted: true }` when it must leave Search, or null (never a channel). */
+function documentFor(userId) {
+ const d = db.getDb();
+ const u = d.prepare('SELECT id, username, display_name, bio, is_banned FROM users WHERE id = ?').get(userId);
+ if (!u) return { deleted: true };
+ const streams = d.prepare('SELECT COUNT(*) AS n, MIN(started_at) AS first FROM streams WHERE user_id = ?').get(userId);
+ if (!streams.n) return null;
+ if (u.is_banned) return { deleted: true };
+ const ch = d.prepare('SELECT description, category, chat_language FROM channels WHERE user_id = ?').get(userId) || {};
+ const last = d.prepare('SELECT title, category, is_live, is_nsfw FROM streams WHERE user_id = ? ORDER BY id DESC LIMIT 1').get(userId) || {};
+ const titles = d.prepare("SELECT DISTINCT title FROM streams WHERE user_id = ? AND title IS NOT NULL AND title != '' ORDER BY id DESC LIMIT 10").all(userId).map((r) => r.title);
+ const followers = d.prepare('SELECT COUNT(*) AS n FROM follows WHERE streamer_id = ?').get(userId).n;
+ const name = u.display_name || u.username;
+ const about = String(ch.description || u.bio || '').trim();
+ const lang = /^[a-z]{2,3}$/.test(String(ch.chat_language || '')) ? ch.chat_language : null;
+ const doc = {
+ owner: 'live', type: 'channel', id: String(u.id), deleted: false, visibility: 'public',
+ canonical_url: `${PUBLIC_BASE}/@${encodeURIComponent(u.username)}`,
+ title: `${name} (@${u.username})`.slice(0, 500),
+ summary: (about || `${name} streams on OpenVibe.Live.`).slice(0, 4000),
+ body: [about, ...titles].filter(Boolean).join('\n').slice(0, 8000),
+ facets: { category: String(last.category || ch.category || 'irl').slice(0, 200), live: !!last.is_live, followers },
+ authorship: 'human', publication_state: 'published', published_at: iso(streams.first),
+ indexability: last.is_nsfw ? { decision: 'noindex', reasons: ['sensitive'] } : { decision: 'index' },
+ };
+ if (lang) doc.language = lang;
+ return doc;
+}
+
+const hashOf = (doc) => crypto.createHash('sha256').update(JSON.stringify(doc)).digest('hex').slice(0, 32);
+
+/** Send one channel's document or tombstone when it changed. → 'sent' | 'tombstone' | 'unchanged' | 'skipped' */
+function publish(userId, { now = Date.now() } = {}) {
+ ensureSchema();
+ if (!streamEvents.status().enabled) return 'skipped';
+ const doc = documentFor(userId);
+ const d = db.getDb();
+ const prev = d.prepare('SELECT hash, revision, deleted FROM search_doc_pushes WHERE user_id = ?').get(userId);
+ if (!doc) return 'skipped';
+ 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 id = String(userId);
+ d.transaction(() => {
+ streamEvents.enqueue(doc.deleted
+ ? { event_type: 'live.index_document.deleted', actor: { type: 'service', id: 'live' }, subject: { type: 'channel', id, revision }, visibility: 'internal', priority: 'normal', payload: { type: 'channel', id, revision } }
+ : { event_type: 'live.index_document.upserted', actor: { type: 'service', id: 'live' }, subject: { type: 'channel', id, revision }, visibility: 'internal', priority: 'normal',
+ payload: { ...doc, revision, updated_at: new Date(now).toISOString() } });
+ d.prepare(`INSERT INTO search_doc_pushes (user_id, hash, revision, deleted, pushed_at) VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
+ ON CONFLICT(user_id) DO UPDATE SET hash = excluded.hash, revision = excluded.revision, deleted = excluded.deleted, pushed_at = excluded.pushed_at`)
+ .run(userId, hash, revision, doc.deleted ? 1 : 0);
+ })();
+ streamEvents.kick();
+ if (doc.deleted) { stats.tombstones++; return 'tombstone'; }
+ stats.sent++;
+ return 'sent';
+}
+
+let lastScan = null;
+const sqlTime = (ms) => new Date(ms).toISOString().replace('T', ' ').slice(0, 19);
+function scan({ now = Date.now() } = {}) {
+ if (!streamEvents.status().enabled) return 0;
+ const from = lastScan || sqlTime(now - 10 * 60 * 1000);
+ const to = sqlTime(now);
+ const ids = db.getDb().prepare(`SELECT user_id AS id FROM streams WHERE (started_at >= @from AND started_at < @to) OR (ended_at >= @from AND ended_at < @to)
+ UNION SELECT streamer_id FROM follows WHERE created_at >= @from AND created_at < @to
+ UNION SELECT id FROM users WHERE updated_at >= @from AND updated_at < @to
+ UNION SELECT user_id FROM channels WHERE updated_at >= @from AND updated_at < @to`).all({ from, to }).map((r) => r.id);
+ for (const id of ids) { try { publish(id, { now }); } catch (err) { stats.lastError = err.message; } }
+ lastScan = to;
+ return ids.length;
+}
+
+function refresh({ now = Date.now() } = {}) {
+ if (!streamEvents.status().enabled) return 0;
+ const ids = db.getDb().prepare('SELECT DISTINCT user_id AS id FROM streams UNION SELECT user_id FROM search_doc_pushes').all().map((r) => r.id);
+ for (const id of ids) { try { publish(id, { now }); } catch (err) { stats.lastError = err.message; } }
+ return ids.length;
+}
+
+function init() {
+ if (!streamEvents.status().enabled || process.env.LIVE_SEARCH_DOCUMENTS === 'off') return false;
+ ensureSchema();
+ const jobs = require('../utils/jobs');
+ jobs.every('search-documents-scan', 5 * 60 * 1000, () => scan(), { initialDelayMs: 2 * 60 * 1000, jitterMs: 15 * 1000 });
+ jobs.every('search-documents-refresh', DAY_MS, () => refresh(), { initialDelayMs: 4 * 60 * 1000, jitterMs: 60 * 1000 });
+ return true;
+}
+
+function status() { return { ...stats }; }
+
+module.exports = { init, ensureSchema, documentFor, publish, scan, refresh, status };
@@ -1556,6 +1556,8 @@ async function start() {
@@ -1556,6 +1556,8 @@ async function start() {
// 8c. Durable events (roadmap Wave 3): stream lifecycle goes to OpenVibe.Events through the
// 8c. Durable events (roadmap Wave 3): stream lifecycle goes to OpenVibe.Events through the
// transactional outbox (server/events/stream-events.js). Off unless EVENTS_URL is set.
// transactional outbox (server/events/stream-events.js). Off unless EVENTS_URL is set.
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.
+ try { require('./events/search-documents').init(); } catch (err) { console.warn('[Search] channel 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.
new file mode 100644
new file mode 100644
@@ -0,0 +1,89 @@
@@ -0,0 +1,89 @@
+'use strict';
+// Live's channels in OpenVibe.Search (WS-O task 10, Contracts 0.44.0): a streamer's channel document goes
+// through the outbox as live.index_document.upserted (valid against the contract), again only when it
+// changed with the revision up by one; an NSFW channel is noindex; a banned channel gets a tombstone once;
+// someone who never streamed is not a channel; the scan finds a stream that just ended.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-live-searchdocs-'));
+process.env.DB_PATH = path.join(tmp, 'live.db');
+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 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('{}');
+ });
+});
+
+(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, bio) VALUES (1, 'alex', 'Alex', 'x', 'Builds things live.'), (2, 'viewer', 'Viewer', 'x', '')").run();
+ const streamEvents = require('../server/events/stream-events');
+ const outbox = streamEvents.init({ eventsUrl: base, clientSecret: 's3cret', intervalMs: 50 });
+ const docs = require('../server/events/search-documents');
+ const index = () => published.filter((e) => /^live\.index_document\./.test(e.event_type));
+
+ const s = db.createStream({ user_id: 1, title: 'Building a forum', category: 'tech', protocol: 'webrtc', is_nsfw: 0 });
+ db.endStream(s.lastInsertRowid);
+ assert.strictEqual(docs.publish(2), 'skipped', 'someone who never streamed is not a channel');
+ assert.strictEqual(docs.publish(1), 'sent');
+ await outbox.flush();
+ let ev = index()[0];
+ assert.strictEqual(ev.event_type, 'live.index_document.upserted'); assert.strictEqual(ev.source, 'live');
+ assert.deepStrictEqual(ev.subject, { type: 'channel', id: '1', revision: 1 });
+ const v = validate('live.index_document.upserted@1', ev.payload);
+ assert.ok(v.valid, JSON.stringify(v.errors));
+ assert.strictEqual(ev.payload.canonical_url, 'https://openvibe.live/@alex');
+ assert.strictEqual(ev.payload.summary, 'Builds things live.');
+ assert.ok(ev.payload.body.includes('Building a forum'));
+ assert.strictEqual(docs.publish(1), 'unchanged', 'nothing changed: nothing sent');
+
+ d.prepare("INSERT INTO follows (follower_id, streamer_id) VALUES (2, 1)").run();
+ assert.strictEqual(docs.publish(1), 'sent');
+ await outbox.flush();
+ assert.strictEqual(index()[1].payload.revision, 2); assert.strictEqual(index()[1].payload.facets.followers, 1);
+
+ const nsfw = db.createStream({ user_id: 1, title: 'Late night', protocol: 'webrtc', is_nsfw: 1 });
+ db.endStream(nsfw.lastInsertRowid);
+ assert.ok(docs.scan({ now: Date.now() + 1000 }) >= 1, 'the scan finds the stream that ended');
+ await outbox.flush();
+ assert.strictEqual(index()[2].payload.indexability.decision, 'noindex');
+
+ d.prepare('UPDATE users SET is_banned = 1 WHERE id = 1').run();
+ assert.strictEqual(docs.publish(1), 'tombstone');
+ await outbox.flush();
+ ev = index()[3];
+ assert.strictEqual(ev.event_type, 'live.index_document.deleted');
+ assert.ok(validate('live.index_document.deleted@1', ev.payload).valid);
+ assert.deepStrictEqual(ev.payload, { type: 'channel', id: '1', revision: 4 });
+ assert.strictEqual(docs.publish(1), 'unchanged', 'a tombstone is sent once');
+
+ outbox.stop && outbox.stop();
+ stub.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ quiet('search documents: all checks passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });