openvibers/openvibe.community Commits
VIP convergence (W10): Community subscribes to vip.membership.changed on its Events endpoint; a delivery from VIP drops the member's cached members-only answers for that creator at once (the VIP gate's cache handleEvent, once per event through the inbox) instead of waiting out the TTL; other sources are ignored
Changes
@@@ -166,7 +166,7 @@ function createApp(opts = {}) {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) });++ const pulseConsumer = require('./pulse/consumer').createPulseConsumer({ db, vipCache: vip && vip.cache, 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 }));@@@ -9,6 +9,9 @@* wiki.page.published "Wiki: <space> / <slug>" → the page (public, indexable only)* news.story.published "News: <topic>" → the story (public, indexable only)*++ * And VIP convergence: vip.membership.changed (source vip) drops the member's cached members-only++ * answers for that creator at once (the VIP gate's cache handleEvent) instead of waiting out its TTL.++ ** 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@@@ -20,7 +23,7 @@ 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 TOPICS = Object.freeze(['live.stream.started', 'blog.post.published', 'wiki.page.published', 'news.story.published', 'vip.membership.changed']);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);@@@ -55,7 +58,7 @@ function itemFor(event) {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 } = {}) {++function createPulseConsumer({ db, secrets = [], vipCache = null, 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();@@@ -73,6 +76,12 @@ function createPulseConsumer({ db, secrets = [], now = () => Date.now(), log = cconst 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();++ if (event.event_type === 'vip.membership.changed') {++ if (event.source !== 'vip' || !vipCache) { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: event.source !== 'vip' ? 'ignored:source' : 'ignored:no_gate' }); }++ const r = inbox.once(CONSUMER, event.event_id, () => ({ outcome: vipCache.handleEvent(event) ? 'vip:invalidated' : 'vip:unchanged' }));++ 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 });++ }const item = itemFor(event);if (typeof item === 'string') { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: item }); }try {@@@ -10,7 +10,8 @@ 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 vipDrops = [];++app.use('/internal/events', createPulseConsumer({ db, secrets: [SECRET], vipCache: { handleEvent: (e) => { vipDrops.push(e.event_id); return true; } }, 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(),@@@ -47,6 +48,18 @@ const evt = (over) => ({ event_id: `evt_01JAB2C3D4E5F6G7H8J9K0M${String(++seq).ppayload: { 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);++ // VIP convergence: a membership change from VIP drops cached members-only answers once; from anyone else, nothing.++ const vipEv = (source) => evt({ event_type: 'vip.membership.changed', source, subject: { type: 'membership', id: 'mbr_1', revision: 2 }, actor: { type: 'service', id: 'vip' },++ payload: { member: { type: 'user', id: USR }, creator: { type: 'user', id: USR }, status: 'canceled' } });++ const v1 = vipEv('vip');++ r = await post(v1);++ assert.strictEqual(r.json.outcome, 'vip:invalidated');++ r = await post(v1);++ assert.strictEqual(r.json.duplicate, true, 'a redelivery drops nothing twice');++ r = await post(vipEv('live'));++ assert.strictEqual(r.json.outcome, 'ignored:source');++ assert.deepStrictEqual(vipDrops, [v1.event_id]);++ assert.strictEqual(items().length, 2, 'no Pulse item for a membership');srv.close();console.log('pulse consumer: all checks passed');process.exit(0);
@@ -166,7 +166,7 @@ function createApp(opts = {}) {
@@ -166,7 +166,7 @@ function createApp(opts = {}) {
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.
// 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) });
+ const pulseConsumer = require('./pulse/consumer').createPulseConsumer({ db, vipCache: vip && vip.cache, secrets: String(process.env.COMMUNITY_EVENTS_SECRET || '').split(',').map((s) => s.trim()).filter(Boolean) });
app.locals.pulseConsumer = pulseConsumer;
app.locals.pulseConsumer = pulseConsumer;
app.use('/internal/events', pulseConsumer.router);
app.use('/internal/events', pulseConsumer.router);
app.use('/api/v1/spaces', createSpacesApi({ forum, viewers }));
app.use('/api/v1/spaces', createSpacesApi({ forum, viewers }));
@@ -9,6 +9,9 @@
@@ -9,6 +9,9 @@
* wiki.page.published "Wiki: <space> / <slug>" → the page (public, indexable only)
* wiki.page.published "Wiki: <space> / <slug>" → the page (public, indexable only)
* news.story.published "News: <topic>" → the story (public, indexable only)
* news.story.published "News: <topic>" → the story (public, indexable only)
*
*
+ * And VIP convergence: vip.membership.changed (source vip) drops the member's cached members-only
+ * answers for that creator at once (the VIP gate's cache handleEvent) instead of waiting out its TTL.
+ *
* Exactly once: the openvibe-sdk inbox claims (consumer, event_id) in the same transaction as the Pulse
* 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
* 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
* rotation, 32+ characters each); unset = 503. Loopback only: a request carrying a forwarding header
@@ -20,7 +23,7 @@ const { parseDelivery, createInbox } = require('openvibe-sdk/events');
@@ -20,7 +23,7 @@ const { parseDelivery, createInbox } = require('openvibe-sdk/events');
const store = require('./store');
const store = require('./store');
const CONSUMER = 'community';
const CONSUMER = 'community';
-const TOPICS = Object.freeze(['live.stream.started', 'blog.post.published', 'wiki.page.published', 'news.story.published']);
+const TOPICS = Object.freeze(['live.stream.started', 'blog.post.published', 'wiki.page.published', 'news.story.published', 'vip.membership.changed']);
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
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 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 httpsUrl = (u) => (typeof u === 'string' && /^https:\/\/[a-z0-9.-]+\.[a-z]{2,}(\/|$)/i.test(u) ? u.slice(0, 500) : null);
@@ -55,7 +58,7 @@ function itemFor(event) {
@@ -55,7 +58,7 @@ function itemFor(event) {
return { source_service: k[0], source_type: k[1], source_id: subjectId, title, url, actor_subject: actor, origin: actor ? 'user' : 'system', occurred_at: at };
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 } = {}) {
+function createPulseConsumer({ db, secrets = [], vipCache = null, now = () => Date.now(), log = console } = {}) {
const keys = (secrets || []).filter((s) => typeof s === 'string' && s.length >= 32);
const keys = (secrets || []).filter((s) => typeof s === 'string' && s.length >= 32);
const inbox = createInbox(db, { table: 'community_event_inbox', now });
const inbox = createInbox(db, { table: 'community_event_inbox', now });
inbox.ensureSchema();
inbox.ensureSchema();
@@ -73,6 +76,12 @@ function createPulseConsumer({ db, secrets = [], now = () => Date.now(), log = c
@@ -73,6 +76,12 @@ function createPulseConsumer({ db, secrets = [], now = () => Date.now(), log = c
const event = delivery.event;
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 }'); }
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();
stats.received++; stats.last_at = new Date(now()).toISOString();
+ if (event.event_type === 'vip.membership.changed') {
+ if (event.source !== 'vip' || !vipCache) { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: event.source !== 'vip' ? 'ignored:source' : 'ignored:no_gate' }); }
+ const r = inbox.once(CONSUMER, event.event_id, () => ({ outcome: vipCache.handleEvent(event) ? 'vip:invalidated' : 'vip:unchanged' }));
+ 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 });
+ }
const item = itemFor(event);
const item = itemFor(event);
if (typeof item === 'string') { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: item }); }
if (typeof item === 'string') { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: item }); }
try {
try {
@@ -10,7 +10,8 @@ const { createPulseConsumer } = require('../server/pulse/consumer');
@@ -10,7 +10,8 @@ const { createPulseConsumer } = require('../server/pulse/consumer');
const SECRET = `whsec_${'ab'.repeat(32)}`;
const SECRET = `whsec_${'ab'.repeat(32)}`;
const db = openDb(':memory:');
const db = openDb(':memory:');
const app = express();
const app = express();
-app.use('/internal/events', createPulseConsumer({ db, secrets: [SECRET], log: { error() {}, log() {}, warn() {} } }).router);
+const vipDrops = [];
+app.use('/internal/events', createPulseConsumer({ db, secrets: [SECRET], vipCache: { handleEvent: (e) => { vipDrops.push(e.event_id); return true; } }, log: { error() {}, log() {}, warn() {} } }).router);
const USR = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ';
const USR = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ';
let seq = 0;
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(),
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(),
@@ -47,6 +48,18 @@ const evt = (over) => ({ event_id: `evt_01JAB2C3D4E5F6G7H8J9K0M${String(++seq).p
@@ -47,6 +48,18 @@ const evt = (over) => ({ event_id: `evt_01JAB2C3D4E5F6G7H8J9K0M${String(++seq).p
payload: { canonical_url: 'https://openvibe.wiki/w/help/start', publication_state: 'published', visibility: 'public', space: 'help', slug: 'start', indexability: { decision: 'noindex', reasons: ['thin'] } } }));
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(r.json.outcome, 'ignored:not_public', 'a noindex page stays out of Pulse');
assert.strictEqual(items().length, 2);
assert.strictEqual(items().length, 2);
+ // VIP convergence: a membership change from VIP drops cached members-only answers once; from anyone else, nothing.
+ const vipEv = (source) => evt({ event_type: 'vip.membership.changed', source, subject: { type: 'membership', id: 'mbr_1', revision: 2 }, actor: { type: 'service', id: 'vip' },
+ payload: { member: { type: 'user', id: USR }, creator: { type: 'user', id: USR }, status: 'canceled' } });
+ const v1 = vipEv('vip');
+ r = await post(v1);
+ assert.strictEqual(r.json.outcome, 'vip:invalidated');
+ r = await post(v1);
+ assert.strictEqual(r.json.duplicate, true, 'a redelivery drops nothing twice');
+ r = await post(vipEv('live'));
+ assert.strictEqual(r.json.outcome, 'ignored:source');
+ assert.deepStrictEqual(vipDrops, [v1.event_id]);
+ assert.strictEqual(items().length, 2, 'no Pulse item for a membership');
srv.close();
srv.close();
console.log('pulse consumer: all checks passed');
console.log('pulse consumer: all checks passed');
process.exit(0);
process.exit(0);