openvibers/openvibe.chat Commits
VIP member badges: a member's messages in a creator's room carry the creator's VIP badge (a perk of their plan version bound to `chat badge`), resolved through OpenVibe.VIP's entitlement check (product chat) behind the shared product cache
Changes
@@@ -42,6 +42,16 @@ OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000LIVE_MIRROR=0#LIVE_MIRROR_INTERVAL_MS=1000++# OpenVibe.VIP — member badges on chat messages (grant: chat vip.entitlement.check openvibe.vip).++# Cache lifetimes (ms): a badge outlives a membership by at most CHAT_VIP_BADGE_TTL_MS after VIP++# stops granting; a "no" is re-asked after the deny TTL, a failure after the unavailable TTL.++#CHAT_VIP_BADGES=1++#OV_VIP_INTERNAL_URL=http://127.0.0.1:4620++#VIP_TIMEOUT_MS=1500++#CHAT_VIP_BADGE_TTL_MS=60000++#CHAT_VIP_BADGE_DENY_TTL_MS=30000++#CHAT_VIP_BADGE_UNAVAILABLE_TTL_MS=5000++# OpenVibe.Events — the outbox relays only when set (otherwise events wait in events_outbox)#EVENTS_URL=http://127.0.0.1:4300#EVENTS_RELAY_INTERVAL_MS=5000@@@ -64,6 +64,7 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th| Live's own chat pushes and writes (AI viewers, relays, donations, `/api/mod`, recaps, calls) | **Live → Chat** | `POST /internal/live/calls` (`chat.live_bridge.write`), presence `GET /internal/live/presence` (`chat.presence.read`) || Identity | **OpenVibe.Network** | user tokens are resolved by Live (its account links); service tokens from `/oauth/token` || Events | **OpenVibe.Events** | `events_outbox` → `POST /api/v1/events` when `EVENTS_URL` is set (`events.event.publish`) |++| Member badges | **OpenVibe.VIP** (Billing holds the entitlement) | `server/vip/badges.js` → `POST /api/v1/entitlements/check` with `product: 'chat'` (`vip.entitlement.check`), behind the shared product cache |### Data@@@ -96,6 +97,34 @@ in one transaction with its outbox row, and the relay publishes in outbox order,always follows the message it removes. Messages deleted before this existed are redacted byOpenVibe.Events' `scripts/redact-backfill.js`.++### VIP member badges++++A member's messages in a creator's room (stream or offline channel chat) carry that creator's VIP++badge: a perk of the member's plan **version** with a `chat badge` product binding (VIP's network++perk `subscriber-badge` has one), unless the member turned it off in VIP (`show_badge`). The frame++gets `vip_badge: { creator, perk, name, badge, label? }` and the row's `metadata` keeps it, so history++shows it. `badge` is a short id (`subscriber`, else `member`) and `label` plain text: binding config is++written by creators, so nothing else of it passes and clients render both as text.++++- **Never blocks sending.** The badge comes from the cache only (`peekEntitlement`); the lookup starts++ when a member joins a room, so their first line normally has it. On a miss the message goes out++ without a badge and, if the lookup then finds one, a `chat_vip_badge` frame `{ id, vip_badge,++ stream_id, channel_user_id }` follows to the same rooms (clients attach it by message id, as with++ `chat_translation`) and the row's metadata is updated.++- **Fails closed.** VIP unreachable or slow, no client secret, a refused grant, a malformed answer, an++ inactive or `unknown` entitlement → no badge. `CHAT_VIP_BADGES=0` turns lookups off.++- **Convergence bound.** Chat has no Events inbox, so when a membership ends (Billing's++ `billing.entitlement.changed` → VIP's `vip.membership.changed`) the badge stops at most++ `CHAT_VIP_BADGE_TTL_MS` (60 s) after VIP stops granting, never past the entitlement's `expires_at`;++ end to end, add VIP's own bound (seconds with events, at most `VIP_PROJECTION_MAX_AGE_MS` without —++ OpenVibe.VIP README, "The product cache and the convergence bound"). A new member's badge appears++ within `CHAT_VIP_BADGE_DENY_TTL_MS` (30 s). `badges.handleEvent()` drops a pair at once, for when++ the events are routed to Chat. `test/vip-badge.test.js` drives all of this against a stub VIP with an++ injected clock.++- Needs the Network grant `chat vip.entitlement.check openvibe.vip`; without it VIP answers 403 and++ nobody gets a badge. The client is vendored from OpenVibe.VIP (`server/vip/vip-client.js`, copied++ at VIP 2accbeb) until VIP publishes a tag to pin.++## Running it```bash@@@ -170,6 +199,7 @@ Events: `chat.message.created`, `chat.message.deleted`, `chat.dm.created`, `chat- OpenVibe.Live (until the rest of chat's data moves)- OpenVibe.Events- OpenVibe.Media++- OpenVibe.VIP (member badges; optional — without it nobody has a badge)- OpenVibe.Contracts## Acceptance (must be true before "done")@@@ -9,10 +9,11 @@"domain": "openvibe.live (/ws/chat, /api/chat/, /api/dm/, /api/tts/, /api/sounds via an nginx include in openvibe.live's vhost); openvibe.chat is still the OpenVibe.Sites placeholder","contracts": "openvibe-contracts v0.13.0","capabilities": ["chat.live_bridge.write", "chat.presence.read"],-- "capabilitiesUsed": ["live.chat_context.read", "live.chat_effects.write", "live.chat_mirror.write", "events.event.publish"],++ "capabilitiesUsed": ["live.chat_context.read", "live.chat_effects.write", "live.chat_mirror.write", "events.event.publish", "vip.entitlement.check"],"capabilitiesRegistered": true,"events": ["chat.message.created", "chat.message.deleted", "chat.dm.created", "chat.moderation.action"],"liveAuthorityFlag": "CHAT_AUTHORITY=chat is set in production live.env since the cutover (2026-09-23 02:03 UTC); Live's chat tables are a read mirror Chat keeps current",++ "vip": "Member badges from OpenVibe.VIP (server/vip/badges.js: entitlements/check with product chat, cached 60 s, never blocks sending, fails closed). In code only, not deployed; needs the Network grant chat vip.entitlement.check openvibe.vip and Live's chat UI to render vip_badge / chat_vip_badge","deployed": true,"deployment": "openvibe-ovh /opt/openvibe.chat, unit openvibe-chat, 127.0.0.1:4400 (loopback); public only through openvibe.live's nginx include for the chat paths; release 45f2b1f; EVENTS_URL set (chat.message.created reaches OpenVibe.Events); restore drill passed 2026-09-23","updated": "2026-09-23"@@@ -69,6 +69,8 @@ Live data maintained by `live-context` (never authority), `events_outbox`, `livea default). Without `chat.message.send`, Chat refuses the bridge ops that send a message (AIviewer, relay and donation lines, deploy notices). Add them to`DEFAULT_GRANTS` in `server/identity/principals.js` so every boot keeps them.++ VIP member badges (after this cutover, any time): `chat vip.entitlement.check openvibe.vip`;++ without it no message carries a badge (README, "VIP member badges").3. **Live.** Apply `docs/live-patch.diff` to Live `main` (`git apply docs/live-patch.diff`), run itstests (`node test/chat-context.test.js`, `npm test`), deploy it **without** `CHAT_AUTHORITY` —nothing changes for users; the new internal routes answer only service tokens and every effect@@@ -30,6 +30,7 @@ const ttsEngine = require('./tts-engine');const soundboard = require('./soundboard-service');const audioQueue = require('./audio-queue');const dm = require('./dm');++const vipBadges = require('../vip/badges');// Cosmetics and tags are Live's (monetization/cosmetics, game/tags); read through live-context.const cosmetics = { getCosmeticProfile: (userId) => ctx.getCosmeticProfile(userId) };const tags = { getTagProfile: (userId) => ctx.getTagProfile(userId) };@@@ -434,6 +435,8 @@ class ChatServer {}client.channelUserId = channelUserId;client._modStream = null;++ // Look up this member's VIP badge for the room now, so their first message has it.++ if (client.user && channelUserId) { try { vipBadges.warm(this._subjectOfUser(client.user), db.subjectFor(channelUserId)); } catch { /* */ } }// A TTS/sound queue held since a restart plays once its room has listeners again.audioQueue.roomJoined(audioQueue.roomKey({ streamId: client.streamId, channelUserId: client.streamId ? null : channelUserId }));// Update viewer counts for old and new streams@@@ -778,6 +781,18 @@ class ChatServer {} catch { /* non-critical */ }}++ // The creator's VIP member badge (a perk bound to `chat badge`). From the cache only: the++ // message never waits on VIP, and a miss is sent without a badge while the lookup runs; a++ // badge found then follows as a chat_vip_badge frame for this message (below).++ let vipBadgePending = null;++ if (client.user?.id && client.channelUserId) {++ try {++ const r = vipBadges.forMessage(this._subjectOfUser(client.user), db.subjectFor(client.channelUserId));++ if (r.badge) chatMsg.vip_badge = r.badge;++ else vipBadgePending = r.pending;++ } catch { /* no badge */ }++ }++// Save to databasetry {const result = db.saveChatMessage({@@@ -791,6 +806,7 @@ class ChatServer {is_global: !client.streamId && !client.channelUserId,reply_to_id: replyToId,auto_delete_at: autoDeleteAt,++ metadata: chatMsg.vip_badge ? { vip_badge: chatMsg.vip_badge } : undefined,});if (result.lastInsertRowid) chatMsg.id = Number(result.lastInsertRowid);} catch { /* non-critical */ }@@@ -841,6 +857,7 @@ class ChatServer {// Auto-translate (async): foreign → English for everyone, English → the channel's// language for a non-English streamer. Lands as a follow-up 'chat_translation'.this._maybeTranslate(chatMsg, client.channelUserId, client.streamId);++ if (vipBadgePending) this._followVipBadge(vipBadgePending, chatMsg, client.channelUserId, client.streamId);}if (client.streamId) {@@@ -2132,6 +2149,28 @@ class ChatServer {try { return ctx.channelLanguage(channelUserId); } catch { return 'en'; }}++ /** A signed-in user's Network subject (the projection's subject_id, else the ctx_users row). */++ _subjectOfUser(user) {++ if (!user) return null;++ return user.subject_id || db.subjectFor(user.id) || null;++ }++++ /**++ * A VIP badge that was not cached when its message went out: when the lookup finds one, push++ * it to the same rooms as a 'chat_vip_badge' event (clients attach it to the message by id) and++ * keep it in the row's metadata so history shows it. Fire-and-forget; never throws.++ */++ _followVipBadge(pending, chatMsg, channelUserId, streamId) {++ Promise.resolve(pending).then((badge) => {++ if (!badge || !chatMsg || !chatMsg.id) return;++ const evt = { type: 'chat_vip_badge', id: chatMsg.id, vip_badge: badge, stream_id: streamId || null, channel_user_id: channelUserId || null, timestamp: new Date().toISOString() };++ this.broadcastToChannelRoom(channelUserId, streamId, evt);++ if (streamId) this.forwardToGlobal(streamId, evt);++ else this.forwardToGlobalByChannel(channelUserId, evt);++ try { db.mergeChatMessageMetadata(chatMsg.id, { vip_badge: badge }); } catch { /* */ }++ }).catch(() => { /* no badge */ });++ }++/*** Translate a just-broadcast chat line and push the translation to the same rooms as a* 'chat_translation' event (clients attach it under the message by id). Persisted into@@@ -64,6 +64,22 @@ module.exports = {syncStaleMs: int(process.env.LIVE_SYNC_STALE_MS, 60_000),},++ // OpenVibe.VIP — the member badge on chat messages (a creator's perk bound to `chat badge`).++ // Asked through a cache only (server/vip/badges.js): sending never waits on VIP, and any doubt++ // (VIP down, no token, no answer yet) is "no badge". Needs Chat's client secret and the Network++ // grant vip.entitlement.check on openvibe.vip.++ vip: {++ enabled: bool(process.env.CHAT_VIP_BADGES, true),++ internalUrl: strip(process.env.OV_VIP_INTERNAL_URL || 'http://127.0.0.1:4620'),++ timeoutMs: int(process.env.VIP_TIMEOUT_MS, 1500),++ // How long a badge outlives the membership at most (VIP applied the change; Chat has no++ // Events inbox): the convergence bound. A "no" is re-asked after denyTtlMs, a failure after++ // unavailableTtlMs.++ ttlMs: int(process.env.CHAT_VIP_BADGE_TTL_MS, 60_000),++ denyTtlMs: int(process.env.CHAT_VIP_BADGE_DENY_TTL_MS, 30_000),++ unavailableTtlMs: int(process.env.CHAT_VIP_BADGE_UNAVAILABLE_TTL_MS, 5_000),++ },++// OpenVibe.Events — the outbox relays only when EVENTS_URL is set.events: {url: strip(process.env.EVENTS_URL || ''),nnew file mode 100644@@@ -0,0 +1,106 @@++/**++ * The VIP member badge on chat messages.++ *++ * A member's messages in a creator's room carry that creator's VIP badge: a perk of the member's plan++ * version with a `chat badge` product binding (the network perk `subscriber-badge` has one), unless++ * the member turned the badge off in VIP (show_badge). Resolved through VIP's entitlement check++ * (vip.entitlement.check, product 'chat') behind the shared product cache (openvibe-vip/client++ * createVipCache):++ *++ * - sending NEVER waits on VIP: forMessage() reads the cache synchronously. A miss sends the++ * message without a badge and starts the lookup; when it finds a badge, chat-server pushes a++ * `chat_vip_badge` frame for that message id (and stores it in the row's metadata);++ * - it fails closed: VIP down, no token, a refused grant, a malformed answer → no badge;++ * - convergence: Chat has no Events inbox, so a badge outlives the membership by at most++ * config.vip.ttlMs (60 s) after VIP stopped granting (VIP README: "The product cache and the++ * convergence bound"). handleEvent() drops a pair at once if events are ever routed here.++ *++ * The badge is plain data for Live's chat UI to render as text: { creator, perk, name, badge, label }.++ * Binding config comes from creators, so only a short id (`badge`) and a short text label pass.++ */++'use strict';++++const config = require('../config');++const serviceAuth = require('../net/service-auth');++const { createVipClient, createVipCache } = require('./vip-client');++++const AUDIENCE = 'openvibe.vip';++const SCOPE = 'vip.entitlement.check';++const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++++let cache = null;++let clock = () => Date.now();++let fetchImpl = null;++++function enabled() {++ return !!(config.vip.enabled && config.vip.internalUrl && config.oauth.clientSecret);++}++++function build() {++ const tokenClient = {++ authHeaders: () => serviceAuth.headers(AUDIENCE, SCOPE),++ invalidate: () => serviceAuth.invalidate(AUDIENCE, SCOPE),++ };++ const vip = createVipClient({++ baseUrl: config.vip.internalUrl, tokenClient, timeoutMs: config.vip.timeoutMs,++ ...(fetchImpl ? { fetch: fetchImpl } : {}),++ });++ return createVipCache({++ vip, ttlMs: config.vip.ttlMs, denyTtlMs: config.vip.denyTtlMs, unavailableTtlMs: config.vip.unavailableTtlMs, now: () => clock(),++ });++}++function theCache() { if (!cache) cache = build(); return cache; }++++const text = (v, max) => (typeof v === 'string' ? v.replace(/[\u0000-\u001f\u007f<>]/g, '').trim().slice(0, max) : '');++++/** An entitlement answer (product 'chat') → the badge to show, or null. */++function badgeOf(answer, creator) {++ if (!answer || answer.active !== true || answer.status !== 'active') return null;++ if (answer.preferences && answer.preferences.show_badge === false) return null;++ for (const perk of Array.isArray(answer.product_perks) ? answer.product_perks : []) {++ const b = (perk.bindings || []).find((x) => x && x.binding === 'badge');++ if (!b) continue;++ const cfg = b.config && typeof b.config === 'object' ? b.config : {};++ const id = typeof cfg.badge === 'string' && /^[a-z0-9_-]{1,32}$/.test(cfg.badge) ? cfg.badge : 'member';++ const out = { creator, perk: text(perk.key, 48), name: text(perk.name, 80), badge: id };++ const label = text(cfg.label, 32);++ if (label) out.label = label;++ return out;++ }++ return null;++}++++const pairOf = (memberSubject, creatorSubject) => (++ USER_SUBJECT.test(String(memberSubject || '')) && USER_SUBJECT.test(String(creatorSubject || '')) && memberSubject !== creatorSubject++ ? { subject: memberSubject, creator: creatorSubject, product: 'chat' } : null);++++/**++ * For a message being sent: { badge } from the cache, or { badge: null, pending } — a promise of++ * the badge (or null) that never rejects — when the cache has no answer yet. Never waits.++ */++function forMessage(memberSubject, creatorSubject) {++ const args = enabled() ? pairOf(memberSubject, creatorSubject) : null;++ if (!args) return { badge: null, pending: null };++ const c = theCache();++ const hit = c.peekEntitlement(args);++ if (hit !== undefined) return { badge: badgeOf(hit, creatorSubject), pending: null };++ return { badge: null, pending: c.entitlement(args).then((a) => badgeOf(a, creatorSubject), () => null) };++}++++/** Look the pair up ahead of the first message (a join). Fire-and-forget. */++function warm(memberSubject, creatorSubject) {++ const args = enabled() ? pairOf(memberSubject, creatorSubject) : null;++ if (args) theCache().entitlement(args).catch(() => {});++}++++/** vip.membership.changed / billing.entitlement.changed / … → drop that pair now. */++function handleEvent(envelope) { return enabled() ? theCache().handleEvent(envelope) : false; }++++/** Tests: an injected clock and fetch; rebuilds the cache. */++function _configure({ now, fetch } = {}) {++ if (now) clock = now;++ if (fetch !== undefined) fetchImpl = fetch;++ cache = null;++}++++module.exports = { forMessage, warm, handleEvent, badgeOf, enabled, _configure, get bounds() { return theCache().bounds; } };nnew file mode 100644@@@ -0,0 +1,235 @@++// Copied verbatim from OpenVibe.VIP client/vip-client.js at 2accbeb (openvibe-vip/client). Replace with the++// pinned package once VIP publishes a tag; do not edit here — change it in OpenVibe.VIP.++'use strict';++++/**++ * openvibe-vip/client — the consumer seam for products that honour VIP memberships (Blog, Wiki,++ * Chat, Live, Community, …). No dependencies; Node 18+ (global fetch) or any fetch you pass.++ *++ * const { createVipClient } = require('openvibe-vip/client');++ * const tokens = contracts.serviceAuth.createTokenClient({ tokenUrl, clientId, clientSecret,++ * audience: 'openvibe.vip', scope: 'vip.resource.policy.evaluate vip.entitlement.check' });++ * const vip = createVipClient({ baseUrl: 'http://127.0.0.1:4620', tokenClient: tokens });++ *++ * const d = await vip.evaluate({ subject: viewerSubjectOrNull, resource: { service: 'blog', type: 'post', id: '42' }, owner: authorSubject,++ * fallback: { requirement: 'member', binding: 'blog:gated_post' } }); // the product's default gate++ * if (!d.allow) return renderTeaser(d.reason); // FAILS CLOSED++ *++ * const e = await vip.checkEntitlement({ subject: 'usr_…', creator: 'usr_…', product: 'chat' }); // e.active, e.product_perks++ * await vip.isMember('usr_…', 'usr_…'); // boolean, false on any doubt++ *++ * Every failure — VIP unreachable, a timeout, a refused token, a malformed answer — is a denial++ * ({ allow: false, reason: 'vip_unavailable' } / { status: 'unknown', active: false }), never an++ * exception and never a "yes". Pass mode: 'authoritative' for sensitive content: VIP then asks++ * OpenVibe.Billing directly instead of its projection.++ *++ * Put createVipCache (below) in front of it: it holds the convergence bound every product shares.++ */++++function createVipClient({ baseUrl = 'http://127.0.0.1:4620', tokenClient = null, getToken = null, fetch: fetchImpl = globalThis.fetch, timeoutMs = 2000, log = null } = {}) {++ if (typeof fetchImpl !== 'function') throw new TypeError('createVipClient: a fetch implementation is required');++ const base = String(baseUrl).replace(/\/+$/, '');++++ async function headers() {++ if (tokenClient && typeof tokenClient.authHeaders === 'function') return tokenClient.authHeaders();++ if (typeof getToken === 'function') return { Authorization: `Bearer ${await getToken()}` };++ throw new Error('createVipClient: tokenClient or getToken is required');++ }++++ async function post(path, body, retried = false) {++ const res = await fetchImpl(`${base}${path}`, {++ method: 'POST',++ headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...(await headers()) },++ body: JSON.stringify(body),++ signal: AbortSignal.timeout(timeoutMs),++ });++ if (res.status === 401 && !retried && tokenClient && typeof tokenClient.invalidate === 'function') { tokenClient.invalidate(); return post(path, body, true); }++ const data = await res.json().catch(() => null);++ if (!res.ok || !data || typeof data !== 'object') throw new Error(`VIP ${res.status}: ${(data && (data.code || data.detail)) || 'bad response'}`);++ return data;++ }++ const note = (what, err) => { if (log && log.warn) log.warn(`[vip-client] ${what} failed closed: ${err.message}`); };++++ /** May `subject` (a usr_ id, a SubjectRef, or null for a signed-out viewer) see `resource`? */++ async function evaluate({ subject = null, resource, owner = null, ruleId, mode, fallback } = {}) {++ try {++ const d = await post('/api/v1/policies/evaluate', { subject: subject || null, resource, owner, rule_id: ruleId, mode, fallback });++ return { ...d, allow: d.allow === true, reason: d.reason || (d.allow === true ? 'member' : 'denied') };++ } catch (err) {++ note('evaluate', err);++ return { allow: false, reason: 'vip_unavailable', error: err.message };++ }++ }++++ /** The entitlement of `subject` to `creator` (usr_ ids or SubjectRefs). */++ async function checkEntitlement({ subject, creator, mode, product } = {}) {++ try {++ const e = await post('/api/v1/entitlements/check', { subject, creator, mode, product });++ return { ...e, active: e.active === true && e.status === 'active' };++ } catch (err) {++ note('checkEntitlement', err);++ return { status: 'unknown', active: false, reason: 'vip_unavailable', error: err.message };++ }++ }++++ const isMember = async (subject, creator, opts = {}) => (await checkEntitlement({ subject, creator, mode: opts.mode })).active;++++ return { evaluate, checkEntitlement, isMember };++}++++/**++ * createVipCache({ vip, ttlMs, denyTtlMs, unavailableTtlMs, maxEntries, now }) — the product-side++ * cache every consumer puts in front of the client, with ONE convergence rule:++ *++ * A cached "yes" lives at most ttlMs (and never past the entitlement's expires_at or VIP's++ * valid_until). So once VIP stops granting — it has applied Billing's billing.entitlement.changed++ * and emitted vip.membership.changed — the product stops granting within ttlMs. handleEvent(envelope)++ * with either event (or billing.subscription.canceled / billing.transaction.reversed) drops that++ * member's entries for that creator at once, so a product that receives the events converges++ * immediately. A "no" lives denyTtlMs (a new member waits at most that long); a failure++ * (vip_unavailable / unknown) lives unavailableTtlMs, and a cached "yes" is never extended by a++ * failure: when VIP is down, entries expire into denials (fail closed).++ *++ * await cache.entitlement({ subject, creator, product }) → the checkEntitlement answer (cached)++ * cache.peekEntitlement({ subject, creator, product }) → the cached answer or undefined; never waits++ * (a miss starts a fetch in the background)++ * await cache.evaluate({ subject, resource, owner, fallback, mode }) → the evaluate answer (cached;++ * mode 'authoritative' is never cached)++ * cache.invalidate({ member, creator }), cache.handleEvent(envelope), cache.clear()++ */++function createVipCache({ vip, ttlMs = 30_000, denyTtlMs = 10_000, unavailableTtlMs = 2_000, maxEntries = 5000, now = () => Date.now() } = {}) {++ if (!vip || typeof vip.evaluate !== 'function' || typeof vip.checkEntitlement !== 'function') throw new TypeError('createVipCache: a VIP client is required');++ const entries = new Map(); // key → { value, expires, member, creator }++ const inflight = new Map(); // key → Promise++ let generation = 0; // bumped by clear(); an answer fetched before a clear is not stored++ const pairGen = new Map(); // `${member}|${creator}` → generation of its last invalidation++++ const idOf = (v) => (v && typeof v === 'object' ? (v.id != null ? String(v.id) : null) : (v != null && v !== '' ? String(v) : null));++ const pairKey = (member, creator) => `${member || ''}|${creator || ''}`;++++ function lifetime(value, allowed) {++ const t = now();++ if (value && (value.reason === 'vip_unavailable' || value.status === 'unknown' || value.reason === 'entitlement_unknown')) return unavailableTtlMs;++ if (!allowed) return denyTtlMs;++ let life = ttlMs;++ const ent = value.entitlement !== undefined ? value.entitlement : value;++ for (const at of [ent && ent.expires_at, ent && ent.valid_until]) {++ const ms = at ? Date.parse(at) : NaN;++ if (Number.isFinite(ms)) life = Math.min(life, ms - t);++ }++ if (ent && ent.stale) life = Math.min(life, unavailableTtlMs);++ return Math.max(0, life);++ }++++ function store(key, value, allowed, member, creator, startedGen, startedPair) {++ if (startedGen !== generation || (pairGen.get(pairKey(member, creator)) || 0) !== startedPair) return; // invalidated meanwhile++ const life = lifetime(value, allowed);++ if (life <= 0) { entries.delete(key); return; }++ if (entries.size >= maxEntries) {++ const t = now();++ for (const [k, e] of entries) if (e.expires <= t) entries.delete(k);++ while (entries.size >= maxEntries) entries.delete(entries.keys().next().value);++ }++ entries.set(key, { value, expires: now() + life, member, creator });++ }++++ function fresh(key) {++ const e = entries.get(key);++ if (!e) return undefined;++ if (e.expires <= now()) { entries.delete(key); return undefined; }++ return e.value;++ }++++ function load(key, member, creator, fetcher, allowedOf) {++ const hit = fresh(key);++ if (hit !== undefined) return Promise.resolve(hit);++ if (inflight.has(key)) return inflight.get(key);++ const g = generation;++ const pg = pairGen.get(pairKey(member, creator)) || 0;++ const p = (async () => {++ try {++ const value = await fetcher();++ store(key, value, allowedOf(value), member, creator, g, pg);++ return value;++ } finally { inflight.delete(key); }++ })();++ inflight.set(key, p);++ return p;++ }++++ function entitlement({ subject, creator, product = null, mode } = {}) {++ const member = idOf(subject);++ const owner = idOf(creator);++ if (!member || !owner) return Promise.resolve({ status: 'inactive', active: false, reason: 'not_signed_in' });++ if (mode === 'authoritative') return vip.checkEntitlement({ subject: member, creator: owner, mode, product: product || undefined });++ const key = `e|${member}|${owner}|${product || ''}`;++ return load(key, member, owner, () => vip.checkEntitlement({ subject: member, creator: owner, product: product || undefined }), (v) => v.active === true);++ }++++ function peekEntitlement(args = {}) {++ const member = idOf(args.subject);++ const owner = idOf(args.creator);++ if (!member || !owner) return undefined;++ const key = `e|${member}|${owner}|${args.product || ''}`;++ const hit = fresh(key);++ if (hit === undefined) entitlement(args).catch(() => {});++ return hit;++ }++++ function evaluate({ subject = null, resource, owner, fallback, mode } = {}) {++ const member = idOf(subject);++ const creator = idOf(owner);++ if (mode === 'authoritative' || !resource || typeof resource !== 'object') return vip.evaluate({ subject: member, resource, owner: creator, fallback, mode });++ if (!member) return Promise.resolve({ allow: false, reason: 'not_signed_in' });++ const fb = fallback == null ? '' : JSON.stringify(fallback);++ const key = `p|${member}|${creator || ''}|${resource.service}|${resource.type}|${resource.id}|${fb}`;++ return load(key, member, creator, () => vip.evaluate({ subject: member, resource, owner: creator, fallback }), (v) => v.allow === true);++ }++++ /** Drop what the cache knows about (member, creator); either may be omitted to drop more. */++ function invalidate({ member = null, creator = null } = {}) {++ const m = idOf(member);++ const c = idOf(creator);++ if (m && c) pairGen.set(pairKey(m, c), (pairGen.get(pairKey(m, c)) || 0) + 1);++ else generation++;++ let dropped = 0;++ for (const [k, e] of entries) {++ if ((!m || e.member === m) && (!c || e.creator === c)) { entries.delete(k); dropped++; }++ }++ return dropped;++ }++++ /**++ * An OpenVibe.Events envelope about a membership → invalidate that pair. Returns true when the++ * event was understood. vip.membership.changed { member, creator }; billing.entitlement.changed++ * { subject, streamer }; billing.subscription.canceled { subscription: { subscriber, streamer } };++ * billing.transaction.reversed { to_subject (member), from_subject (creator) }.++ */++ function handleEvent(envelope) {++ const e = envelope && envelope.event && envelope.event.event_type ? envelope.event : envelope;++ const p = (e && e.payload) || {};++ let member = null;++ let creator = null;++ switch (e && e.event_type) {++ case 'vip.membership.changed': member = idOf(p.member); creator = idOf(p.creator); break;++ case 'billing.entitlement.changed': member = idOf(p.subject); creator = idOf(p.streamer); break;++ case 'billing.subscription.canceled': member = idOf(p.subscription && p.subscription.subscriber); creator = idOf(p.subscription && p.subscription.streamer); break;++ case 'billing.transaction.reversed': member = idOf(p.to_subject); creator = idOf(p.from_subject); break;++ default: return false;++ }++ if (!member || !creator) return false;++ invalidate({ member, creator });++ return true;++ }++++ function clear() { generation++; entries.clear(); }++++ return {++ entitlement, peekEntitlement, evaluate, invalidate, handleEvent, clear,++ get size() { return entries.size; },++ /** The convergence bound, for docs and /ready: how long a cached "yes" can outlive VIP's "no". */++ bounds: Object.freeze({ grantMs: ttlMs, denyMs: denyTtlMs, unavailableMs: unavailableTtlMs }),++ };++}++++module.exports = { createVipClient, createVipCache };nnew file mode 100644@@@ -0,0 +1,213 @@++'use strict';++/**++ * The VIP member badge (roadmap Wave 10): a member's messages in a creator's room carry the++ * creator's VIP badge — a perk of the member's plan version bound to `chat badge` — resolved through++ * VIP's entitlement check (product 'chat') behind the shared product cache.++ *++ * - sending never waits on VIP (a slow VIP: the message goes out at once, the badge follows as++ * a chat_vip_badge frame for that id and lands in the row's metadata)++ * - fail closed: VIP down, not a member, badge turned off, no chat binding → no badge++ * - convergence: once VIP stops granting, the badge stops within config.vip.ttlMs (60 s; Chat has++ * no Events inbox), and at once when handleEvent gets vip.membership.changed++ *++ * A stub OpenVibe.VIP answers /api/v1/entitlements/check from a table and checks the service token++ * (audience openvibe.vip, vip.entitlement.check).++ */++const assert = require('assert');++const http = require('http');++const { serviceAuth } = require('openvibe-contracts');++const { boot, suite } = require('./helpers');++++const t = suite('vip-badge');++const CREATOR = 'usr_01J9CREAT0R0000000000000AA';++const MEMBER = 'usr_01J9MEMBER0000000000000AAA';++const SLOWPOKE = 'usr_01J9S10WP0KE000000000000AA';++const STRANGER = 'usr_01J9STRANGER00000000000AAA';++const SHY = 'usr_01J9SHY0000000000000000AAA';++const PLAIN = 'usr_01J9P1A1N00000000000000AAA';++++let h, badges, vip, streamer, member, slowpoke, stranger, shy, plain, streamId;++const clock = { t: Date.now(), now() { return clock.t; } };++++/** An entitlement answer as VIP gives it with product 'chat'. */++const active = (perks, { showBadge = true } = {}) => ({++ status: 'active', active: true, expires_at: new Date(Date.now() + 30 * 86400e3).toISOString(), valid_until: new Date(Date.now() + 30 * 86400e3).toISOString(),++ source: 'projection', stale: false, product: 'chat', product_perks: perks, preferences: { show_badge: showBadge, listed: false },++});++const subscriberPerk = { key: 'subscriber-badge', name: 'Subscriber badge', kind: 'badge', scope: 'network', bindings: [{ binding: 'badge', config: { badge: 'subscriber' } }] };++const inactive = { status: 'inactive', active: false, expires_at: null, product: 'chat', product_perks: [], preferences: { show_badge: true, listed: false } };++++async function startVip(keyOf) {++ const state = { answers: new Map(), calls: [], down: false, hold: new Map(), badTokens: 0 };++ const server = http.createServer((req, res) => {++ let raw = '';++ req.on('data', (c) => { raw += c; });++ req.on('end', async () => {++ if (state.down) { req.socket.destroy(); return; }++ const send = (status, body) => { res.statusCode = status; res.setHeader('Content-Type', 'application/json'); res.end(JSON.stringify(body)); };++ const v = serviceAuth.verifyServiceToken(String(req.headers.authorization || '').slice(7), { publicKey: keyOf(), issuer: 'https://openvibe.network', audience: 'openvibe.vip' });++ if (!v.ok) { state.badTokens++; return send(401, { code: v.code }); }++ if (!(v.claims.cap || []).includes('vip.entitlement.check')) return send(403, { code: 'capability.denied' });++ if (req.url !== '/api/v1/entitlements/check' || req.method !== 'POST') return send(404, { code: 'not_found' });++ const b = JSON.parse(raw || '{}');++ state.calls.push(b);++ const hold = state.hold.get(b.subject);++ if (hold) await hold;++ send(200, { member: { type: 'user', id: b.subject }, creator: { type: 'user', id: b.creator }, ...(state.answers.get(`${b.subject}|${b.creator}`) || inactive) });++ });++ });++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ return { state, server, url: `http://127.0.0.1:${server.address().port}` };++}++++const callsFor = (subject) => vip.state.calls.filter((c) => c.subject === subject).length;++async function until(pred, ms = 3000) {++ const end = Date.now() + ms;++ while (!pred()) { if (Date.now() > end) throw new Error('timed out'); await new Promise((r) => setTimeout(r, 10)); }++}++async function joinRoom(user, ip) {++ const ws = await h.ws({ ip, token: user.token, stream: streamId });++ ws.sendJson({ type: 'join', streamId, token: user.token });++ await ws.next((m) => m.type === 'auth');++ return ws;++}++let seq = 0;++async function say(ws, watcher) {++ const text = `vip line ${++seq}`;++ const started = Date.now();++ ws.sendJson({ type: 'chat', message: text });++ const got = await watcher.next((m) => m.type === 'chat' && m.message === text);++ return { msg: got, ms: Date.now() - started };++}++++t('boot with a stub VIP', async () => {++ vip = await startVip(() => h.keys.publicKey);++ h = await boot({ env: { OV_VIP_INTERNAL_URL: vip.url, CHAT_VIP_BADGE_TTL_MS: '60000', CHAT_VIP_BADGE_DENY_TTL_MS: '30000', CHAT_VIP_BADGE_UNAVAILABLE_TTL_MS: '5000' } });++ badges = require('../server/vip/badges');++ badges._configure({ now: () => clock.now() });++ assert.ok(badges.enabled());++ h.chatServer.DEFAULT_RATE_LIMIT_MS = 0; // many lines per test, same senders++++ streamer = h.addUser('streamer', { role: 'streamer', subject: CREATOR });++ member = h.addUser('member', { subject: MEMBER });++ slowpoke = h.addUser('slowpoke', { subject: SLOWPOKE });++ stranger = h.addUser('stranger', { subject: STRANGER });++ shy = h.addUser('shy', { subject: SHY });++ plain = h.addUser('plain', { subject: PLAIN });++ const ch = h.addChannel(streamer.id);++ streamId = h.addStream(streamer.id, ch);++ vip.state.answers.set(`${MEMBER}|${CREATOR}`, active([subscriberPerk]));++ vip.state.answers.set(`${SLOWPOKE}|${CREATOR}`, active([{ ...subscriberPerk, bindings: [{ binding: 'badge', config: { badge: 'Bad Id!', label: '<b>Crew</b>\u0007' } }] }]));++ vip.state.answers.set(`${SHY}|${CREATOR}`, active([subscriberPerk], { showBadge: false }));++ vip.state.answers.set(`${PLAIN}|${CREATOR}`, active([{ key: 'backstage', name: 'Backstage', kind: 'gated_content', scope: 'creator', bindings: [{ binding: 'gated_post', config: {} }] }]));++});++++let memberWs, watcher;++t('a member\'s message carries the creator\'s badge (looked up at join, so the first line has it), persisted in metadata', async () => {++ watcher = await joinRoom(stranger, '198.51.100.41');++ memberWs = await joinRoom(member, '198.51.100.40');++ await until(() => callsFor(MEMBER) >= 1);++ const call = vip.state.calls.find((c) => c.subject === MEMBER);++ assert.deepStrictEqual([call.creator, call.product], [CREATOR, 'chat']);++ await new Promise((r) => setTimeout(r, 50));++ const { msg } = await say(memberWs, watcher);++ assert.deepStrictEqual(msg.vip_badge, { creator: CREATOR, perk: 'subscriber-badge', name: 'Subscriber badge', badge: 'subscriber' });++ const row = h.db.getChatMessageById(msg.id);++ assert.deepStrictEqual(JSON.parse(row.metadata).vip_badge, msg.vip_badge);++ assert.ok(h.tokenRequests.some((r) => r.audience === 'openvibe.vip' && r.scope === 'vip.entitlement.check'), 'Chat asked the Network for a VIP token');++ const before = callsFor(MEMBER);++ await say(memberWs, watcher);++ assert.strictEqual(callsFor(MEMBER), before, 'the second message comes from the cache');++});++++t('fail closed: not a member, badge turned off, no chat binding → no badge; the creator\'s own lines never ask VIP', async () => {++ for (const [u, ip, subject] of [[stranger, null, STRANGER], [shy, '198.51.100.43', SHY], [plain, '198.51.100.44', PLAIN]]) {++ const ws = ip ? await joinRoom(u, ip) : watcher;++ await until(() => callsFor(subject) >= 1);++ await new Promise((r) => setTimeout(r, 30));++ const { msg } = await say(ws, memberWs);++ assert.strictEqual(msg.vip_badge, undefined, `${u.username}: no badge`);++ }++ const own = await joinRoom(streamer, '198.51.100.45');++ const { msg } = await say(own, watcher);++ assert.strictEqual(msg.vip_badge, undefined);++ assert.strictEqual(vip.state.calls.filter((c) => c.subject === CREATOR).length, 0);++});++++t('sending never waits on VIP: a slow lookup sends at once, the badge follows as chat_vip_badge and lands in metadata', async () => {++ let release;++ vip.state.hold.set(SLOWPOKE, new Promise((r) => { release = r; }));++ const ws = await joinRoom(slowpoke, '198.51.100.46');++ await until(() => callsFor(SLOWPOKE) >= 1);++ const { msg, ms } = await say(ws, watcher);++ assert.strictEqual(msg.vip_badge, undefined, 'sent before VIP answered');++ assert.ok(ms < 1000, `delivered in ${ms} ms while VIP was holding`);++ release();++ vip.state.hold.delete(SLOWPOKE);++ const follow = await watcher.next((m) => m.type === 'chat_vip_badge' && m.id === msg.id);++ // Creator-written binding config is cleaned: an unknown badge id becomes "member", the label is plain text.++ assert.deepStrictEqual(follow.vip_badge, { creator: CREATOR, perk: 'subscriber-badge', name: 'Subscriber badge', badge: 'member', label: 'bCrew/b' });++ await until(() => { const r = h.db.getChatMessageById(msg.id); return r.metadata && JSON.parse(r.metadata).vip_badge; });++ const next = await say(ws, watcher);++ assert.strictEqual(next.msg.vip_badge.badge, 'member', 'cached from then on');++});++++t('VIP down: messages flow, without a badge', async () => {++ vip.state.down = true;++ try {++ clock.t += 61_000; // every cached answer is stale++ const { msg, ms } = await say(memberWs, watcher);++ assert.strictEqual(msg.vip_badge, undefined);++ assert.ok(ms < 1000);++ await new Promise((r) => setTimeout(r, 200));++ const again = await say(memberWs, watcher);++ assert.strictEqual(again.msg.vip_badge, undefined, 'the failure is cached briefly, never as a badge');++ } finally { vip.state.down = false; }++ clock.t += 5_001;++ const warmUp = await say(memberWs, watcher); // miss → lookup → follow-up frame++ await watcher.next((m) => m.type === 'chat_vip_badge' && m.id === warmUp.msg.id);++ assert.strictEqual((await say(memberWs, watcher)).msg.vip_badge.badge, 'subscriber');++});++++t('convergence: VIP stops granting → the badge stops within ttlMs (60 s); vip.membership.changed stops it at once', async () => {++ assert.deepStrictEqual(badges.bounds, { grantMs: 60_000, denyMs: 30_000, unavailableMs: 5_000 });++ // VIP applied a refund (billing.entitlement.changed) and emitted vip.membership.changed; Chat did not get it.++ vip.state.answers.set(`${MEMBER}|${CREATOR}`, inactive);++ clock.t += 59_000;++ assert.strictEqual((await say(memberWs, watcher)).msg.vip_badge.badge, 'subscriber', 'inside the bound the cached badge may still show');++ clock.t += 1_001;++ const past = await say(memberWs, watcher);++ assert.strictEqual(past.msg.vip_badge, undefined, 'past ttlMs: no badge');++ await until(() => vip.state.calls.filter((c) => c.subject === MEMBER).length >= 3);++ await new Promise((r) => setTimeout(r, 100));++ assert.ok(!watcher.all.some((m) => m.type === 'chat_vip_badge' && m.id === past.msg.id), 'no follow-up badge either');++ assert.strictEqual((await say(memberWs, watcher)).msg.vip_badge, undefined);++++ // The member rejoins; the badge is cached again. Then the event reaches Chat: gone at once.++ vip.state.answers.set(`${MEMBER}|${CREATOR}`, active([subscriberPerk]));++ clock.t += 30_001;++ const back = await say(memberWs, watcher);++ await watcher.next((m) => m.type === 'chat_vip_badge' && m.id === back.msg.id);++ assert.strictEqual((await say(memberWs, watcher)).msg.vip_badge.badge, 'subscriber');++ vip.state.answers.set(`${MEMBER}|${CREATOR}`, inactive);++ assert.strictEqual(badges.handleEvent({ event_type: 'vip.membership.changed', payload: { member: { type: 'user', id: MEMBER }, creator: { type: 'user', id: CREATOR }, active: false } }), true);++ assert.strictEqual((await say(memberWs, watcher)).msg.vip_badge, undefined, 'no clock movement needed');++});++++t('global chat (no creator room) never asks VIP', async () => {++ await new Promise((r) => setTimeout(r, 200)); // lookups started by the last step settle++ const n = vip.state.calls.length;++ const ws = await h.ws({ ip: '198.51.100.47', token: member.token });++ ws.sendJson({ type: 'join', token: member.token });++ await ws.next((m) => m.type === 'auth');++ ws.sendJson({ type: 'chat', message: 'hello global' });++ const got = await ws.next((m) => m.type === 'chat' && m.message === 'hello global');++ assert.strictEqual(got.vip_badge, undefined);++ assert.strictEqual(vip.state.calls.length, n);++ ws.close();++});++++t.run(async () => {++ try { h.chatServer.close(); } catch { /* */ }++ vip.server.close();++});
@@ -42,6 +42,16 @@ OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000
@@ -42,6 +42,16 @@ OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000
LIVE_MIRROR=0
LIVE_MIRROR=0
#LIVE_MIRROR_INTERVAL_MS=1000
#LIVE_MIRROR_INTERVAL_MS=1000
+# OpenVibe.VIP — member badges on chat messages (grant: chat vip.entitlement.check openvibe.vip).
+# Cache lifetimes (ms): a badge outlives a membership by at most CHAT_VIP_BADGE_TTL_MS after VIP
+# stops granting; a "no" is re-asked after the deny TTL, a failure after the unavailable TTL.
+#CHAT_VIP_BADGES=1
+#OV_VIP_INTERNAL_URL=http://127.0.0.1:4620
+#VIP_TIMEOUT_MS=1500
+#CHAT_VIP_BADGE_TTL_MS=60000
+#CHAT_VIP_BADGE_DENY_TTL_MS=30000
+#CHAT_VIP_BADGE_UNAVAILABLE_TTL_MS=5000
+
# OpenVibe.Events — the outbox relays only when set (otherwise events wait in events_outbox)
# OpenVibe.Events — the outbox relays only when set (otherwise events wait in events_outbox)
#EVENTS_URL=http://127.0.0.1:4300
#EVENTS_URL=http://127.0.0.1:4300
#EVENTS_RELAY_INTERVAL_MS=5000
#EVENTS_RELAY_INTERVAL_MS=5000
@@ -64,6 +64,7 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
@@ -64,6 +64,7 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
| Live's own chat pushes and writes (AI viewers, relays, donations, `/api/mod`, recaps, calls) | **Live → Chat** | `POST /internal/live/calls` (`chat.live_bridge.write`), presence `GET /internal/live/presence` (`chat.presence.read`) |
| Live's own chat pushes and writes (AI viewers, relays, donations, `/api/mod`, recaps, calls) | **Live → Chat** | `POST /internal/live/calls` (`chat.live_bridge.write`), presence `GET /internal/live/presence` (`chat.presence.read`) |
| Identity | **OpenVibe.Network** | user tokens are resolved by Live (its account links); service tokens from `/oauth/token` |
| Identity | **OpenVibe.Network** | user tokens are resolved by Live (its account links); service tokens from `/oauth/token` |
| Events | **OpenVibe.Events** | `events_outbox` → `POST /api/v1/events` when `EVENTS_URL` is set (`events.event.publish`) |
| Events | **OpenVibe.Events** | `events_outbox` → `POST /api/v1/events` when `EVENTS_URL` is set (`events.event.publish`) |
+| Member badges | **OpenVibe.VIP** (Billing holds the entitlement) | `server/vip/badges.js` → `POST /api/v1/entitlements/check` with `product: 'chat'` (`vip.entitlement.check`), behind the shared product cache |
### Data
### Data
@@ -96,6 +97,34 @@ in one transaction with its outbox row, and the relay publishes in outbox order,
@@ -96,6 +97,34 @@ in one transaction with its outbox row, and the relay publishes in outbox order,
always follows the message it removes. Messages deleted before this existed are redacted by
always follows the message it removes. Messages deleted before this existed are redacted by
OpenVibe.Events' `scripts/redact-backfill.js`.
OpenVibe.Events' `scripts/redact-backfill.js`.
+### VIP member badges
+
+A member's messages in a creator's room (stream or offline channel chat) carry that creator's VIP
+badge: a perk of the member's plan **version** with a `chat badge` product binding (VIP's network
+perk `subscriber-badge` has one), unless the member turned it off in VIP (`show_badge`). The frame
+gets `vip_badge: { creator, perk, name, badge, label? }` and the row's `metadata` keeps it, so history
+shows it. `badge` is a short id (`subscriber`, else `member`) and `label` plain text: binding config is
+written by creators, so nothing else of it passes and clients render both as text.
+
+- **Never blocks sending.** The badge comes from the cache only (`peekEntitlement`); the lookup starts
+ when a member joins a room, so their first line normally has it. On a miss the message goes out
+ without a badge and, if the lookup then finds one, a `chat_vip_badge` frame `{ id, vip_badge,
+ stream_id, channel_user_id }` follows to the same rooms (clients attach it by message id, as with
+ `chat_translation`) and the row's metadata is updated.
+- **Fails closed.** VIP unreachable or slow, no client secret, a refused grant, a malformed answer, an
+ inactive or `unknown` entitlement → no badge. `CHAT_VIP_BADGES=0` turns lookups off.
+- **Convergence bound.** Chat has no Events inbox, so when a membership ends (Billing's
+ `billing.entitlement.changed` → VIP's `vip.membership.changed`) the badge stops at most
+ `CHAT_VIP_BADGE_TTL_MS` (60 s) after VIP stops granting, never past the entitlement's `expires_at`;
+ end to end, add VIP's own bound (seconds with events, at most `VIP_PROJECTION_MAX_AGE_MS` without —
+ OpenVibe.VIP README, "The product cache and the convergence bound"). A new member's badge appears
+ within `CHAT_VIP_BADGE_DENY_TTL_MS` (30 s). `badges.handleEvent()` drops a pair at once, for when
+ the events are routed to Chat. `test/vip-badge.test.js` drives all of this against a stub VIP with an
+ injected clock.
+- Needs the Network grant `chat vip.entitlement.check openvibe.vip`; without it VIP answers 403 and
+ nobody gets a badge. The client is vendored from OpenVibe.VIP (`server/vip/vip-client.js`, copied
+ at VIP 2accbeb) until VIP publishes a tag to pin.
+
## Running it
## Running it
```bash
```bash
@@ -170,6 +199,7 @@ Events: `chat.message.created`, `chat.message.deleted`, `chat.dm.created`, `chat
@@ -170,6 +199,7 @@ Events: `chat.message.created`, `chat.message.deleted`, `chat.dm.created`, `chat
- OpenVibe.Live (until the rest of chat's data moves)
- OpenVibe.Live (until the rest of chat's data moves)
- OpenVibe.Events
- OpenVibe.Events
- OpenVibe.Media
- OpenVibe.Media
+- OpenVibe.VIP (member badges; optional — without it nobody has a badge)
- OpenVibe.Contracts
- OpenVibe.Contracts
## Acceptance (must be true before "done")
## Acceptance (must be true before "done")
@@ -9,10 +9,11 @@
@@ -9,10 +9,11 @@
"domain": "openvibe.live (/ws/chat, /api/chat/, /api/dm/, /api/tts/, /api/sounds via an nginx include in openvibe.live's vhost); openvibe.chat is still the OpenVibe.Sites placeholder",
"domain": "openvibe.live (/ws/chat, /api/chat/, /api/dm/, /api/tts/, /api/sounds via an nginx include in openvibe.live's vhost); openvibe.chat is still the OpenVibe.Sites placeholder",
"contracts": "openvibe-contracts v0.13.0",
"contracts": "openvibe-contracts v0.13.0",
"capabilities": ["chat.live_bridge.write", "chat.presence.read"],
"capabilities": ["chat.live_bridge.write", "chat.presence.read"],
- "capabilitiesUsed": ["live.chat_context.read", "live.chat_effects.write", "live.chat_mirror.write", "events.event.publish"],
+ "capabilitiesUsed": ["live.chat_context.read", "live.chat_effects.write", "live.chat_mirror.write", "events.event.publish", "vip.entitlement.check"],
"capabilitiesRegistered": true,
"capabilitiesRegistered": true,
"events": ["chat.message.created", "chat.message.deleted", "chat.dm.created", "chat.moderation.action"],
"events": ["chat.message.created", "chat.message.deleted", "chat.dm.created", "chat.moderation.action"],
"liveAuthorityFlag": "CHAT_AUTHORITY=chat is set in production live.env since the cutover (2026-09-23 02:03 UTC); Live's chat tables are a read mirror Chat keeps current",
"liveAuthorityFlag": "CHAT_AUTHORITY=chat is set in production live.env since the cutover (2026-09-23 02:03 UTC); Live's chat tables are a read mirror Chat keeps current",
+ "vip": "Member badges from OpenVibe.VIP (server/vip/badges.js: entitlements/check with product chat, cached 60 s, never blocks sending, fails closed). In code only, not deployed; needs the Network grant chat vip.entitlement.check openvibe.vip and Live's chat UI to render vip_badge / chat_vip_badge",
"deployed": true,
"deployed": true,
"deployment": "openvibe-ovh /opt/openvibe.chat, unit openvibe-chat, 127.0.0.1:4400 (loopback); public only through openvibe.live's nginx include for the chat paths; release 45f2b1f; EVENTS_URL set (chat.message.created reaches OpenVibe.Events); restore drill passed 2026-09-23",
"deployment": "openvibe-ovh /opt/openvibe.chat, unit openvibe-chat, 127.0.0.1:4400 (loopback); public only through openvibe.live's nginx include for the chat paths; release 45f2b1f; EVENTS_URL set (chat.message.created reaches OpenVibe.Events); restore drill passed 2026-09-23",
"updated": "2026-09-23"
"updated": "2026-09-23"
@@ -69,6 +69,8 @@ Live data maintained by `live-context` (never authority), `events_outbox`, `live
@@ -69,6 +69,8 @@ Live data maintained by `live-context` (never authority), `events_outbox`, `live
a default). Without `chat.message.send`, Chat refuses the bridge ops that send a message (AI
a default). Without `chat.message.send`, Chat refuses the bridge ops that send a message (AI
viewer, relay and donation lines, deploy notices). Add them to
viewer, relay and donation lines, deploy notices). Add them to
`DEFAULT_GRANTS` in `server/identity/principals.js` so every boot keeps them.
`DEFAULT_GRANTS` in `server/identity/principals.js` so every boot keeps them.
+ VIP member badges (after this cutover, any time): `chat vip.entitlement.check openvibe.vip`;
+ without it no message carries a badge (README, "VIP member badges").
3. **Live.** Apply `docs/live-patch.diff` to Live `main` (`git apply docs/live-patch.diff`), run its
3. **Live.** Apply `docs/live-patch.diff` to Live `main` (`git apply docs/live-patch.diff`), run its
tests (`node test/chat-context.test.js`, `npm test`), deploy it **without** `CHAT_AUTHORITY` —
tests (`node test/chat-context.test.js`, `npm test`), deploy it **without** `CHAT_AUTHORITY` —
nothing changes for users; the new internal routes answer only service tokens and every effect
nothing changes for users; the new internal routes answer only service tokens and every effect
@@ -30,6 +30,7 @@ const ttsEngine = require('./tts-engine');
@@ -30,6 +30,7 @@ const ttsEngine = require('./tts-engine');
const soundboard = require('./soundboard-service');
const soundboard = require('./soundboard-service');
const audioQueue = require('./audio-queue');
const audioQueue = require('./audio-queue');
const dm = require('./dm');
const dm = require('./dm');
+const vipBadges = require('../vip/badges');
// Cosmetics and tags are Live's (monetization/cosmetics, game/tags); read through live-context.
// Cosmetics and tags are Live's (monetization/cosmetics, game/tags); read through live-context.
const cosmetics = { getCosmeticProfile: (userId) => ctx.getCosmeticProfile(userId) };
const cosmetics = { getCosmeticProfile: (userId) => ctx.getCosmeticProfile(userId) };
const tags = { getTagProfile: (userId) => ctx.getTagProfile(userId) };
const tags = { getTagProfile: (userId) => ctx.getTagProfile(userId) };
@@ -434,6 +435,8 @@ class ChatServer {
@@ -434,6 +435,8 @@ class ChatServer {
}
}
client.channelUserId = channelUserId;
client.channelUserId = channelUserId;
client._modStream = null;
client._modStream = null;
+ // Look up this member's VIP badge for the room now, so their first message has it.
+ if (client.user && channelUserId) { try { vipBadges.warm(this._subjectOfUser(client.user), db.subjectFor(channelUserId)); } catch { /* */ } }
// A TTS/sound queue held since a restart plays once its room has listeners again.
// A TTS/sound queue held since a restart plays once its room has listeners again.
audioQueue.roomJoined(audioQueue.roomKey({ streamId: client.streamId, channelUserId: client.streamId ? null : channelUserId }));
audioQueue.roomJoined(audioQueue.roomKey({ streamId: client.streamId, channelUserId: client.streamId ? null : channelUserId }));
// Update viewer counts for old and new streams
// Update viewer counts for old and new streams
@@ -778,6 +781,18 @@ class ChatServer {
@@ -778,6 +781,18 @@ class ChatServer {
} catch { /* non-critical */ }
} catch { /* non-critical */ }
}
}
+ // The creator's VIP member badge (a perk bound to `chat badge`). From the cache only: the
+ // message never waits on VIP, and a miss is sent without a badge while the lookup runs; a
+ // badge found then follows as a chat_vip_badge frame for this message (below).
+ let vipBadgePending = null;
+ if (client.user?.id && client.channelUserId) {
+ try {
+ const r = vipBadges.forMessage(this._subjectOfUser(client.user), db.subjectFor(client.channelUserId));
+ if (r.badge) chatMsg.vip_badge = r.badge;
+ else vipBadgePending = r.pending;
+ } catch { /* no badge */ }
+ }
+
// Save to database
// Save to database
try {
try {
const result = db.saveChatMessage({
const result = db.saveChatMessage({
@@ -791,6 +806,7 @@ class ChatServer {
@@ -791,6 +806,7 @@ class ChatServer {
is_global: !client.streamId && !client.channelUserId,
is_global: !client.streamId && !client.channelUserId,
reply_to_id: replyToId,
reply_to_id: replyToId,
auto_delete_at: autoDeleteAt,
auto_delete_at: autoDeleteAt,
+ metadata: chatMsg.vip_badge ? { vip_badge: chatMsg.vip_badge } : undefined,
});
});
if (result.lastInsertRowid) chatMsg.id = Number(result.lastInsertRowid);
if (result.lastInsertRowid) chatMsg.id = Number(result.lastInsertRowid);
} catch { /* non-critical */ }
} catch { /* non-critical */ }
@@ -841,6 +857,7 @@ class ChatServer {
@@ -841,6 +857,7 @@ class ChatServer {
// Auto-translate (async): foreign → English for everyone, English → the channel's
// Auto-translate (async): foreign → English for everyone, English → the channel's
// language for a non-English streamer. Lands as a follow-up 'chat_translation'.
// language for a non-English streamer. Lands as a follow-up 'chat_translation'.
this._maybeTranslate(chatMsg, client.channelUserId, client.streamId);
this._maybeTranslate(chatMsg, client.channelUserId, client.streamId);
+ if (vipBadgePending) this._followVipBadge(vipBadgePending, chatMsg, client.channelUserId, client.streamId);
}
}
if (client.streamId) {
if (client.streamId) {
@@ -2132,6 +2149,28 @@ class ChatServer {
@@ -2132,6 +2149,28 @@ class ChatServer {
try { return ctx.channelLanguage(channelUserId); } catch { return 'en'; }
try { return ctx.channelLanguage(channelUserId); } catch { return 'en'; }
}
}
+ /** A signed-in user's Network subject (the projection's subject_id, else the ctx_users row). */
+ _subjectOfUser(user) {
+ if (!user) return null;
+ return user.subject_id || db.subjectFor(user.id) || null;
+ }
+
+ /**
+ * A VIP badge that was not cached when its message went out: when the lookup finds one, push
+ * it to the same rooms as a 'chat_vip_badge' event (clients attach it to the message by id) and
+ * keep it in the row's metadata so history shows it. Fire-and-forget; never throws.
+ */
+ _followVipBadge(pending, chatMsg, channelUserId, streamId) {
+ Promise.resolve(pending).then((badge) => {
+ if (!badge || !chatMsg || !chatMsg.id) return;
+ const evt = { type: 'chat_vip_badge', id: chatMsg.id, vip_badge: badge, stream_id: streamId || null, channel_user_id: channelUserId || null, timestamp: new Date().toISOString() };
+ this.broadcastToChannelRoom(channelUserId, streamId, evt);
+ if (streamId) this.forwardToGlobal(streamId, evt);
+ else this.forwardToGlobalByChannel(channelUserId, evt);
+ try { db.mergeChatMessageMetadata(chatMsg.id, { vip_badge: badge }); } catch { /* */ }
+ }).catch(() => { /* no badge */ });
+ }
+
/**
/**
* Translate a just-broadcast chat line and push the translation to the same rooms as a
* Translate a just-broadcast chat line and push the translation to the same rooms as a
* 'chat_translation' event (clients attach it under the message by id). Persisted into
* 'chat_translation' event (clients attach it under the message by id). Persisted into
@@ -64,6 +64,22 @@ module.exports = {
@@ -64,6 +64,22 @@ module.exports = {
syncStaleMs: int(process.env.LIVE_SYNC_STALE_MS, 60_000),
syncStaleMs: int(process.env.LIVE_SYNC_STALE_MS, 60_000),
},
},
+ // OpenVibe.VIP — the member badge on chat messages (a creator's perk bound to `chat badge`).
+ // Asked through a cache only (server/vip/badges.js): sending never waits on VIP, and any doubt
+ // (VIP down, no token, no answer yet) is "no badge". Needs Chat's client secret and the Network
+ // grant vip.entitlement.check on openvibe.vip.
+ vip: {
+ enabled: bool(process.env.CHAT_VIP_BADGES, true),
+ internalUrl: strip(process.env.OV_VIP_INTERNAL_URL || 'http://127.0.0.1:4620'),
+ timeoutMs: int(process.env.VIP_TIMEOUT_MS, 1500),
+ // How long a badge outlives the membership at most (VIP applied the change; Chat has no
+ // Events inbox): the convergence bound. A "no" is re-asked after denyTtlMs, a failure after
+ // unavailableTtlMs.
+ ttlMs: int(process.env.CHAT_VIP_BADGE_TTL_MS, 60_000),
+ denyTtlMs: int(process.env.CHAT_VIP_BADGE_DENY_TTL_MS, 30_000),
+ unavailableTtlMs: int(process.env.CHAT_VIP_BADGE_UNAVAILABLE_TTL_MS, 5_000),
+ },
+
// OpenVibe.Events — the outbox relays only when EVENTS_URL is set.
// OpenVibe.Events — the outbox relays only when EVENTS_URL is set.
events: {
events: {
url: strip(process.env.EVENTS_URL || ''),
url: strip(process.env.EVENTS_URL || ''),
new file mode 100644
new file mode 100644
@@ -0,0 +1,106 @@
@@ -0,0 +1,106 @@
+/**
+ * The VIP member badge on chat messages.
+ *
+ * A member's messages in a creator's room carry that creator's VIP badge: a perk of the member's plan
+ * version with a `chat badge` product binding (the network perk `subscriber-badge` has one), unless
+ * the member turned the badge off in VIP (show_badge). Resolved through VIP's entitlement check
+ * (vip.entitlement.check, product 'chat') behind the shared product cache (openvibe-vip/client
+ * createVipCache):
+ *
+ * - sending NEVER waits on VIP: forMessage() reads the cache synchronously. A miss sends the
+ * message without a badge and starts the lookup; when it finds a badge, chat-server pushes a
+ * `chat_vip_badge` frame for that message id (and stores it in the row's metadata);
+ * - it fails closed: VIP down, no token, a refused grant, a malformed answer → no badge;
+ * - convergence: Chat has no Events inbox, so a badge outlives the membership by at most
+ * config.vip.ttlMs (60 s) after VIP stopped granting (VIP README: "The product cache and the
+ * convergence bound"). handleEvent() drops a pair at once if events are ever routed here.
+ *
+ * The badge is plain data for Live's chat UI to render as text: { creator, perk, name, badge, label }.
+ * Binding config comes from creators, so only a short id (`badge`) and a short text label pass.
+ */
+'use strict';
+
+const config = require('../config');
+const serviceAuth = require('../net/service-auth');
+const { createVipClient, createVipCache } = require('./vip-client');
+
+const AUDIENCE = 'openvibe.vip';
+const SCOPE = 'vip.entitlement.check';
+const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+
+let cache = null;
+let clock = () => Date.now();
+let fetchImpl = null;
+
+function enabled() {
+ return !!(config.vip.enabled && config.vip.internalUrl && config.oauth.clientSecret);
+}
+
+function build() {
+ const tokenClient = {
+ authHeaders: () => serviceAuth.headers(AUDIENCE, SCOPE),
+ invalidate: () => serviceAuth.invalidate(AUDIENCE, SCOPE),
+ };
+ const vip = createVipClient({
+ baseUrl: config.vip.internalUrl, tokenClient, timeoutMs: config.vip.timeoutMs,
+ ...(fetchImpl ? { fetch: fetchImpl } : {}),
+ });
+ return createVipCache({
+ vip, ttlMs: config.vip.ttlMs, denyTtlMs: config.vip.denyTtlMs, unavailableTtlMs: config.vip.unavailableTtlMs, now: () => clock(),
+ });
+}
+function theCache() { if (!cache) cache = build(); return cache; }
+
+const text = (v, max) => (typeof v === 'string' ? v.replace(/[\u0000-\u001f\u007f<>]/g, '').trim().slice(0, max) : '');
+
+/** An entitlement answer (product 'chat') → the badge to show, or null. */
+function badgeOf(answer, creator) {
+ if (!answer || answer.active !== true || answer.status !== 'active') return null;
+ if (answer.preferences && answer.preferences.show_badge === false) return null;
+ for (const perk of Array.isArray(answer.product_perks) ? answer.product_perks : []) {
+ const b = (perk.bindings || []).find((x) => x && x.binding === 'badge');
+ if (!b) continue;
+ const cfg = b.config && typeof b.config === 'object' ? b.config : {};
+ const id = typeof cfg.badge === 'string' && /^[a-z0-9_-]{1,32}$/.test(cfg.badge) ? cfg.badge : 'member';
+ const out = { creator, perk: text(perk.key, 48), name: text(perk.name, 80), badge: id };
+ const label = text(cfg.label, 32);
+ if (label) out.label = label;
+ return out;
+ }
+ return null;
+}
+
+const pairOf = (memberSubject, creatorSubject) => (
+ USER_SUBJECT.test(String(memberSubject || '')) && USER_SUBJECT.test(String(creatorSubject || '')) && memberSubject !== creatorSubject
+ ? { subject: memberSubject, creator: creatorSubject, product: 'chat' } : null);
+
+/**
+ * For a message being sent: { badge } from the cache, or { badge: null, pending } — a promise of
+ * the badge (or null) that never rejects — when the cache has no answer yet. Never waits.
+ */
+function forMessage(memberSubject, creatorSubject) {
+ const args = enabled() ? pairOf(memberSubject, creatorSubject) : null;
+ if (!args) return { badge: null, pending: null };
+ const c = theCache();
+ const hit = c.peekEntitlement(args);
+ if (hit !== undefined) return { badge: badgeOf(hit, creatorSubject), pending: null };
+ return { badge: null, pending: c.entitlement(args).then((a) => badgeOf(a, creatorSubject), () => null) };
+}
+
+/** Look the pair up ahead of the first message (a join). Fire-and-forget. */
+function warm(memberSubject, creatorSubject) {
+ const args = enabled() ? pairOf(memberSubject, creatorSubject) : null;
+ if (args) theCache().entitlement(args).catch(() => {});
+}
+
+/** vip.membership.changed / billing.entitlement.changed / … → drop that pair now. */
+function handleEvent(envelope) { return enabled() ? theCache().handleEvent(envelope) : false; }
+
+/** Tests: an injected clock and fetch; rebuilds the cache. */
+function _configure({ now, fetch } = {}) {
+ if (now) clock = now;
+ if (fetch !== undefined) fetchImpl = fetch;
+ cache = null;
+}
+
+module.exports = { forMessage, warm, handleEvent, badgeOf, enabled, _configure, get bounds() { return theCache().bounds; } };
new file mode 100644
new file mode 100644
@@ -0,0 +1,235 @@
@@ -0,0 +1,235 @@
+// Copied verbatim from OpenVibe.VIP client/vip-client.js at 2accbeb (openvibe-vip/client). Replace with the
+// pinned package once VIP publishes a tag; do not edit here — change it in OpenVibe.VIP.
+'use strict';
+
+/**
+ * openvibe-vip/client — the consumer seam for products that honour VIP memberships (Blog, Wiki,
+ * Chat, Live, Community, …). No dependencies; Node 18+ (global fetch) or any fetch you pass.
+ *
+ * const { createVipClient } = require('openvibe-vip/client');
+ * const tokens = contracts.serviceAuth.createTokenClient({ tokenUrl, clientId, clientSecret,
+ * audience: 'openvibe.vip', scope: 'vip.resource.policy.evaluate vip.entitlement.check' });
+ * const vip = createVipClient({ baseUrl: 'http://127.0.0.1:4620', tokenClient: tokens });
+ *
+ * const d = await vip.evaluate({ subject: viewerSubjectOrNull, resource: { service: 'blog', type: 'post', id: '42' }, owner: authorSubject,
+ * fallback: { requirement: 'member', binding: 'blog:gated_post' } }); // the product's default gate
+ * if (!d.allow) return renderTeaser(d.reason); // FAILS CLOSED
+ *
+ * const e = await vip.checkEntitlement({ subject: 'usr_…', creator: 'usr_…', product: 'chat' }); // e.active, e.product_perks
+ * await vip.isMember('usr_…', 'usr_…'); // boolean, false on any doubt
+ *
+ * Every failure — VIP unreachable, a timeout, a refused token, a malformed answer — is a denial
+ * ({ allow: false, reason: 'vip_unavailable' } / { status: 'unknown', active: false }), never an
+ * exception and never a "yes". Pass mode: 'authoritative' for sensitive content: VIP then asks
+ * OpenVibe.Billing directly instead of its projection.
+ *
+ * Put createVipCache (below) in front of it: it holds the convergence bound every product shares.
+ */
+
+function createVipClient({ baseUrl = 'http://127.0.0.1:4620', tokenClient = null, getToken = null, fetch: fetchImpl = globalThis.fetch, timeoutMs = 2000, log = null } = {}) {
+ if (typeof fetchImpl !== 'function') throw new TypeError('createVipClient: a fetch implementation is required');
+ const base = String(baseUrl).replace(/\/+$/, '');
+
+ async function headers() {
+ if (tokenClient && typeof tokenClient.authHeaders === 'function') return tokenClient.authHeaders();
+ if (typeof getToken === 'function') return { Authorization: `Bearer ${await getToken()}` };
+ throw new Error('createVipClient: tokenClient or getToken is required');
+ }
+
+ async function post(path, body, retried = false) {
+ const res = await fetchImpl(`${base}${path}`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...(await headers()) },
+ body: JSON.stringify(body),
+ signal: AbortSignal.timeout(timeoutMs),
+ });
+ if (res.status === 401 && !retried && tokenClient && typeof tokenClient.invalidate === 'function') { tokenClient.invalidate(); return post(path, body, true); }
+ const data = await res.json().catch(() => null);
+ if (!res.ok || !data || typeof data !== 'object') throw new Error(`VIP ${res.status}: ${(data && (data.code || data.detail)) || 'bad response'}`);
+ return data;
+ }
+ const note = (what, err) => { if (log && log.warn) log.warn(`[vip-client] ${what} failed closed: ${err.message}`); };
+
+ /** May `subject` (a usr_ id, a SubjectRef, or null for a signed-out viewer) see `resource`? */
+ async function evaluate({ subject = null, resource, owner = null, ruleId, mode, fallback } = {}) {
+ try {
+ const d = await post('/api/v1/policies/evaluate', { subject: subject || null, resource, owner, rule_id: ruleId, mode, fallback });
+ return { ...d, allow: d.allow === true, reason: d.reason || (d.allow === true ? 'member' : 'denied') };
+ } catch (err) {
+ note('evaluate', err);
+ return { allow: false, reason: 'vip_unavailable', error: err.message };
+ }
+ }
+
+ /** The entitlement of `subject` to `creator` (usr_ ids or SubjectRefs). */
+ async function checkEntitlement({ subject, creator, mode, product } = {}) {
+ try {
+ const e = await post('/api/v1/entitlements/check', { subject, creator, mode, product });
+ return { ...e, active: e.active === true && e.status === 'active' };
+ } catch (err) {
+ note('checkEntitlement', err);
+ return { status: 'unknown', active: false, reason: 'vip_unavailable', error: err.message };
+ }
+ }
+
+ const isMember = async (subject, creator, opts = {}) => (await checkEntitlement({ subject, creator, mode: opts.mode })).active;
+
+ return { evaluate, checkEntitlement, isMember };
+}
+
+/**
+ * createVipCache({ vip, ttlMs, denyTtlMs, unavailableTtlMs, maxEntries, now }) — the product-side
+ * cache every consumer puts in front of the client, with ONE convergence rule:
+ *
+ * A cached "yes" lives at most ttlMs (and never past the entitlement's expires_at or VIP's
+ * valid_until). So once VIP stops granting — it has applied Billing's billing.entitlement.changed
+ * and emitted vip.membership.changed — the product stops granting within ttlMs. handleEvent(envelope)
+ * with either event (or billing.subscription.canceled / billing.transaction.reversed) drops that
+ * member's entries for that creator at once, so a product that receives the events converges
+ * immediately. A "no" lives denyTtlMs (a new member waits at most that long); a failure
+ * (vip_unavailable / unknown) lives unavailableTtlMs, and a cached "yes" is never extended by a
+ * failure: when VIP is down, entries expire into denials (fail closed).
+ *
+ * await cache.entitlement({ subject, creator, product }) → the checkEntitlement answer (cached)
+ * cache.peekEntitlement({ subject, creator, product }) → the cached answer or undefined; never waits
+ * (a miss starts a fetch in the background)
+ * await cache.evaluate({ subject, resource, owner, fallback, mode }) → the evaluate answer (cached;
+ * mode 'authoritative' is never cached)
+ * cache.invalidate({ member, creator }), cache.handleEvent(envelope), cache.clear()
+ */
+function createVipCache({ vip, ttlMs = 30_000, denyTtlMs = 10_000, unavailableTtlMs = 2_000, maxEntries = 5000, now = () => Date.now() } = {}) {
+ if (!vip || typeof vip.evaluate !== 'function' || typeof vip.checkEntitlement !== 'function') throw new TypeError('createVipCache: a VIP client is required');
+ const entries = new Map(); // key → { value, expires, member, creator }
+ const inflight = new Map(); // key → Promise
+ let generation = 0; // bumped by clear(); an answer fetched before a clear is not stored
+ const pairGen = new Map(); // `${member}|${creator}` → generation of its last invalidation
+
+ const idOf = (v) => (v && typeof v === 'object' ? (v.id != null ? String(v.id) : null) : (v != null && v !== '' ? String(v) : null));
+ const pairKey = (member, creator) => `${member || ''}|${creator || ''}`;
+
+ function lifetime(value, allowed) {
+ const t = now();
+ if (value && (value.reason === 'vip_unavailable' || value.status === 'unknown' || value.reason === 'entitlement_unknown')) return unavailableTtlMs;
+ if (!allowed) return denyTtlMs;
+ let life = ttlMs;
+ const ent = value.entitlement !== undefined ? value.entitlement : value;
+ for (const at of [ent && ent.expires_at, ent && ent.valid_until]) {
+ const ms = at ? Date.parse(at) : NaN;
+ if (Number.isFinite(ms)) life = Math.min(life, ms - t);
+ }
+ if (ent && ent.stale) life = Math.min(life, unavailableTtlMs);
+ return Math.max(0, life);
+ }
+
+ function store(key, value, allowed, member, creator, startedGen, startedPair) {
+ if (startedGen !== generation || (pairGen.get(pairKey(member, creator)) || 0) !== startedPair) return; // invalidated meanwhile
+ const life = lifetime(value, allowed);
+ if (life <= 0) { entries.delete(key); return; }
+ if (entries.size >= maxEntries) {
+ const t = now();
+ for (const [k, e] of entries) if (e.expires <= t) entries.delete(k);
+ while (entries.size >= maxEntries) entries.delete(entries.keys().next().value);
+ }
+ entries.set(key, { value, expires: now() + life, member, creator });
+ }
+
+ function fresh(key) {
+ const e = entries.get(key);
+ if (!e) return undefined;
+ if (e.expires <= now()) { entries.delete(key); return undefined; }
+ return e.value;
+ }
+
+ function load(key, member, creator, fetcher, allowedOf) {
+ const hit = fresh(key);
+ if (hit !== undefined) return Promise.resolve(hit);
+ if (inflight.has(key)) return inflight.get(key);
+ const g = generation;
+ const pg = pairGen.get(pairKey(member, creator)) || 0;
+ const p = (async () => {
+ try {
+ const value = await fetcher();
+ store(key, value, allowedOf(value), member, creator, g, pg);
+ return value;
+ } finally { inflight.delete(key); }
+ })();
+ inflight.set(key, p);
+ return p;
+ }
+
+ function entitlement({ subject, creator, product = null, mode } = {}) {
+ const member = idOf(subject);
+ const owner = idOf(creator);
+ if (!member || !owner) return Promise.resolve({ status: 'inactive', active: false, reason: 'not_signed_in' });
+ if (mode === 'authoritative') return vip.checkEntitlement({ subject: member, creator: owner, mode, product: product || undefined });
+ const key = `e|${member}|${owner}|${product || ''}`;
+ return load(key, member, owner, () => vip.checkEntitlement({ subject: member, creator: owner, product: product || undefined }), (v) => v.active === true);
+ }
+
+ function peekEntitlement(args = {}) {
+ const member = idOf(args.subject);
+ const owner = idOf(args.creator);
+ if (!member || !owner) return undefined;
+ const key = `e|${member}|${owner}|${args.product || ''}`;
+ const hit = fresh(key);
+ if (hit === undefined) entitlement(args).catch(() => {});
+ return hit;
+ }
+
+ function evaluate({ subject = null, resource, owner, fallback, mode } = {}) {
+ const member = idOf(subject);
+ const creator = idOf(owner);
+ if (mode === 'authoritative' || !resource || typeof resource !== 'object') return vip.evaluate({ subject: member, resource, owner: creator, fallback, mode });
+ if (!member) return Promise.resolve({ allow: false, reason: 'not_signed_in' });
+ const fb = fallback == null ? '' : JSON.stringify(fallback);
+ const key = `p|${member}|${creator || ''}|${resource.service}|${resource.type}|${resource.id}|${fb}`;
+ return load(key, member, creator, () => vip.evaluate({ subject: member, resource, owner: creator, fallback }), (v) => v.allow === true);
+ }
+
+ /** Drop what the cache knows about (member, creator); either may be omitted to drop more. */
+ function invalidate({ member = null, creator = null } = {}) {
+ const m = idOf(member);
+ const c = idOf(creator);
+ if (m && c) pairGen.set(pairKey(m, c), (pairGen.get(pairKey(m, c)) || 0) + 1);
+ else generation++;
+ let dropped = 0;
+ for (const [k, e] of entries) {
+ if ((!m || e.member === m) && (!c || e.creator === c)) { entries.delete(k); dropped++; }
+ }
+ return dropped;
+ }
+
+ /**
+ * An OpenVibe.Events envelope about a membership → invalidate that pair. Returns true when the
+ * event was understood. vip.membership.changed { member, creator }; billing.entitlement.changed
+ * { subject, streamer }; billing.subscription.canceled { subscription: { subscriber, streamer } };
+ * billing.transaction.reversed { to_subject (member), from_subject (creator) }.
+ */
+ function handleEvent(envelope) {
+ const e = envelope && envelope.event && envelope.event.event_type ? envelope.event : envelope;
+ const p = (e && e.payload) || {};
+ let member = null;
+ let creator = null;
+ switch (e && e.event_type) {
+ case 'vip.membership.changed': member = idOf(p.member); creator = idOf(p.creator); break;
+ case 'billing.entitlement.changed': member = idOf(p.subject); creator = idOf(p.streamer); break;
+ case 'billing.subscription.canceled': member = idOf(p.subscription && p.subscription.subscriber); creator = idOf(p.subscription && p.subscription.streamer); break;
+ case 'billing.transaction.reversed': member = idOf(p.to_subject); creator = idOf(p.from_subject); break;
+ default: return false;
+ }
+ if (!member || !creator) return false;
+ invalidate({ member, creator });
+ return true;
+ }
+
+ function clear() { generation++; entries.clear(); }
+
+ return {
+ entitlement, peekEntitlement, evaluate, invalidate, handleEvent, clear,
+ get size() { return entries.size; },
+ /** The convergence bound, for docs and /ready: how long a cached "yes" can outlive VIP's "no". */
+ bounds: Object.freeze({ grantMs: ttlMs, denyMs: denyTtlMs, unavailableMs: unavailableTtlMs }),
+ };
+}
+
+module.exports = { createVipClient, createVipCache };
new file mode 100644
new file mode 100644
@@ -0,0 +1,213 @@
@@ -0,0 +1,213 @@
+'use strict';
+/**
+ * The VIP member badge (roadmap Wave 10): a member's messages in a creator's room carry the
+ * creator's VIP badge — a perk of the member's plan version bound to `chat badge` — resolved through
+ * VIP's entitlement check (product 'chat') behind the shared product cache.
+ *
+ * - sending never waits on VIP (a slow VIP: the message goes out at once, the badge follows as
+ * a chat_vip_badge frame for that id and lands in the row's metadata)
+ * - fail closed: VIP down, not a member, badge turned off, no chat binding → no badge
+ * - convergence: once VIP stops granting, the badge stops within config.vip.ttlMs (60 s; Chat has
+ * no Events inbox), and at once when handleEvent gets vip.membership.changed
+ *
+ * A stub OpenVibe.VIP answers /api/v1/entitlements/check from a table and checks the service token
+ * (audience openvibe.vip, vip.entitlement.check).
+ */
+const assert = require('assert');
+const http = require('http');
+const { serviceAuth } = require('openvibe-contracts');
+const { boot, suite } = require('./helpers');
+
+const t = suite('vip-badge');
+const CREATOR = 'usr_01J9CREAT0R0000000000000AA';
+const MEMBER = 'usr_01J9MEMBER0000000000000AAA';
+const SLOWPOKE = 'usr_01J9S10WP0KE000000000000AA';
+const STRANGER = 'usr_01J9STRANGER00000000000AAA';
+const SHY = 'usr_01J9SHY0000000000000000AAA';
+const PLAIN = 'usr_01J9P1A1N00000000000000AAA';
+
+let h, badges, vip, streamer, member, slowpoke, stranger, shy, plain, streamId;
+const clock = { t: Date.now(), now() { return clock.t; } };
+
+/** An entitlement answer as VIP gives it with product 'chat'. */
+const active = (perks, { showBadge = true } = {}) => ({
+ status: 'active', active: true, expires_at: new Date(Date.now() + 30 * 86400e3).toISOString(), valid_until: new Date(Date.now() + 30 * 86400e3).toISOString(),
+ source: 'projection', stale: false, product: 'chat', product_perks: perks, preferences: { show_badge: showBadge, listed: false },
+});
+const subscriberPerk = { key: 'subscriber-badge', name: 'Subscriber badge', kind: 'badge', scope: 'network', bindings: [{ binding: 'badge', config: { badge: 'subscriber' } }] };
+const inactive = { status: 'inactive', active: false, expires_at: null, product: 'chat', product_perks: [], preferences: { show_badge: true, listed: false } };
+
+async function startVip(keyOf) {
+ const state = { answers: new Map(), calls: [], down: false, hold: new Map(), badTokens: 0 };
+ const server = http.createServer((req, res) => {
+ let raw = '';
+ req.on('data', (c) => { raw += c; });
+ req.on('end', async () => {
+ if (state.down) { req.socket.destroy(); return; }
+ const send = (status, body) => { res.statusCode = status; res.setHeader('Content-Type', 'application/json'); res.end(JSON.stringify(body)); };
+ const v = serviceAuth.verifyServiceToken(String(req.headers.authorization || '').slice(7), { publicKey: keyOf(), issuer: 'https://openvibe.network', audience: 'openvibe.vip' });
+ if (!v.ok) { state.badTokens++; return send(401, { code: v.code }); }
+ if (!(v.claims.cap || []).includes('vip.entitlement.check')) return send(403, { code: 'capability.denied' });
+ if (req.url !== '/api/v1/entitlements/check' || req.method !== 'POST') return send(404, { code: 'not_found' });
+ const b = JSON.parse(raw || '{}');
+ state.calls.push(b);
+ const hold = state.hold.get(b.subject);
+ if (hold) await hold;
+ send(200, { member: { type: 'user', id: b.subject }, creator: { type: 'user', id: b.creator }, ...(state.answers.get(`${b.subject}|${b.creator}`) || inactive) });
+ });
+ });
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ return { state, server, url: `http://127.0.0.1:${server.address().port}` };
+}
+
+const callsFor = (subject) => vip.state.calls.filter((c) => c.subject === subject).length;
+async function until(pred, ms = 3000) {
+ const end = Date.now() + ms;
+ while (!pred()) { if (Date.now() > end) throw new Error('timed out'); await new Promise((r) => setTimeout(r, 10)); }
+}
+async function joinRoom(user, ip) {
+ const ws = await h.ws({ ip, token: user.token, stream: streamId });
+ ws.sendJson({ type: 'join', streamId, token: user.token });
+ await ws.next((m) => m.type === 'auth');
+ return ws;
+}
+let seq = 0;
+async function say(ws, watcher) {
+ const text = `vip line ${++seq}`;
+ const started = Date.now();
+ ws.sendJson({ type: 'chat', message: text });
+ const got = await watcher.next((m) => m.type === 'chat' && m.message === text);
+ return { msg: got, ms: Date.now() - started };
+}
+
+t('boot with a stub VIP', async () => {
+ vip = await startVip(() => h.keys.publicKey);
+ h = await boot({ env: { OV_VIP_INTERNAL_URL: vip.url, CHAT_VIP_BADGE_TTL_MS: '60000', CHAT_VIP_BADGE_DENY_TTL_MS: '30000', CHAT_VIP_BADGE_UNAVAILABLE_TTL_MS: '5000' } });
+ badges = require('../server/vip/badges');
+ badges._configure({ now: () => clock.now() });
+ assert.ok(badges.enabled());
+ h.chatServer.DEFAULT_RATE_LIMIT_MS = 0; // many lines per test, same senders
+
+ streamer = h.addUser('streamer', { role: 'streamer', subject: CREATOR });
+ member = h.addUser('member', { subject: MEMBER });
+ slowpoke = h.addUser('slowpoke', { subject: SLOWPOKE });
+ stranger = h.addUser('stranger', { subject: STRANGER });
+ shy = h.addUser('shy', { subject: SHY });
+ plain = h.addUser('plain', { subject: PLAIN });
+ const ch = h.addChannel(streamer.id);
+ streamId = h.addStream(streamer.id, ch);
+ vip.state.answers.set(`${MEMBER}|${CREATOR}`, active([subscriberPerk]));
+ vip.state.answers.set(`${SLOWPOKE}|${CREATOR}`, active([{ ...subscriberPerk, bindings: [{ binding: 'badge', config: { badge: 'Bad Id!', label: '<b>Crew</b>\u0007' } }] }]));
+ vip.state.answers.set(`${SHY}|${CREATOR}`, active([subscriberPerk], { showBadge: false }));
+ vip.state.answers.set(`${PLAIN}|${CREATOR}`, active([{ key: 'backstage', name: 'Backstage', kind: 'gated_content', scope: 'creator', bindings: [{ binding: 'gated_post', config: {} }] }]));
+});
+
+let memberWs, watcher;
+t('a member\'s message carries the creator\'s badge (looked up at join, so the first line has it), persisted in metadata', async () => {
+ watcher = await joinRoom(stranger, '198.51.100.41');
+ memberWs = await joinRoom(member, '198.51.100.40');
+ await until(() => callsFor(MEMBER) >= 1);
+ const call = vip.state.calls.find((c) => c.subject === MEMBER);
+ assert.deepStrictEqual([call.creator, call.product], [CREATOR, 'chat']);
+ await new Promise((r) => setTimeout(r, 50));
+ const { msg } = await say(memberWs, watcher);
+ assert.deepStrictEqual(msg.vip_badge, { creator: CREATOR, perk: 'subscriber-badge', name: 'Subscriber badge', badge: 'subscriber' });
+ const row = h.db.getChatMessageById(msg.id);
+ assert.deepStrictEqual(JSON.parse(row.metadata).vip_badge, msg.vip_badge);
+ assert.ok(h.tokenRequests.some((r) => r.audience === 'openvibe.vip' && r.scope === 'vip.entitlement.check'), 'Chat asked the Network for a VIP token');
+ const before = callsFor(MEMBER);
+ await say(memberWs, watcher);
+ assert.strictEqual(callsFor(MEMBER), before, 'the second message comes from the cache');
+});
+
+t('fail closed: not a member, badge turned off, no chat binding → no badge; the creator\'s own lines never ask VIP', async () => {
+ for (const [u, ip, subject] of [[stranger, null, STRANGER], [shy, '198.51.100.43', SHY], [plain, '198.51.100.44', PLAIN]]) {
+ const ws = ip ? await joinRoom(u, ip) : watcher;
+ await until(() => callsFor(subject) >= 1);
+ await new Promise((r) => setTimeout(r, 30));
+ const { msg } = await say(ws, memberWs);
+ assert.strictEqual(msg.vip_badge, undefined, `${u.username}: no badge`);
+ }
+ const own = await joinRoom(streamer, '198.51.100.45');
+ const { msg } = await say(own, watcher);
+ assert.strictEqual(msg.vip_badge, undefined);
+ assert.strictEqual(vip.state.calls.filter((c) => c.subject === CREATOR).length, 0);
+});
+
+t('sending never waits on VIP: a slow lookup sends at once, the badge follows as chat_vip_badge and lands in metadata', async () => {
+ let release;
+ vip.state.hold.set(SLOWPOKE, new Promise((r) => { release = r; }));
+ const ws = await joinRoom(slowpoke, '198.51.100.46');
+ await until(() => callsFor(SLOWPOKE) >= 1);
+ const { msg, ms } = await say(ws, watcher);
+ assert.strictEqual(msg.vip_badge, undefined, 'sent before VIP answered');
+ assert.ok(ms < 1000, `delivered in ${ms} ms while VIP was holding`);
+ release();
+ vip.state.hold.delete(SLOWPOKE);
+ const follow = await watcher.next((m) => m.type === 'chat_vip_badge' && m.id === msg.id);
+ // Creator-written binding config is cleaned: an unknown badge id becomes "member", the label is plain text.
+ assert.deepStrictEqual(follow.vip_badge, { creator: CREATOR, perk: 'subscriber-badge', name: 'Subscriber badge', badge: 'member', label: 'bCrew/b' });
+ await until(() => { const r = h.db.getChatMessageById(msg.id); return r.metadata && JSON.parse(r.metadata).vip_badge; });
+ const next = await say(ws, watcher);
+ assert.strictEqual(next.msg.vip_badge.badge, 'member', 'cached from then on');
+});
+
+t('VIP down: messages flow, without a badge', async () => {
+ vip.state.down = true;
+ try {
+ clock.t += 61_000; // every cached answer is stale
+ const { msg, ms } = await say(memberWs, watcher);
+ assert.strictEqual(msg.vip_badge, undefined);
+ assert.ok(ms < 1000);
+ await new Promise((r) => setTimeout(r, 200));
+ const again = await say(memberWs, watcher);
+ assert.strictEqual(again.msg.vip_badge, undefined, 'the failure is cached briefly, never as a badge');
+ } finally { vip.state.down = false; }
+ clock.t += 5_001;
+ const warmUp = await say(memberWs, watcher); // miss → lookup → follow-up frame
+ await watcher.next((m) => m.type === 'chat_vip_badge' && m.id === warmUp.msg.id);
+ assert.strictEqual((await say(memberWs, watcher)).msg.vip_badge.badge, 'subscriber');
+});
+
+t('convergence: VIP stops granting → the badge stops within ttlMs (60 s); vip.membership.changed stops it at once', async () => {
+ assert.deepStrictEqual(badges.bounds, { grantMs: 60_000, denyMs: 30_000, unavailableMs: 5_000 });
+ // VIP applied a refund (billing.entitlement.changed) and emitted vip.membership.changed; Chat did not get it.
+ vip.state.answers.set(`${MEMBER}|${CREATOR}`, inactive);
+ clock.t += 59_000;
+ assert.strictEqual((await say(memberWs, watcher)).msg.vip_badge.badge, 'subscriber', 'inside the bound the cached badge may still show');
+ clock.t += 1_001;
+ const past = await say(memberWs, watcher);
+ assert.strictEqual(past.msg.vip_badge, undefined, 'past ttlMs: no badge');
+ await until(() => vip.state.calls.filter((c) => c.subject === MEMBER).length >= 3);
+ await new Promise((r) => setTimeout(r, 100));
+ assert.ok(!watcher.all.some((m) => m.type === 'chat_vip_badge' && m.id === past.msg.id), 'no follow-up badge either');
+ assert.strictEqual((await say(memberWs, watcher)).msg.vip_badge, undefined);
+
+ // The member rejoins; the badge is cached again. Then the event reaches Chat: gone at once.
+ vip.state.answers.set(`${MEMBER}|${CREATOR}`, active([subscriberPerk]));
+ clock.t += 30_001;
+ const back = await say(memberWs, watcher);
+ await watcher.next((m) => m.type === 'chat_vip_badge' && m.id === back.msg.id);
+ assert.strictEqual((await say(memberWs, watcher)).msg.vip_badge.badge, 'subscriber');
+ vip.state.answers.set(`${MEMBER}|${CREATOR}`, inactive);
+ assert.strictEqual(badges.handleEvent({ event_type: 'vip.membership.changed', payload: { member: { type: 'user', id: MEMBER }, creator: { type: 'user', id: CREATOR }, active: false } }), true);
+ assert.strictEqual((await say(memberWs, watcher)).msg.vip_badge, undefined, 'no clock movement needed');
+});
+
+t('global chat (no creator room) never asks VIP', async () => {
+ await new Promise((r) => setTimeout(r, 200)); // lookups started by the last step settle
+ const n = vip.state.calls.length;
+ const ws = await h.ws({ ip: '198.51.100.47', token: member.token });
+ ws.sendJson({ type: 'join', token: member.token });
+ await ws.next((m) => m.type === 'auth');
+ ws.sendJson({ type: 'chat', message: 'hello global' });
+ const got = await ws.next((m) => m.type === 'chat' && m.message === 'hello global');
+ assert.strictEqual(got.vip_badge, undefined);
+ assert.strictEqual(vip.state.calls.length, n);
+ ws.close();
+});
+
+t.run(async () => {
+ try { h.chatServer.close(); } catch { /* */ }
+ vip.server.close();
+});