openvibers/openvibe.chat Commits
VIP convergence (W10): Chat subscribes to vip.membership.changed; a delivery from VIP drops that member's cached subscriber-badge answers for the creator at once (vip/badges handleEvent) instead of converging by the cache TTL; any other source is ignored
Changes
@@@ -13,6 +13,9 @@* revision newer than the cached copy drops it (../prefs/chat-preferences.js* handleEvent), so a change made elsewhere shows at once instead of within* CHAT_PREFS_TTL_MS. Other namespaces are acknowledged and not recorded.++ * vip.membership.changed a membership started, lapsed or was revoked (source vip): the member's cached++ * subscriber-badge answers for that creator are dropped at once++ * (../vip/badges.js handleEvent) instead of converging by the cache TTL.** Exactly once: the openvibe-sdk inbox claims (consumer, event_id) in the same SQLite transaction as* the change, so a redelivery does nothing and a failure rolls both back (Events retries). Broadcasts@@@ -33,10 +36,11 @@ const db = require('../db/database');const { viaProxy } = require('../net/service-auth');const deployNotice = require('../chat/deploy-notice');const prefs = require('../prefs/chat-preferences');++const vipBadges = require('../vip/badges');const CONSUMER = 'chat';const INBOX_TABLE = 'chat_event_inbox';--const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated']);++const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated', 'vip.membership.changed']);const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;const INBOX_KEEP_MS = 35 * 24 * 3600 * 1000; // Events keeps events 30 days: nothing older can be redelivered@@@ -67,6 +71,12 @@ function createEventsConsumer({ chatServer, secrets = [], now = () => Date.now()if (p.namespace !== prefs.NAMESPACE) return 'ignored:namespace';return () => (prefs.handleEvent(event) ? 'invalidated' : 'unchanged');}++ if (event.event_type === 'vip.membership.changed') {++ // A membership started, lapsed or was revoked: drop that member's cached badge answers for++ // that creator now instead of waiting out the cache TTL (VIP convergence).++ if (event.source !== 'vip') return 'ignored:source';++ return () => (vipBadges.handleEvent(event) ? 'invalidated' : 'unchanged');++ }return 'ignored:type';}@@@ -123,8 +123,8 @@ t('boot with the consumer on and a stub Events', async () => {t('boot created one subscription per topic, to /internal/events, with the consumer’s secret', async () => {const results = await h.subscriptions.done;-- assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created']]);-- assert.strictEqual(stubEvents.subs.length, 2);++ assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created'], ['vip.membership.changed', 'created']]);++ assert.strictEqual(stubEvents.subs.length, 3);for (const s of stubEvents.subs) {assert.strictEqual(s.endpoint, `http://127.0.0.1:${h.port}/internal/events`);assert.strictEqual(s.secret, SECRET, 'the first CHAT_EVENTS_SECRET is handed to Events');@@@ -139,17 +139,17 @@ t('a second boot changes nothing, and never re-enables a subscription an operatoconst endpoint = subs.endpointFor(config, h.port);const posts = stubEvents.posts;let r = await subs.ensure({ config, endpoint });-- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists']);++ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists']);assert.strictEqual(stubEvents.posts, posts, 'nothing was created again');r = await subs.ensure({ config, endpoint, action: 'disable', topics: ['live.release.deployed'] });assert.deepStrictEqual(r.map((x) => [x.result, x.enabled]), [['disabled', false]]);r = await subs.ensure({ config, endpoint });-- assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true]]);++ assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true], ['vip.membership.changed', 'exists', true]]);assert.strictEqual(stubEvents.subs[0].enabled, false, 'still disabled');r = await subs.ensure({ config, endpoint, action: 'enable', topics: ['live.release.deployed'] });assert.strictEqual(stubEvents.subs[0].enabled, true);r = await subs.ensure({ config, endpoint, action: 'list' });-- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists']);++ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists']);});t('signature v2 only: bad signature, stale or future timestamp, v1-only, unsigned → 401', async () => {@@@ -344,6 +344,28 @@ t('network.module.updated drops a cached chat.preferences copy when newer; an olassert.strictEqual(prefs._cached(ANN).revision, 3);});++t('vip.membership.changed from VIP drops the cached badge answers at once; from anyone else it is ignored', async () => {++ const vipBadges = require('../server/vip/badges');++ const calls = [];++ const orig = vipBadges.handleEvent;++ vipBadges.handleEvent = (e) => { calls.push(e.event_id); return true; };++ try {++ const ev = (source) => ({++ event_id: ids.newId('event'), event_type: 'vip.membership.changed', version: 1, source, actor: { type: 'service', id: 'vip' },++ timestamp: new Date().toISOString(), priority: 'important', visibility: 'internal', subject: { type: 'membership', id: 'mbr_1', revision: 2 },++ payload: { member: { type: 'user', id: ANN }, creator: { type: 'user', id: ANN }, status: 'canceled' },++ });++ const good = ev('vip');++ let r = await deliver(good);++ assert.strictEqual(r.status, 200);++ assert.strictEqual(r.body.outcome, 'invalidated');++ assert.deepStrictEqual(calls, [good.event_id]);++ r = await deliver(ev('live'));++ assert.strictEqual(r.body.outcome, 'ignored:source');++ assert.strictEqual(calls.length, 1, 'only VIP speaks for memberships');++ } finally { vipBadges.handleEvent = orig; }++});++t('an event delivered with the secret the subscription holds verifies; /ready reports the consumer', async () => {const sub = stubEvents.subs.find((s) => s.topic_pattern === 'live.release.deployed');speak('end-to-end next');
@@ -13,6 +13,9 @@
@@ -13,6 +13,9 @@
* revision newer than the cached copy drops it (../prefs/chat-preferences.js
* revision newer than the cached copy drops it (../prefs/chat-preferences.js
* handleEvent), so a change made elsewhere shows at once instead of within
* handleEvent), so a change made elsewhere shows at once instead of within
* CHAT_PREFS_TTL_MS. Other namespaces are acknowledged and not recorded.
* CHAT_PREFS_TTL_MS. Other namespaces are acknowledged and not recorded.
+ * vip.membership.changed a membership started, lapsed or was revoked (source vip): the member's cached
+ * subscriber-badge answers for that creator are dropped at once
+ * (../vip/badges.js handleEvent) instead of converging by the cache TTL.
*
*
* Exactly once: the openvibe-sdk inbox claims (consumer, event_id) in the same SQLite transaction as
* Exactly once: the openvibe-sdk inbox claims (consumer, event_id) in the same SQLite transaction as
* the change, so a redelivery does nothing and a failure rolls both back (Events retries). Broadcasts
* the change, so a redelivery does nothing and a failure rolls both back (Events retries). Broadcasts
@@ -33,10 +36,11 @@ const db = require('../db/database');
@@ -33,10 +36,11 @@ const db = require('../db/database');
const { viaProxy } = require('../net/service-auth');
const { viaProxy } = require('../net/service-auth');
const deployNotice = require('../chat/deploy-notice');
const deployNotice = require('../chat/deploy-notice');
const prefs = require('../prefs/chat-preferences');
const prefs = require('../prefs/chat-preferences');
+const vipBadges = require('../vip/badges');
const CONSUMER = 'chat';
const CONSUMER = 'chat';
const INBOX_TABLE = 'chat_event_inbox';
const INBOX_TABLE = 'chat_event_inbox';
-const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated']);
+const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated', '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 INBOX_KEEP_MS = 35 * 24 * 3600 * 1000; // Events keeps events 30 days: nothing older can be redelivered
const INBOX_KEEP_MS = 35 * 24 * 3600 * 1000; // Events keeps events 30 days: nothing older can be redelivered
@@ -67,6 +71,12 @@ function createEventsConsumer({ chatServer, secrets = [], now = () => Date.now()
@@ -67,6 +71,12 @@ function createEventsConsumer({ chatServer, secrets = [], now = () => Date.now()
if (p.namespace !== prefs.NAMESPACE) return 'ignored:namespace';
if (p.namespace !== prefs.NAMESPACE) return 'ignored:namespace';
return () => (prefs.handleEvent(event) ? 'invalidated' : 'unchanged');
return () => (prefs.handleEvent(event) ? 'invalidated' : 'unchanged');
}
}
+ if (event.event_type === 'vip.membership.changed') {
+ // A membership started, lapsed or was revoked: drop that member's cached badge answers for
+ // that creator now instead of waiting out the cache TTL (VIP convergence).
+ if (event.source !== 'vip') return 'ignored:source';
+ return () => (vipBadges.handleEvent(event) ? 'invalidated' : 'unchanged');
+ }
return 'ignored:type';
return 'ignored:type';
}
}
@@ -123,8 +123,8 @@ t('boot with the consumer on and a stub Events', async () => {
@@ -123,8 +123,8 @@ t('boot with the consumer on and a stub Events', async () => {
t('boot created one subscription per topic, to /internal/events, with the consumer’s secret', async () => {
t('boot created one subscription per topic, to /internal/events, with the consumer’s secret', async () => {
const results = await h.subscriptions.done;
const results = await h.subscriptions.done;
- assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created']]);
- assert.strictEqual(stubEvents.subs.length, 2);
+ assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created'], ['vip.membership.changed', 'created']]);
+ assert.strictEqual(stubEvents.subs.length, 3);
for (const s of stubEvents.subs) {
for (const s of stubEvents.subs) {
assert.strictEqual(s.endpoint, `http://127.0.0.1:${h.port}/internal/events`);
assert.strictEqual(s.endpoint, `http://127.0.0.1:${h.port}/internal/events`);
assert.strictEqual(s.secret, SECRET, 'the first CHAT_EVENTS_SECRET is handed to Events');
assert.strictEqual(s.secret, SECRET, 'the first CHAT_EVENTS_SECRET is handed to Events');
@@ -139,17 +139,17 @@ t('a second boot changes nothing, and never re-enables a subscription an operato
@@ -139,17 +139,17 @@ t('a second boot changes nothing, and never re-enables a subscription an operato
const endpoint = subs.endpointFor(config, h.port);
const endpoint = subs.endpointFor(config, h.port);
const posts = stubEvents.posts;
const posts = stubEvents.posts;
let r = await subs.ensure({ config, endpoint });
let r = await subs.ensure({ config, endpoint });
- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists']);
+ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists']);
assert.strictEqual(stubEvents.posts, posts, 'nothing was created again');
assert.strictEqual(stubEvents.posts, posts, 'nothing was created again');
r = await subs.ensure({ config, endpoint, action: 'disable', topics: ['live.release.deployed'] });
r = await subs.ensure({ config, endpoint, action: 'disable', topics: ['live.release.deployed'] });
assert.deepStrictEqual(r.map((x) => [x.result, x.enabled]), [['disabled', false]]);
assert.deepStrictEqual(r.map((x) => [x.result, x.enabled]), [['disabled', false]]);
r = await subs.ensure({ config, endpoint });
r = await subs.ensure({ config, endpoint });
- assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true]]);
+ assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true], ['vip.membership.changed', 'exists', true]]);
assert.strictEqual(stubEvents.subs[0].enabled, false, 'still disabled');
assert.strictEqual(stubEvents.subs[0].enabled, false, 'still disabled');
r = await subs.ensure({ config, endpoint, action: 'enable', topics: ['live.release.deployed'] });
r = await subs.ensure({ config, endpoint, action: 'enable', topics: ['live.release.deployed'] });
assert.strictEqual(stubEvents.subs[0].enabled, true);
assert.strictEqual(stubEvents.subs[0].enabled, true);
r = await subs.ensure({ config, endpoint, action: 'list' });
r = await subs.ensure({ config, endpoint, action: 'list' });
- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists']);
+ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists']);
});
});
t('signature v2 only: bad signature, stale or future timestamp, v1-only, unsigned → 401', async () => {
t('signature v2 only: bad signature, stale or future timestamp, v1-only, unsigned → 401', async () => {
@@ -344,6 +344,28 @@ t('network.module.updated drops a cached chat.preferences copy when newer; an ol
@@ -344,6 +344,28 @@ t('network.module.updated drops a cached chat.preferences copy when newer; an ol
assert.strictEqual(prefs._cached(ANN).revision, 3);
assert.strictEqual(prefs._cached(ANN).revision, 3);
});
});
+t('vip.membership.changed from VIP drops the cached badge answers at once; from anyone else it is ignored', async () => {
+ const vipBadges = require('../server/vip/badges');
+ const calls = [];
+ const orig = vipBadges.handleEvent;
+ vipBadges.handleEvent = (e) => { calls.push(e.event_id); return true; };
+ try {
+ const ev = (source) => ({
+ event_id: ids.newId('event'), event_type: 'vip.membership.changed', version: 1, source, actor: { type: 'service', id: 'vip' },
+ timestamp: new Date().toISOString(), priority: 'important', visibility: 'internal', subject: { type: 'membership', id: 'mbr_1', revision: 2 },
+ payload: { member: { type: 'user', id: ANN }, creator: { type: 'user', id: ANN }, status: 'canceled' },
+ });
+ const good = ev('vip');
+ let r = await deliver(good);
+ assert.strictEqual(r.status, 200);
+ assert.strictEqual(r.body.outcome, 'invalidated');
+ assert.deepStrictEqual(calls, [good.event_id]);
+ r = await deliver(ev('live'));
+ assert.strictEqual(r.body.outcome, 'ignored:source');
+ assert.strictEqual(calls.length, 1, 'only VIP speaks for memberships');
+ } finally { vipBadges.handleEvent = orig; }
+});
+
t('an event delivered with the secret the subscription holds verifies; /ready reports the consumer', async () => {
t('an event delivered with the secret the subscription holds verifies; /ready reports the consumer', async () => {
const sub = stubEvents.subs.find((s) => s.topic_pattern === 'live.release.deployed');
const sub = stubEvents.subs.find((s) => s.topic_pattern === 'live.release.deployed');
speak('end-to-end next');
speak('end-to-end next');