openvibers/openvibe.community Commits
Pulse is fed from OpenVibe.Events: POST /internal/events (signature v2 only, loopback only, exactly once via the openvibe-sdk inbox) turns public live.stream.started, blog.post.published, wiki.page.published and news.story.published into Pulse items with provenance (a person only when a person acted; noindex or non-public items never enter). Subscriptions are created at boot (COMMUNITY_EVENTS_SECRET, EVENTS_URL). test/pulse-consumer.test.js
Changes
@@@ -165,6 +165,10 @@ function createApp(opts = {}) {const cors = v1.cors(config.apiCorsOrigins);app.use('/api/v1/comments', cors, createCommentsApi({ service: comments, viewers, anonWriteLimiter, resolveLimiter }));app.use('/api/v1/pulse', cors, createPulseApi({ pulse, viewers }));++ // OpenVibe.Events → Pulse (server/pulse/consumer.js): public activity from Live, Blog, Wiki and News.++ const pulseConsumer = require('./pulse/consumer').createPulseConsumer({ db, secrets: String(process.env.COMMUNITY_EVENTS_SECRET || '').split(',').map((s) => s.trim()).filter(Boolean) });++ app.locals.pulseConsumer = pulseConsumer;++ app.use('/internal/events', pulseConsumer.router);app.use('/api/v1/spaces', createSpacesApi({ forum, viewers }));app.use('/api/v1/posts', createPostsApi({ forum, viewers }));app.use('/api/v1/relay', createRelayApi({ relay, db, viewers }));@@@ -17,6 +17,11 @@ const server = app.listen(config.port, config.host, () => {: `[Community] pastes via ${config.liveInternalUrl}/api/pastes, identity via ${config.networkUrl}`);});server.keepAliveTimeout = 65_000;++// Subscribe Pulse to public activity at Events (idempotent; off without EVENTS_URL / COMMUNITY_EVENTS_SECRET).++try {++ const secret = String(process.env.COMMUNITY_EVENTS_SECRET || '').split(',')[0].trim();++ require('./pulse/consumer').startSubscriptions({ config, port: config.port, secret });++} catch (err) { console.warn('[Pulse consumer] not subscribed:', err.message); }function shutdown(signal) {console.log(`[Community] ${signal} — closing`);nnew file mode 100644@@@ -0,0 +1,122 @@++'use strict';++/**++ * OpenVibe.Events → Pulse: POST /internal/events, the endpoint of Community's Events subscriptions++ * (consumer `community`, created at boot by startSubscriptions()). Public activity from other services++ * becomes Pulse items (roadmap §29 / §33: provenance kept, AI never shown as a person):++ *++ * live.stream.started "<channel> went live: <title>" → the channel page, actor the streamer++ * blog.post.published "New post on the <blog> blog" → the post (public, indexable only)++ * wiki.page.published "Wiki: <space> / <slug>" → the page (public, indexable only)++ * news.story.published "News: <topic>" → the story (public, indexable only)++ *++ * Exactly once: the openvibe-sdk inbox claims (consumer, event_id) in the same transaction as the Pulse++ * write. Signature v2 only (parseDelivery requireV2) under COMMUNITY_EVENTS_SECRET (comma-separated for++ * rotation, 32+ characters each); unset = 503. Loopback only: a request carrying a forwarding header++ * came through nginx and is refused.++ */++const express = require('express');++const { http, serviceAuth } = require('openvibe-contracts');++const { parseDelivery, createInbox } = require('openvibe-sdk/events');++const store = require('./store');++++const CONSUMER = 'community';++const TOPICS = Object.freeze(['live.stream.started', 'blog.post.published', 'wiki.page.published', 'news.story.published']);++const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;++const clean = (s, n) => String(s == null ? '' : s).replace(/\s+/g, ' ').trim().slice(0, n);++const httpsUrl = (u) => (typeof u === 'string' && /^https:\/\/[a-z0-9.-]+\.[a-z]{2,}(\/|$)/i.test(u) ? u.slice(0, 500) : null);++const subjectOf = (ref) => (ref && ref.type === 'user' && /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(ref.id) ? ref.id : null);++const publicIndexable = (p) => p && p.visibility === 'public' && (!p.indexability || p.indexability.decision !== 'noindex');++++/** An envelope → a Pulse item, or an 'ignored:*' reason. */++function itemFor(event) {++ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};++ const at = event.occurred_at || event.timestamp || new Date().toISOString();++ const subjectId = String((event.subject && event.subject.id) || '');++ if (event.event_type === 'live.stream.started') {++ if (event.visibility !== 'public' || event.source !== 'live') return 'ignored:not_public';++ const ch = p.channel || {};++ const url = httpsUrl(ch.url);++ if (!url || !subjectId) return 'ignored:payload';++ const name = clean(ch.display_name || ch.username, 60);++ const title = clean(p.title, 120);++ return { source_service: 'live', source_type: 'stream', source_id: subjectId, title: title ? `${name} went live: ${title}` : `${name} went live`, url, actor_subject: subjectOf(ch.subject), origin: 'user', occurred_at: p.started_at || at };++ }++ const kinds = { 'blog.post.published': ['blog', 'post'], 'wiki.page.published': ['wiki', 'page'], 'news.story.published': ['news', 'story'] };++ const k = kinds[event.event_type];++ if (!k) return 'ignored:type';++ if (event.source !== k[0]) return 'ignored:source';++ if (!publicIndexable(p)) return 'ignored:not_public';++ const url = httpsUrl(p.canonical_url);++ if (!url || !subjectId) return 'ignored:payload';++ const title = k[0] === 'blog' ? `New post on the ${clean(p.blog, 60) || 'OpenVibe'} blog`++ : k[0] === 'wiki' ? `Wiki: ${clean(p.space, 40)} / ${clean(p.slug, 60)}`++ : `News: ${clean(p.topic, 60) || 'a new story'}`;++ const actor = event.actor && event.actor.type === 'user' ? subjectOf(event.actor) : null;++ return { source_service: k[0], source_type: k[1], source_id: subjectId, title, url, actor_subject: actor, origin: actor ? 'user' : 'system', occurred_at: at };++}++++function createPulseConsumer({ db, secrets = [], now = () => Date.now(), log = console } = {}) {++ const keys = (secrets || []).filter((s) => typeof s === 'string' && s.length >= 32);++ const inbox = createInbox(db, { table: 'community_event_inbox', now });++ inbox.ensureSchema();++ const stats = { received: 0, applied: 0, duplicates: 0, ignored: 0, refused: 0, failed: 0, last_at: null };++ const router = express.Router();++ router.post('/', express.raw({ type: () => true, limit: '256kb' }), (req, res) => {++ const ctx = http.requestContext(req.headers);++ const problem = (status, code, detail) => http.sendProblem(res, status, code, { detail, ctx });++ if (req.headers['x-forwarded-for'] || req.headers['x-real-ip'] || req.headers['cf-connecting-ip']) return problem(403, 'community.internal_only', 'internal route');++ if (!keys.length) return problem(503, 'community.events_disabled', 'COMMUNITY_EVENTS_SECRET is not set');++ const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);++ let delivery = null;++ for (const s of keys) { delivery = parseDelivery(raw, req.headers, s, { requireV2: true, now: now() }); if (delivery) break; }++ if (!delivery) { stats.refused++; return problem(401, 'community.bad_signature', 'X-OpenVibe-Signature-V2 does not verify or is outside the replay window'); }++ const event = delivery.event;++ if (!event || !EVENT_ID_RE.test(String(event.event_id || '')) || typeof event.event_type !== 'string') { stats.refused++; return problem(400, 'community.bad_delivery', 'body must be { event: <envelope>, seq }'); }++ stats.received++; stats.last_at = new Date(now()).toISOString();++ const item = itemFor(event);++ if (typeof item === 'string') { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: item }); }++ try {++ const r = inbox.once(CONSUMER, event.event_id, () => ({ outcome: store.upsertItem(db, item).created ? 'pulse:created' : 'pulse:updated' }));++ if (r.duplicate) stats.duplicates++; else stats.applied++;++ return res.json({ event_id: event.event_id, duplicate: r.duplicate, outcome: r.duplicate ? null : r.result.outcome });++ } catch (err) {++ stats.failed++;++ log.error(`[Pulse consumer] ${event.event_id} (${event.event_type}) failed:`, err.message);++ return problem(500, 'community.event_failed', 'processing failed; it will be retried');++ }++ });++ return { router, stats: () => ({ ...stats, enabled: keys.length > 0 }) };++}++++/** Create any missing subscription for TOPICS at Events (idempotent; retried in the background at boot). */++function startSubscriptions({ config, port, secret, eventsUrl = process.env.EVENTS_URL, fetchImpl = globalThis.fetch, log = console }) {++ if (!eventsUrl || !secret || !config.oauth.clientSecret || process.env.COMMUNITY_EVENTS_SUBSCRIBE === '0') return null;++ const base = String(eventsUrl).replace(/\/+$/, '');++ const endpoint = process.env.COMMUNITY_EVENTS_ENDPOINT || `http://127.0.0.1:${port}/internal/events`;++ const tokens = serviceAuth.createTokenClient({ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, audience: 'openvibe.events', scope: 'events.subscription.manage', fetchImpl });++ const call = async (method, path, body) => {++ const res = await fetchImpl(`${base}${path}`, { method, headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await tokens.authHeaders()) }, body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(15000) });++ const json = await res.json().catch(() => ({}));++ return { status: res.status, ok: res.ok, body: json };++ };++ const attempt = async () => {++ const listed = await call('GET', '/api/v1/subscriptions');++ if (!listed.ok) throw new Error(`listing subscriptions: ${listed.status}`);++ const mine = (listed.body.subscriptions || []).filter((s) => s.endpoint === endpoint);++ for (const topic of TOPICS) {++ if (mine.some((s) => s.topic_pattern === topic)) continue;++ const r = await call('POST', '/api/v1/subscriptions', { topic_pattern: topic, endpoint, secret });++ if (!r.ok && r.status !== 409) throw new Error(`subscribing to ${topic}: ${r.status} ${r.body.code || ''}`);++ if (r.ok) log.log(`[Pulse consumer] subscription created: ${r.body.id} (${topic} → ${endpoint})`);++ }++ };++ const delays = [0, 10_000, 60_000, 5 * 60_000, 15 * 60_000];++ let i = 0;++ const run = () => attempt().catch((err) => {++ if (++i < delays.length) { const t = setTimeout(run, delays[i]); if (t.unref) t.unref(); } else log.warn('[Pulse consumer] subscriptions not created:', err.message);++ });++ const t = setTimeout(run, delays[0]); if (t.unref) t.unref();++ return { topics: TOPICS, endpoint };++}++++module.exports = { createPulseConsumer, startSubscriptions, itemFor, TOPICS };nnew file mode 100644@@@ -0,0 +1,53 @@++'use strict';++/** Pulse's Events consumer: signed deliveries become Pulse items once; forged, proxied or non-public ones never do. */++const assert = require('assert');++const http = require('http');++const express = require('express');++const { signDeliveryHeaders } = require('openvibe-sdk/events');++const { openDb } = require('../server/db');++const { createPulseConsumer } = require('../server/pulse/consumer');++++const SECRET = `whsec_${'ab'.repeat(32)}`;++const db = openDb(':memory:');++const app = express();++app.use('/internal/events', createPulseConsumer({ db, secrets: [SECRET], log: { error() {}, log() {}, warn() {} } }).router);++const USR = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ';++let seq = 0;++const evt = (over) => ({ event_id: `evt_01JAB2C3D4E5F6G7H8J9K0M${String(++seq).padStart(3, '0')}`.slice(0, 30), event_type: 'live.stream.started', version: 1, source: 'live', visibility: 'public', timestamp: new Date().toISOString(),++ subject: { type: 'stream', id: '2477', revision: 1 }, actor: { type: 'user', id: USR },++ payload: { stream_id: 2477, channel: { username: 'goosely', display_name: 'Goosely', url: 'https://openvibe.live/@goosely', subject: { type: 'user', id: USR } }, title: 'Night walk', started_at: new Date().toISOString() }, ...over });++++(async () => {++ const srv = await new Promise((r) => { const s = http.createServer(app).listen(0, '127.0.0.1', () => r(s)); });++ const post = (event, { secret = SECRET, headers = {} } = {}) => {++ const body = JSON.stringify({ event, seq: 1 });++ return fetch(`http://127.0.0.1:${srv.address().port}/internal/events`, { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, secret), ...headers } })++ .then(async (r) => ({ status: r.status, json: await r.json().catch(() => null) }));++ };++ const items = () => db.prepare('SELECT source_service, source_type, source_id, title, url, actor_subject, origin FROM pulse_items ORDER BY id').all();++++ const e1 = evt();++ let r = await post(e1);++ assert.strictEqual(r.status, 200); assert.strictEqual(r.json.outcome, 'pulse:created');++ assert.deepStrictEqual(items(), [{ source_service: 'live', source_type: 'stream', source_id: '2477', title: 'Goosely went live: Night walk', url: 'https://openvibe.live/@goosely', actor_subject: USR, origin: 'user' }]);++ r = await post(e1);++ assert.strictEqual(r.json.duplicate, true, 'a redelivery does nothing');++ assert.strictEqual(items().length, 1);++++ assert.strictEqual((await post(evt(), { secret: `whsec_${'cd'.repeat(32)}` })).status, 401, 'forged signature');++ assert.strictEqual((await post(evt(), { headers: { 'x-forwarded-for': '1.2.3.4' } })).status, 403, 'came through the proxy');++ r = await post(evt({ visibility: 'internal', subject: { type: 'stream', id: '9', revision: 1 } }));++ assert.strictEqual(r.json.outcome, 'ignored:not_public');++ r = await post(evt({ event_type: 'blog.post.published', source: 'blog', subject: { type: 'post', id: 'pst_1' }, actor: { type: 'service', id: 'blog' },++ payload: { blog: 'openvibe', canonical_url: 'https://openvibe.blog/@openvibe/hello', publication_state: 'published', visibility: 'public', indexability: { decision: 'index', reasons: [] } } }));++ assert.strictEqual(r.json.outcome, 'pulse:created');++ assert.strictEqual(items()[1].title, 'New post on the openvibe blog');++ assert.strictEqual(items()[1].origin, 'system', 'no person when a service published it');++ r = await post(evt({ event_type: 'wiki.page.published', source: 'wiki', subject: { type: 'page', id: 'pg_2' },++ payload: { canonical_url: 'https://openvibe.wiki/w/help/start', publication_state: 'published', visibility: 'public', space: 'help', slug: 'start', indexability: { decision: 'noindex', reasons: ['thin'] } } }));++ assert.strictEqual(r.json.outcome, 'ignored:not_public', 'a noindex page stays out of Pulse');++ assert.strictEqual(items().length, 2);++ srv.close();++ console.log('pulse consumer: all checks passed');++ process.exit(0);++})().catch((e) => { console.error(e); process.exit(1); });
@@ -165,6 +165,10 @@ function createApp(opts = {}) {
@@ -165,6 +165,10 @@ function createApp(opts = {}) {
const cors = v1.cors(config.apiCorsOrigins);
const cors = v1.cors(config.apiCorsOrigins);
app.use('/api/v1/comments', cors, createCommentsApi({ service: comments, viewers, anonWriteLimiter, resolveLimiter }));
app.use('/api/v1/comments', cors, createCommentsApi({ service: comments, viewers, anonWriteLimiter, resolveLimiter }));
app.use('/api/v1/pulse', cors, createPulseApi({ pulse, viewers }));
app.use('/api/v1/pulse', cors, createPulseApi({ pulse, viewers }));
+ // OpenVibe.Events → Pulse (server/pulse/consumer.js): public activity from Live, Blog, Wiki and News.
+ const pulseConsumer = require('./pulse/consumer').createPulseConsumer({ db, secrets: String(process.env.COMMUNITY_EVENTS_SECRET || '').split(',').map((s) => s.trim()).filter(Boolean) });
+ app.locals.pulseConsumer = pulseConsumer;
+ app.use('/internal/events', pulseConsumer.router);
app.use('/api/v1/spaces', createSpacesApi({ forum, viewers }));
app.use('/api/v1/spaces', createSpacesApi({ forum, viewers }));
app.use('/api/v1/posts', createPostsApi({ forum, viewers }));
app.use('/api/v1/posts', createPostsApi({ forum, viewers }));
app.use('/api/v1/relay', createRelayApi({ relay, db, viewers }));
app.use('/api/v1/relay', createRelayApi({ relay, db, viewers }));
@@ -17,6 +17,11 @@ const server = app.listen(config.port, config.host, () => {
@@ -17,6 +17,11 @@ const server = app.listen(config.port, config.host, () => {
: `[Community] pastes via ${config.liveInternalUrl}/api/pastes, identity via ${config.networkUrl}`);
: `[Community] pastes via ${config.liveInternalUrl}/api/pastes, identity via ${config.networkUrl}`);
});
});
server.keepAliveTimeout = 65_000;
server.keepAliveTimeout = 65_000;
+// Subscribe Pulse to public activity at Events (idempotent; off without EVENTS_URL / COMMUNITY_EVENTS_SECRET).
+try {
+ const secret = String(process.env.COMMUNITY_EVENTS_SECRET || '').split(',')[0].trim();
+ require('./pulse/consumer').startSubscriptions({ config, port: config.port, secret });
+} catch (err) { console.warn('[Pulse consumer] not subscribed:', err.message); }
function shutdown(signal) {
function shutdown(signal) {
console.log(`[Community] ${signal} — closing`);
console.log(`[Community] ${signal} — closing`);
new file mode 100644
new file mode 100644
@@ -0,0 +1,122 @@
@@ -0,0 +1,122 @@
+'use strict';
+/**
+ * OpenVibe.Events → Pulse: POST /internal/events, the endpoint of Community's Events subscriptions
+ * (consumer `community`, created at boot by startSubscriptions()). Public activity from other services
+ * becomes Pulse items (roadmap §29 / §33: provenance kept, AI never shown as a person):
+ *
+ * live.stream.started "<channel> went live: <title>" → the channel page, actor the streamer
+ * blog.post.published "New post on the <blog> blog" → the post (public, indexable only)
+ * wiki.page.published "Wiki: <space> / <slug>" → the page (public, indexable only)
+ * news.story.published "News: <topic>" → the story (public, indexable only)
+ *
+ * Exactly once: the openvibe-sdk inbox claims (consumer, event_id) in the same transaction as the Pulse
+ * write. Signature v2 only (parseDelivery requireV2) under COMMUNITY_EVENTS_SECRET (comma-separated for
+ * rotation, 32+ characters each); unset = 503. Loopback only: a request carrying a forwarding header
+ * came through nginx and is refused.
+ */
+const express = require('express');
+const { http, serviceAuth } = require('openvibe-contracts');
+const { parseDelivery, createInbox } = require('openvibe-sdk/events');
+const store = require('./store');
+
+const CONSUMER = 'community';
+const TOPICS = Object.freeze(['live.stream.started', 'blog.post.published', 'wiki.page.published', 'news.story.published']);
+const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
+const clean = (s, n) => String(s == null ? '' : s).replace(/\s+/g, ' ').trim().slice(0, n);
+const httpsUrl = (u) => (typeof u === 'string' && /^https:\/\/[a-z0-9.-]+\.[a-z]{2,}(\/|$)/i.test(u) ? u.slice(0, 500) : null);
+const subjectOf = (ref) => (ref && ref.type === 'user' && /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(ref.id) ? ref.id : null);
+const publicIndexable = (p) => p && p.visibility === 'public' && (!p.indexability || p.indexability.decision !== 'noindex');
+
+/** An envelope → a Pulse item, or an 'ignored:*' reason. */
+function itemFor(event) {
+ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
+ const at = event.occurred_at || event.timestamp || new Date().toISOString();
+ const subjectId = String((event.subject && event.subject.id) || '');
+ if (event.event_type === 'live.stream.started') {
+ if (event.visibility !== 'public' || event.source !== 'live') return 'ignored:not_public';
+ const ch = p.channel || {};
+ const url = httpsUrl(ch.url);
+ if (!url || !subjectId) return 'ignored:payload';
+ const name = clean(ch.display_name || ch.username, 60);
+ const title = clean(p.title, 120);
+ return { source_service: 'live', source_type: 'stream', source_id: subjectId, title: title ? `${name} went live: ${title}` : `${name} went live`, url, actor_subject: subjectOf(ch.subject), origin: 'user', occurred_at: p.started_at || at };
+ }
+ const kinds = { 'blog.post.published': ['blog', 'post'], 'wiki.page.published': ['wiki', 'page'], 'news.story.published': ['news', 'story'] };
+ const k = kinds[event.event_type];
+ if (!k) return 'ignored:type';
+ if (event.source !== k[0]) return 'ignored:source';
+ if (!publicIndexable(p)) return 'ignored:not_public';
+ const url = httpsUrl(p.canonical_url);
+ if (!url || !subjectId) return 'ignored:payload';
+ const title = k[0] === 'blog' ? `New post on the ${clean(p.blog, 60) || 'OpenVibe'} blog`
+ : k[0] === 'wiki' ? `Wiki: ${clean(p.space, 40)} / ${clean(p.slug, 60)}`
+ : `News: ${clean(p.topic, 60) || 'a new story'}`;
+ const actor = event.actor && event.actor.type === 'user' ? subjectOf(event.actor) : null;
+ return { source_service: k[0], source_type: k[1], source_id: subjectId, title, url, actor_subject: actor, origin: actor ? 'user' : 'system', occurred_at: at };
+}
+
+function createPulseConsumer({ db, secrets = [], now = () => Date.now(), log = console } = {}) {
+ const keys = (secrets || []).filter((s) => typeof s === 'string' && s.length >= 32);
+ const inbox = createInbox(db, { table: 'community_event_inbox', now });
+ inbox.ensureSchema();
+ const stats = { received: 0, applied: 0, duplicates: 0, ignored: 0, refused: 0, failed: 0, last_at: null };
+ const router = express.Router();
+ router.post('/', express.raw({ type: () => true, limit: '256kb' }), (req, res) => {
+ const ctx = http.requestContext(req.headers);
+ const problem = (status, code, detail) => http.sendProblem(res, status, code, { detail, ctx });
+ if (req.headers['x-forwarded-for'] || req.headers['x-real-ip'] || req.headers['cf-connecting-ip']) return problem(403, 'community.internal_only', 'internal route');
+ if (!keys.length) return problem(503, 'community.events_disabled', 'COMMUNITY_EVENTS_SECRET is not set');
+ const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
+ let delivery = null;
+ for (const s of keys) { delivery = parseDelivery(raw, req.headers, s, { requireV2: true, now: now() }); if (delivery) break; }
+ if (!delivery) { stats.refused++; return problem(401, 'community.bad_signature', 'X-OpenVibe-Signature-V2 does not verify or is outside the replay window'); }
+ const event = delivery.event;
+ if (!event || !EVENT_ID_RE.test(String(event.event_id || '')) || typeof event.event_type !== 'string') { stats.refused++; return problem(400, 'community.bad_delivery', 'body must be { event: <envelope>, seq }'); }
+ stats.received++; stats.last_at = new Date(now()).toISOString();
+ const item = itemFor(event);
+ if (typeof item === 'string') { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: item }); }
+ try {
+ const r = inbox.once(CONSUMER, event.event_id, () => ({ outcome: store.upsertItem(db, item).created ? 'pulse:created' : 'pulse:updated' }));
+ if (r.duplicate) stats.duplicates++; else stats.applied++;
+ return res.json({ event_id: event.event_id, duplicate: r.duplicate, outcome: r.duplicate ? null : r.result.outcome });
+ } catch (err) {
+ stats.failed++;
+ log.error(`[Pulse consumer] ${event.event_id} (${event.event_type}) failed:`, err.message);
+ return problem(500, 'community.event_failed', 'processing failed; it will be retried');
+ }
+ });
+ return { router, stats: () => ({ ...stats, enabled: keys.length > 0 }) };
+}
+
+/** Create any missing subscription for TOPICS at Events (idempotent; retried in the background at boot). */
+function startSubscriptions({ config, port, secret, eventsUrl = process.env.EVENTS_URL, fetchImpl = globalThis.fetch, log = console }) {
+ if (!eventsUrl || !secret || !config.oauth.clientSecret || process.env.COMMUNITY_EVENTS_SUBSCRIBE === '0') return null;
+ const base = String(eventsUrl).replace(/\/+$/, '');
+ const endpoint = process.env.COMMUNITY_EVENTS_ENDPOINT || `http://127.0.0.1:${port}/internal/events`;
+ const tokens = serviceAuth.createTokenClient({ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, audience: 'openvibe.events', scope: 'events.subscription.manage', fetchImpl });
+ const call = async (method, path, body) => {
+ const res = await fetchImpl(`${base}${path}`, { method, headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await tokens.authHeaders()) }, body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(15000) });
+ const json = await res.json().catch(() => ({}));
+ return { status: res.status, ok: res.ok, body: json };
+ };
+ const attempt = async () => {
+ const listed = await call('GET', '/api/v1/subscriptions');
+ if (!listed.ok) throw new Error(`listing subscriptions: ${listed.status}`);
+ const mine = (listed.body.subscriptions || []).filter((s) => s.endpoint === endpoint);
+ for (const topic of TOPICS) {
+ if (mine.some((s) => s.topic_pattern === topic)) continue;
+ const r = await call('POST', '/api/v1/subscriptions', { topic_pattern: topic, endpoint, secret });
+ if (!r.ok && r.status !== 409) throw new Error(`subscribing to ${topic}: ${r.status} ${r.body.code || ''}`);
+ if (r.ok) log.log(`[Pulse consumer] subscription created: ${r.body.id} (${topic} → ${endpoint})`);
+ }
+ };
+ const delays = [0, 10_000, 60_000, 5 * 60_000, 15 * 60_000];
+ let i = 0;
+ const run = () => attempt().catch((err) => {
+ if (++i < delays.length) { const t = setTimeout(run, delays[i]); if (t.unref) t.unref(); } else log.warn('[Pulse consumer] subscriptions not created:', err.message);
+ });
+ const t = setTimeout(run, delays[0]); if (t.unref) t.unref();
+ return { topics: TOPICS, endpoint };
+}
+
+module.exports = { createPulseConsumer, startSubscriptions, itemFor, TOPICS };
new file mode 100644
new file mode 100644
@@ -0,0 +1,53 @@
@@ -0,0 +1,53 @@
+'use strict';
+/** Pulse's Events consumer: signed deliveries become Pulse items once; forged, proxied or non-public ones never do. */
+const assert = require('assert');
+const http = require('http');
+const express = require('express');
+const { signDeliveryHeaders } = require('openvibe-sdk/events');
+const { openDb } = require('../server/db');
+const { createPulseConsumer } = require('../server/pulse/consumer');
+
+const SECRET = `whsec_${'ab'.repeat(32)}`;
+const db = openDb(':memory:');
+const app = express();
+app.use('/internal/events', createPulseConsumer({ db, secrets: [SECRET], log: { error() {}, log() {}, warn() {} } }).router);
+const USR = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ';
+let seq = 0;
+const evt = (over) => ({ event_id: `evt_01JAB2C3D4E5F6G7H8J9K0M${String(++seq).padStart(3, '0')}`.slice(0, 30), event_type: 'live.stream.started', version: 1, source: 'live', visibility: 'public', timestamp: new Date().toISOString(),
+ subject: { type: 'stream', id: '2477', revision: 1 }, actor: { type: 'user', id: USR },
+ payload: { stream_id: 2477, channel: { username: 'goosely', display_name: 'Goosely', url: 'https://openvibe.live/@goosely', subject: { type: 'user', id: USR } }, title: 'Night walk', started_at: new Date().toISOString() }, ...over });
+
+(async () => {
+ const srv = await new Promise((r) => { const s = http.createServer(app).listen(0, '127.0.0.1', () => r(s)); });
+ const post = (event, { secret = SECRET, headers = {} } = {}) => {
+ const body = JSON.stringify({ event, seq: 1 });
+ return fetch(`http://127.0.0.1:${srv.address().port}/internal/events`, { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, secret), ...headers } })
+ .then(async (r) => ({ status: r.status, json: await r.json().catch(() => null) }));
+ };
+ const items = () => db.prepare('SELECT source_service, source_type, source_id, title, url, actor_subject, origin FROM pulse_items ORDER BY id').all();
+
+ const e1 = evt();
+ let r = await post(e1);
+ assert.strictEqual(r.status, 200); assert.strictEqual(r.json.outcome, 'pulse:created');
+ assert.deepStrictEqual(items(), [{ source_service: 'live', source_type: 'stream', source_id: '2477', title: 'Goosely went live: Night walk', url: 'https://openvibe.live/@goosely', actor_subject: USR, origin: 'user' }]);
+ r = await post(e1);
+ assert.strictEqual(r.json.duplicate, true, 'a redelivery does nothing');
+ assert.strictEqual(items().length, 1);
+
+ assert.strictEqual((await post(evt(), { secret: `whsec_${'cd'.repeat(32)}` })).status, 401, 'forged signature');
+ assert.strictEqual((await post(evt(), { headers: { 'x-forwarded-for': '1.2.3.4' } })).status, 403, 'came through the proxy');
+ r = await post(evt({ visibility: 'internal', subject: { type: 'stream', id: '9', revision: 1 } }));
+ assert.strictEqual(r.json.outcome, 'ignored:not_public');
+ r = await post(evt({ event_type: 'blog.post.published', source: 'blog', subject: { type: 'post', id: 'pst_1' }, actor: { type: 'service', id: 'blog' },
+ payload: { blog: 'openvibe', canonical_url: 'https://openvibe.blog/@openvibe/hello', publication_state: 'published', visibility: 'public', indexability: { decision: 'index', reasons: [] } } }));
+ assert.strictEqual(r.json.outcome, 'pulse:created');
+ assert.strictEqual(items()[1].title, 'New post on the openvibe blog');
+ assert.strictEqual(items()[1].origin, 'system', 'no person when a service published it');
+ r = await post(evt({ event_type: 'wiki.page.published', source: 'wiki', subject: { type: 'page', id: 'pg_2' },
+ payload: { canonical_url: 'https://openvibe.wiki/w/help/start', publication_state: 'published', visibility: 'public', space: 'help', slug: 'start', indexability: { decision: 'noindex', reasons: ['thin'] } } }));
+ assert.strictEqual(r.json.outcome, 'ignored:not_public', 'a noindex page stays out of Pulse');
+ assert.strictEqual(items().length, 2);
+ srv.close();
+ console.log('pulse consumer: all checks passed');
+ process.exit(0);
+})().catch((e) => { console.error(e); process.exit(1); });