openvibers/openvibe.report Commits

Rundown topic pages in OpenVibe.Search (#69)

OP OpenVibers committed 11 days ago · f0bc772a2f · 10 files, +264 −10

Live's topic pages are noindex and Report's are the public copy, yet none of the 60 topics was on
network search. server/search-index.js sends one report.index_document per topic from Live's
topic list (title, headline, takes and streamers; a topic with no take is noindex), through a new
event_outbox (migration 0004, with report_index_revisions), a minute after boot and every hour;
a topic that left the Rundown gets a tombstone, and a Live that does not answer stops the sweep.
REPORT_EVENTS_URL is the Events address; off under REPORT_DRILL.

Changes

ddiff --git a/README.md b/README.md
iindex 2a1d224..b15a988 100644
---- a/README.md
++++ b/README.md
@@@ -205,7 +205,8 @@ payloads with money or personal data are not stored. `/internal/` is loopback-on
verifies the Events v2 HMAC. See [SECURITY.md](SECURITY.md).
<!-- versions:start -->
--- openvibe-contracts: v0.263.0
++- openvibe-contracts: v0.275.0
++- openvibe-publishing: v1.5.0
- openvibe-sdk: v0.81.0
- openvibe-shared: v4.103.0
<!-- versions:end -->
ddiff --git a/STATUS.json b/STATUS.json
iindex 217e61b..1ce67a4 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -8,11 +8,12 @@
"runtime": "node 22, express 4, PostgreSQL (PGlite in development), no sessions",
"port": "5080",
"domain": "openvibe.report",
-- "contracts": "openvibe-contracts v0.263.0",
++ "contracts": "openvibe-contracts v0.275.0",
"sdk": "openvibe-sdk v0.81.0",
"shared": "openvibe-shared v4.103.0",
"packages": [
-- "openvibe-contracts v0.263.0",
++ "openvibe-contracts v0.275.0",
++ "openvibe-publishing v1.5.0",
"openvibe-sdk v0.81.0",
"openvibe-shared v4.103.0",
"pg v8.23.1",
ddiff --git a/migrations/0004_search_index.sql b/migrations/0004_search_index.sql
nnew file mode 100644
iindex 0000000..db7ec0c
---- /dev/null
++++ b/migrations/0004_search_index.sql
@@@ -0,0 +1,31 @@
++-- phase: expand
++-- OpenVibe.Report: the Rundown topic pages in OpenVibe.Search (server/search-index.js, openvibe-publishing/search-feed).
++-- event_outbox holds the report.index_document.* events until the relay delivers them to OpenVibe.Events (written in
++-- the same transaction as the revision that produced them); report_index_revisions is the index sequencer, the last
++-- revision and content hash Search was sent for each topic, so an unchanged topic is never sent twice.
++
++CREATE TABLE IF NOT EXISTS event_outbox (
++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ event_id text NOT NULL UNIQUE,
++ envelope jsonb NOT NULL,
++ traceparent text,
++ created_at bigint NOT NULL,
++ attempts integer NOT NULL DEFAULT 0,
++ next_attempt_at bigint NOT NULL DEFAULT 0,
++ sent_at bigint,
++ seq bigint,
++ rejected_at bigint,
++ last_error text
++);
++CREATE INDEX IF NOT EXISTS event_outbox_due ON event_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
++CREATE INDEX IF NOT EXISTS event_outbox_sent ON event_outbox (sent_at) WHERE sent_at IS NOT NULL;
++
++CREATE TABLE IF NOT EXISTS report_index_revisions (
++ owner text COLLATE "C" NOT NULL,
++ type text COLLATE "C" NOT NULL,
++ id text COLLATE "C" NOT NULL,
++ revision integer NOT NULL,
++ hash text NOT NULL,
++ updated_at bigint NOT NULL,
++ PRIMARY KEY (owner, type, id)
++);
ddiff --git a/package-lock.json b/package-lock.json
iindex a71c6af..5a95875 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -13,7 +13,8 @@
"express": "^4.21.2",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.263.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.275.0",
++ "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.5.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.81.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.103.0",
"pg": "^8.23.1"
@@@ -1056,9 +1057,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.263.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.263.0",
-- "integrity": "sha512-mzVXLXIzgJc8VuhTZNDJak+iHm547nisRZZNfOp6V4arn8oePlM/S1BU1enaikboD/RkAgRjcITkIHLSyRq0VQ==",
++ "version": "0.275.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.275.0",
++ "integrity": "sha512-s8D3Upj5lfCGSnTVXKEWyzDEG5b7QnY/O2tAPieaZgFhzM4rnZyboGCZv3qthVOZ1ibxx/ZTEtS9nxZ/AVL+Hg==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
@@@ -1071,6 +1072,24 @@
"node": ">=20"
}
},
++ "node_modules/openvibe-publishing": {
++ "version": "1.5.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.5.0",
++ "integrity": "sha512-8cKVhJvpbDgWuGR+Dh1mpgqfWPJoC7EIcw7eTRc5/2ueDc8caCA7XKdcpAHNZYMdukIAL8o7FvxE8QV5/1uSLw==",
++ "license": "MIT",
++ "engines": {
++ "node": ">=20"
++ },
++ "peerDependencies": {
++ "openvibe-sdk": ">=0.15.0",
++ "openvibe-shared": ">=1.5.0"
++ },
++ "peerDependenciesMeta": {
++ "openvibe-shared": {
++ "optional": true
++ }
++ }
++ },
"node_modules/openvibe-sdk": {
"version": "0.81.0",
"resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.81.0",
ddiff --git a/package.json b/package.json
iindex dd0083c..19a3a71 100644
---- a/package.json
++++ b/package.json
@@@ -19,7 +19,8 @@
"express": "^4.21.2",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.263.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.275.0",
++ "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.5.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.81.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.103.0",
"pg": "^8.23.1"
ddiff --git a/server/app.js b/server/app.js
iindex 6c7ca2a..a975280 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -33,6 +33,7 @@ const { createBillingClient } = require('./data/billing');
const { createBenchClient } = require('./data/bench');
const { createAnalytics } = require('./analytics');
const { createAiUsage } = require('./analytics/ai-usage');
++const { createSearchIndex } = require('./search-index');
const { createShadow } = require('./analytics/shadow');
const { createApi } = require('./http/api');
const { createPageRoutes, createPublicPages } = require('./http/pages');
@@@ -63,6 +64,8 @@ async function createApp(opts = {}) {
// The AI ledger (plan L40/L41): the records of every AI call on the network and their roll-ups.
const aiUsage = opts.aiUsage || createAiUsage({ store, now, log });
const shadow = opts.shadow || (store ? createShadow({ store, network, now, log }) : null);
++ // The Rundown topic pages in OpenVibe.Search (started by server/index.js, never under REPORT_DRILL).
++ const searchIndex = opts.searchIndex || (store ? createSearchIndex({ config, store, live, log }) : null);
views.configure({ baseUrl: config.baseUrl });
// A person's sign-in (the `report` client, the same id and secret Report already reads Billing with).
@@@ -72,7 +75,7 @@ async function createApp(opts = {}) {
secureCookies: /^https:/.test(config.baseUrl), fetch: fetchImpl, log,
});
-- const ctx = { config, log, fetchImpl, now, store, network, live, billing, bench, analytics, shadow, sso, aiUsage };
++ const ctx = { config, log, fetchImpl, now, store, network, live, billing, bench, analytics, shadow, sso, aiUsage, searchIndex };
const app = express();
app.disable('x-powered-by');
app.set('trust proxy', config.trustProxy);
ddiff --git a/server/config.js b/server/config.js
iindex a47ac86..21a7941 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -82,6 +82,9 @@ function load(env = process.env) {
// separated for rotation, 32+ characters each); unset makes POST /internal/events answer 503.
events: {
secrets: String(env.REPORT_EVENTS_SECRET || '').split(',').map((x) => x.trim()).filter((x) => x.length >= 32),
++ // Report → OpenVibe.Events: the Rundown topic pages' Search documents (server/search-index.js). Unset: they
++ // wait in event_outbox.
++ url: trim(env.REPORT_EVENTS_URL || env.EVENTS_URL || ''),
},
// The ingest and shadow jobs. REPORT_JOBS=0 turns both off (tests do this by never starting them).
ddiff --git a/server/index.js b/server/index.js
iindex 13f00c2..ca11857 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -23,7 +23,7 @@ function createLifecycle({ server, exit, signals, ctx } = {}) {
exit,
signals,
deadlineMs: 10_000,
-- stop: [() => (jobs ? jobs.stop() : undefined)],
++ stop: [() => (jobs ? jobs.stop() : undefined), () => (ctx && ctx.searchIndex ? ctx.searchIndex.stop() : undefined)],
close: [() => (store ? store.close() : undefined)],
});
}
@@@ -39,6 +39,8 @@ async function start() {
ctx.jobs = allJobs;
jobs.start();
aiUsageJobs.start();
++ // The Rundown topic pages in OpenVibe.Search: a sweep a minute after boot and every hour.
++ if (!config.drill && ctx.searchIndex) ctx.searchIndex.start();
// Socket activation (2026-10-31): systemd's .socket unit holds the listener across restarts, so a deploy
// queues requests in the kernel instead of answering 502. Without the socket unit, the usual listen.
ddiff --git a/server/search-index.js b/server/search-index.js
nnew file mode 100644
iindex 0000000..cd0c203
---- /dev/null
++++ b/server/search-index.js
@@@ -0,0 +1,113 @@
++'use strict';
++
++/**
++ * OpenVibe.Report's Rundown topic pages in OpenVibe.Search (openvibe-publishing/search-feed, contracts 0.275.0): one
++ * search.index-document@1 per topic (/rundown/t/<slug>, the slug as the id), sent as report.index_document.upserted|
++ * deleted through this service's events outbox. Live's own topic pages are noindex; Report's are the public copy, so
++ * they are the ones Search gets. The document is what the page says: the topic's title, its headline and the summary
++ * of what streamers said, its kind and how many takes and streamers it has. 60 topics were on no network search
++ * (2026-11-02).
++ *
++ * The topics come from OpenVibe.Live's Rundown (data/live.js, cached five minutes) and change through the day, so a
++ * sweep a minute after boot and every hour: the sequencer (report_index_revisions) gives an unchanged topic its old
++ * revision and a quiet sweep sends nothing; a topic that left the Rundown gets a tombstone. When Live does not answer
++ * the sweep stops rather than read an empty list as "every topic is gone". Off under REPORT_DRILL and without a
++ * database.
++ */
++const { createServiceOutbox } = require('openvibe-sdk/events');
++const { createSearchFeed } = require('openvibe-publishing/search-feed');
++
++const EVENT_TYPES = ['report.index_document.upserted', 'report.index_document.deleted'];
++const START_DELAY_MS = 60_000;
++const INTERVAL_MS = 60 * 60_000;
++const SLUG_RE = /^[a-z0-9][a-z0-9-]{0,89}$/;
++const KIND = { news: 'News', games: 'Games', tech: 'Tech', life: 'Life' };
++
++const plural = (n, one, many) => `${n} ${n === 1 ? one : many}`;
++// Live's times are UTC without a zone ("2026-10-25 22:37:36").
++const iso = (t) => {
++ if (!t) return null;
++ const d = new Date(/[zZ]|[+-]\d\d:?\d\d$/.test(String(t)) ? t : `${String(t).replace(' ', 'T')}Z`);
++ return Number.isNaN(d.getTime()) ? null : d.toISOString();
++};
++
++/** One topic from Live's list → what its Report page says, as search-feed's document description. */
++function describeTopic(t) {
++ const takes = Number(t.takes) || 0;
++ const streamers = Number(t.streamers) || 0;
++ const count = takes ? `${plural(takes, 'take', 'takes')} from ${plural(streamers || 1, 'streamer', 'streamers')}` : '';
++ const kind = KIND[t.kind] || '';
++ // Live's headlines end without a full stop; the summary adds one before the counts.
++ const head = t.headline ? (/[.!?…]$/.test(String(t.headline).trim()) ? String(t.headline).trim() : `${String(t.headline).trim()}.`) : '';
++ return {
++ listed: true,
++ noindex: !takes,
++ title: `${t.title}: The Rundown`,
++ summary: [head, count ? `${count} on OpenVibe.Live.` : ''].filter(Boolean).join(' '),
++ body: [t.title, t.headline, t.summary, kind ? `${kind} topic on The Rundown.` : '', count].filter(Boolean).join('\n'),
++ facets: { ...(t.kind ? { kind: t.kind } : {}), takes, streamers },
++ authorship: { mode: 'ai' },
++ publishedAt: iso(t.first_seen_at),
++ updatedAt: iso(t.last_seen_at),
++ };
++}
++
++function createSearchIndex({ config, store, live, log = console, outbox = null }) {
++ const db = store.db;
++ const out = outbox || createServiceOutbox({
++ db, source: 'report', eventsUrl: (config.events && config.events.url) || null, networkInternalUrl: config.networkInternalUrl,
++ clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, log, eventTypes: EVENT_TYPES,
++ });
++
++ // Every topic, in slug order. Throws when Live does not answer: a sweep must never take an empty answer for an
++ // empty Rundown and tombstone every page.
++ async function list() {
++ const data = await live.topics();
++ if (!data || !Array.isArray(data.topics)) throw new Error('OpenVibe.Live did not answer the topic list');
++ return data.topics.filter((t) => t && SLUG_RE.test(String(t.slug || '')) && t.title)
++ .map((t) => ({ ...t, id: t.slug }))
++ .sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0));
++ }
++
++ const feed = createSearchFeed({
++ owner: 'report', db, outbox: out, baseUrl: config.baseUrl, now: store.now, log,
++ types: {
++ topic: {
++ page: (row) => `/rundown/t/${row.id}`,
++ document: describeTopic,
++ rows: async (after, limit) => (await list()).filter((t) => t.id > String(after || '')).slice(0, limit),
++ exists: async (ids) => { const have = new Set((await list()).map((t) => t.id)); return ids.filter((id) => have.has(id)); },
++ },
++ },
++ });
++ const timers = [];
++
++ async function sweep() {
++ const res = await feed.sweep();
++ const r = res.topic;
++ if (r.sent || r.removed || r.failed) log.log(`[Search] Rundown topics: ${r.sent} sent, ${r.removed} removed, ${r.failed} failed of ${r.seen}`);
++ if (out.kick) out.kick().catch(() => {});
++ return res;
++ }
++ const quietly = () => { sweep().catch((err) => log.warn(`[Search] sweep failed: ${(err && err.message) || err}`)); };
++
++ function start() {
++ if (config.drill || timers.length) return false;
++ out.start();
++ const kick = setTimeout(quietly, START_DELAY_MS);
++ if (kick.unref) kick.unref();
++ const tick = setInterval(quietly, INTERVAL_MS);
++ if (tick.unref) tick.unref();
++ timers.push(kick, tick);
++ return true;
++ }
++
++ function stop() {
++ for (const t of timers.splice(0)) { clearTimeout(t); clearInterval(t); }
++ out.stop();
++ }
++
++ return { feed, outbox: out, sweep, start, stop, status: () => out.status() };
++}
++
++module.exports = { createSearchIndex, describeTopic, EVENT_TYPES };
ddiff --git a/test/search-index.test.js b/test/search-index.test.js
nnew file mode 100644
iindex 0000000..75681fa
---- /dev/null
++++ b/test/search-index.test.js
@@@ -0,0 +1,80 @@
++'use strict';
++/**
++ * The Rundown topic pages in OpenVibe.Search (server/search-index.js): a sweep puts every topic in the outbox as one
++ * report.index_document.upserted, exactly the contract (contracts 0.275.0), for a /rundown/t/<slug> page that answers;
++ * a second sweep sends nothing; a topic that left the Rundown becomes a tombstone, once; a Live that does not answer
++ * stops the sweep instead of tombstoning every topic.
++ */
++const assert = require('assert');
++const contracts = require('openvibe-contracts');
++const { boot } = require('./helpers/boot');
++const { createSearchIndex, describeTopic } = require('../server/search-index');
++
++const TOPICS = [
++ { slug: 'taking-a-break', title: 'Taking a break', kind: 'life', headline: 'Streamers debate stepping away', summary: 'Fresh air and robot faces.', takes: 5, streamers: 2, first_seen_at: '2026-10-25 22:37:36', last_seen_at: '2026-11-01 10:00:00' },
++ { slug: 'seattle', title: 'Seattle', kind: 'life', headline: 'A headline', summary: 'A summary.', takes: 3, streamers: 1, first_seen_at: '2026-10-20 10:00:00', last_seen_at: '2026-10-27 10:00:00' },
++ { slug: 'ai-progress', title: 'AI progress', kind: 'tech', headline: 'Models got better', summary: 'Two streamers on new models.', takes: 0, streamers: 0, first_seen_at: '2026-10-21 10:00:00', last_seen_at: '2026-10-21 10:00:00' },
++];
++
++let failures = 0;
++async function check(name, fn) {
++ try { await fn(); process.stdout.write(` ✓ ${name}\n`); } catch (e) { failures++; process.stdout.write(` ✗ ${name}\n ${(e.stack || String(e)).split('\n').slice(0, 8).join('\n ')}\n`); }
++}
++
++(async () => {
++ const t = await boot({ db: true });
++ try {
++ let answer = { topics: TOPICS.slice() };
++ const live = { topics: async () => answer };
++ const searchIndex = createSearchIndex({ config: t.config, store: t.ctx.store, live, log: { log() {}, warn() {} } });
++ const db = t.ctx.store.db;
++ const outbox = async () => (await db.many('SELECT envelope FROM event_outbox ORDER BY id')).map((r) => r.envelope);
++ const valid = (env) => {
++ assert.ok(contracts.validate('events.event-envelope@1', env).valid, 'envelope');
++ const p = contracts.validate(env.event_type, env.payload);
++ assert.ok(p.valid, `${env.event_type}: ${JSON.stringify(p.errors)}`);
++ assert.strictEqual(env.source, 'report');
++ return env;
++ };
++
++ await check('a sweep sends every topic once, as the contract says, and the page it names answers', async () => {
++ const res = await searchIndex.sweep();
++ assert.deepStrictEqual(res.topic, { seen: 3, sent: 3, removed: 0, failed: 0 });
++ const docs = (await outbox()).map(valid).map((e) => e.payload);
++ const doc = docs.find((d) => d.id === 'seattle');
++ assert.strictEqual(doc.canonical_url, 'https://openvibe.report/rundown/t/seattle');
++ assert.strictEqual(doc.title, 'Seattle: The Rundown');
++ assert.strictEqual((await t.get('/rundown/t/seattle')).status, 200);
++ assert.strictEqual(docs.find((d) => d.id === 'ai-progress').indexability.decision, 'noindex', 'a topic with no take is noindex');
++ assert.strictEqual((await searchIndex.sweep()).topic.sent, 0, 'a second sweep sends nothing');
++ });
++
++ await check('a topic that left the Rundown becomes a tombstone, once', async () => {
++ answer = { topics: TOPICS.filter((x) => x.slug !== 'seattle') };
++ const before = (await outbox()).length;
++ assert.strictEqual((await searchIndex.sweep()).topic.removed, 1);
++ const env = valid((await outbox()).slice(before).find((e) => e.event_type === 'report.index_document.deleted'));
++ assert.deepStrictEqual([env.payload.type, env.payload.id], ['topic', 'seattle']);
++ assert.strictEqual((await searchIndex.sweep()).topic.removed, 0);
++ });
++
++ await check('when Live does not answer, the sweep stops and tombstones nothing', async () => {
++ answer = null;
++ const before = (await outbox()).length;
++ await assert.rejects(() => searchIndex.sweep(), /did not answer/);
++ assert.strictEqual((await outbox()).length, before);
++ });
++
++ await check('the words: headline, takes and streamers; Live\'s zone-less times are UTC', async () => {
++ const d = describeTopic(TOPICS[0]);
++ assert.strictEqual(d.summary, 'Streamers debate stepping away. 5 takes from 2 streamers on OpenVibe.Live.');
++ assert.deepStrictEqual(d.facets, { kind: 'life', takes: 5, streamers: 2 });
++ assert.strictEqual(d.publishedAt, '2026-10-25T22:37:36.000Z');
++ assert.strictEqual(d.noindex, false);
++ });
++ } finally {
++ await t.close();
++ process.stdout.write(failures ? `\n${failures} failed\n` : '\nall passed\n');
++ process.exit(failures ? 1 : 0);
++ }
++})();