openvibers/openvibe.chat Commits

Events consumer (compatibility register C-84): Chat subscribes to live.release.deployed and network.module.updated and receives them at POST /internal/events (server/events/consumer.js). live.release.deployed makes the deploy card exactly as the bridge op deployNotice does (same fold: newest row in any room is a deploy card under 3 hours old; same broadcast passes and late-joiner replay), keyed by subject.id, the head commit. Both paths claim the head in the new deploy_releases table in the transaction that stores or folds the card, so while Live sends both the same head is one card whichever arrives first; the second only records bridge_at / event_at. The bridge's head is its first commit (Live sends them newest first). A retried bridge op, a redelivered event and Live re-announcing after a crash are the same card; a redeploy of an already announced head (a rollback) says nothing; a release event older than 6 hours (an operator replay) is ignored:stale. The bridge call keeps working. network.module.updated for chat.preferences goes to prefs.handleEvent(): a newer revision drops the cached copy at once instead of within CHAT_PREFS_TTL_MS, an older or equal one and other namespaces change nothing (and are not recorded). Exactly once: the openvibe-sdk inbox (chat_event_inbox, pruned after 35 days) claims (chat, event_id) in the same transaction as the change; broadcasts run after the commit; a failure is 500 and Events retries. Signature v2 only (parseDelivery requireV2, ±300 s) under CHAT_EVENTS_SECRET (comma list for rotation, 32+ characters each): bad, stale, v1-only or unsigned is 401, no secret 503, a forwarded request 403. Subscriptions are created at boot when missing (server/events/subscriptions.js: list, then create, 409 = exists; retried in the background at 10 s, 1, 5 and 15 min), an existing one is left as it is even when disabled; CHAT_EVENTS_SUBSCRIBE=0 turns it off, CHAT_EVENTS_ENDPOINT overrides http://127.0.0.1:<PORT>/internal/events. scripts/subscribe-events.js --dry-run | --disable | --enable. Needs the Network grant chat events.subscription.manage openvibe.events. /ready reports the consumer's counters under events.consumer. Adds openvibe-sdk v0.5.0. test/events-consumer.test.js: signature checks, the inbox, bridge-then-event and event-then-bridge for one head, folding and a stream message starting a new card across both paths, release validation and staleness, the preferences cache with older revisions ignored, idempotent boot subscriptions against a stub Events, inbox pruning. README (Events: Consumed), .env.example, STATUS.json

OP OpenVibers committed 23 Sept 2026 · ed90461e2f · 16 files, +936 −42


Changes

ddiff --git a/.env.example b/.env.example
iindex b7209cc..97d6231 100644
---- a/.env.example
++++ b/.env.example
@@@ -62,6 +62,16 @@ LIVE_MIRROR=0
# 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
++# Events consumer: POST /internal/events receives Chat's subscriptions (live.release.deployed → the
++# deploy card; network.module.updated → the chat.preferences cache). Signature v2 under this secret
++# (32+ characters: openssl rand -hex 32; comma-separate a second one while rotating — the FIRST is
++# the one handed to Events). Unset = the route answers 503 and no subscription is made.
++#CHAT_EVENTS_SECRET=
++# Boot creates the subscriptions when missing (needs EVENTS_URL, the secret, OV_OAUTH_CLIENT_SECRET
++# and the Network grant chat events.subscription.manage on openvibe.events); 0 = leave them alone.
++#CHAT_EVENTS_SUBSCRIBE=1
++# Where Events delivers (default http://127.0.0.1:$PORT/internal/events)
++#CHAT_EVENTS_ENDPOINT=http://127.0.0.1:4400/internal/events
# Public origin used in /paste links (Live's SITE_URL)
SITE_URL=https://openvibe.live
ddiff --git a/README.md b/README.md
iindex 785d15b..380a59b 100644
---- a/README.md
++++ b/README.md
@@@ -63,7 +63,7 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
| Coins, AI viewers, arena, media queue, hardware, pastes, translation, PowerChat, notifications | **Live** | `server/live-context.js` → `POST /internal/chat-effects/*` (`live.chat_effects.write`) |
| 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`) |
++| Events | **OpenVibe.Events** | `events_outbox` → `POST /api/v1/events` when `EVENTS_URL` is set (`events.event.publish`); Chat's subscriptions deliver to `POST /internal/events` (`server/events/consumer.js`, `events.subscription.manage`) |
| A person's chat preferences | **OpenVibe.Network** user module `chat.preferences` (Chat owns the namespace) | `server/prefs/` → `GET/PUT/DELETE /internal/modules/chat.preferences/:subject` (`network.modules.read` / `.write`), cached per person |
| 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 |
@@@ -98,6 +98,46 @@ 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
OpenVibe.Events' `scripts/redact-backfill.js`.
++#### Consumed
++
++Chat subscribes (consumer `chat`) to two topics, delivered to `POST /internal/events`
++(`server/events/consumer.js`):
++
++| Event | From | What Chat does |
++| --- | --- | --- |
++| `live.release.deployed` | Live (`server/events/release-events.js`), subject `{ type: release, id: <head commit> }` | stores or folds the deploy card in global chat, exactly as the bridge op `deployNotice` does (`server/chat/deploy-notice.js`: one rolling card, folded while nobody has spoken in any room and within 3 hours; the broadcast carries the row id; late joiners get it once) |
++| `network.module.updated` | Network (`server/identity/module-events.js`) | for `chat.preferences`, a revision newer than the cached copy drops it (`prefs.handleEvent()`); an older or equal revision and other namespaces change nothing |
++
++- **One card per deploy, whichever path comes first** (compatibility register C-84). Until the bridge
++ op is removed, Live sends each deploy twice: `deployNotice` over the bridge (commits newest first,
++ so the first is the head) and the event. Both claim the head commit in `deploy_releases` in the
++ transaction that stores or folds the card; the second finds it claimed and changes nothing but its
++ `bridge_at` / `event_at`. A retried bridge op, a redelivered event and Live re-announcing after a
++ crash are the same card. A redeploy of a head already announced (a rollback to it) says nothing.
++ A release event older than 6 hours (an operator replay) is `ignored:stale`.
++- **Exactly once.** The openvibe-sdk inbox (`chat_event_inbox`, pruned after 35 days) claims
++ `(chat, event_id)` in the same SQLite transaction as the change, and broadcasts run only after it
++ commits. A failure answers 500 and rolls both back; Events retries.
++- **Signature v2 only** (`parseDelivery` with `requireV2`): HMAC over `"<t>.<raw body>"` under
++ `CHAT_EVENTS_SECRET`, `t` within ±300 s. A bad, stale, v1-only or unsigned delivery is 401; no secret
++ is 503; a request that came through nginx (a forwarding header) is 403.
++- **Subscriptions** are created at boot when missing (`server/events/subscriptions.js`: list, then
++ create; Events answers 409 for a duplicate, so it is idempotent), retried in the background while
++ Network or Events cannot answer. An existing subscription is left as it is, even disabled, so a
++ rollback survives restarts. `CHAT_EVENTS_SUBSCRIBE=0` turns this off. Needs `EVENTS_URL`,
++ `CHAT_EVENTS_SECRET`, `OV_OAUTH_CLIENT_SECRET` and the Network grant
++ `chat events.subscription.manage openvibe.events`.
++
++```bash
++sudo node --env-file=/etc/openvibe/chat.env scripts/subscribe-events.js --dry-run # list them
++sudo node --env-file=/etc/openvibe/chat.env scripts/subscribe-events.js --disable # rollback: no more deliveries
++sudo node --env-file=/etc/openvibe/chat.env scripts/subscribe-events.js --enable # undo it
++```
++
++`/ready` reports the consumer under `events.consumer` (received, applied, duplicates, ignored, refused,
++failed, the last type and outcome); `SELECT * FROM deploy_releases ORDER BY created_at DESC` shows
++which path delivered each deploy first and when the other one arrived.
++
### VIP member badges
A member's messages in a creator's room (stream or offline channel chat) carry that creator's VIP
@@@ -114,13 +154,13 @@ written by creators, so nothing else of it passes and clients render both as tex
`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
++- **Convergence bound.** Chat does not subscribe to VIP's events yet, 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
++ `vip.membership.changed` is added to the Events consumer. `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
@@@ -139,9 +179,10 @@ every write, so another writer is never overwritten (Chat reads again and re-app
If-Match is strict: 412).
- **Cache.** One entry per person for `CHAT_PREFS_TTL_MS` (60 s). Chat's own writes land in it at once; a
-- change made through Network directly (the person's account page) shows within the TTL, and at once
-- when a `network.module.updated` for it reaches `prefs.handleEvent()` (for when Events deliveries reach
-- Chat). Network down: the cached copy with `stale: true`, else 503; writes 503.
++ change made through Network directly (the person's account page) shows at once: its
++ `network.module.updated` reaches the Events consumer, and `prefs.handleEvent()` drops the cached copy
++ when the revision is newer (without the subscription, within the TTL). Network down: the cached copy
++ with `stale: true`, else 503; writes 503.
- API tokens read only. A person Live knows without a Network subject gets 409 `prefs.subject_unknown`.
- **Migration** of what Live kept server-side (`user_preferences.chat_settings`, the browser's whole
`chatSettings`): `scripts/migrate-chat-preferences.js`. Only choices move (`showTimestamps` →
@@@ -166,12 +207,14 @@ npm test # Node 22; stub Live and Network in-process
node scripts/import-from-live.js --live-db /tmp/live-snapshot.db --dry-run
node scripts/parity-check.js --live https://openvibe.live --chat http://127.0.0.1:4401 --before "…"
node scripts/mirror-flush.js # rollback helper: push queued mirror rows to Live
++node scripts/subscribe-events.js --dry-run # Chat's Events subscriptions (boot creates missing ones)
```
`/ready` (openvibe-shared/ready shape: `status` ready/degraded/not_ready and `checks`) is 503 only
when the `db` check (a read of `chat_messages`) fails. `live_sync` is optional: it reads degraded when
the last clean Live sync is older than `LIVE_SYNC_STALE_MS` (default 60 s) or a sync step has missed
--its own interval by that much. It also reports the mirror queue and whether events are relayed.
++its own interval by that much. It also reports the mirror queue, whether events are relayed, and the
++Events consumer's counters.
Production: `deploy/systemd/openvibe-chat.service`, `deploy/nginx/openvibe.live-chat.locations.conf`,
`/etc/openvibe/chat.env`. The whole switch-over — rehearsal, import, parity checks, nginx, the flag,
rollback — is `docs/cutover.md`.
@@@ -186,10 +229,12 @@ server/chat/ moved from Live: chat-server, dm, dm-routes, routes,
server/auth/ token resolution through Live; the chat subset of Live's permissions
server/bridge/ Live → Chat calls + presence (live-bridge.js); Chat → Live read mirror (live-mirror.js)
server/events/outbox.js events.event-envelope@1 outbox and relay
++server/events/consumer.js POST /internal/events: Chat's Events subscriptions (live.release.deployed, network.module.updated)
++server/events/subscriptions.js creates them at boot when missing; list/disable/enable for scripts/subscribe-events.js
server/net/service-auth.js service tokens: client (Chat → others) and guard (others → Chat)
server/prefs/ chat preferences in the Network user module chat.preferences (routes, cache, migration from Live)
server/db/ schema.sql, database.js (Live's chat functions, same names and arguments)
--scripts/ import-from-live, parity-check, mirror-flush, migrate-chat-preferences
++scripts/ import-from-live, parity-check, mirror-flush, migrate-chat-preferences, subscribe-events
docs/ cutover.md, live-patch.diff, capabilities-proposal/
```
@@@ -223,7 +268,9 @@ to the chat manifest (then the check uses a literal id the contracts check can s
`chat.room.*`, `chat.message.*`, `chat.dm.*`, `chat.moderation.*`, `chat.tts.*`, `chat.call.*`.
Events: `chat.message.created`, `chat.message.deleted`, `chat.dm.created`, `chat.moderation.action`
--(produced); planned `chat.room.updated`, `chat.call.*`, `chat.tts.queued|played|failed`.
++(produced); planned `chat.room.updated`, `chat.call.*`, `chat.tts.queued|played|failed`. Consumed:
++`live.release.deployed`, `network.module.updated` (Chat's own subscriptions: Network grant
++`chat events.subscription.manage openvibe.events`).
## Depends on
ddiff --git a/STATUS.json b/STATUS.json
iindex d21d7cf..1d3e057 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -9,9 +9,10 @@
"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", "vip.entitlement.check", "network.modules.read", "network.modules.write"],
++ "capabilitiesUsed": ["live.chat_context.read", "live.chat_effects.write", "live.chat_mirror.write", "events.event.publish", "events.subscription.manage", "vip.entitlement.check", "network.modules.read", "network.modules.write"],
"capabilitiesRegistered": true,
"events": ["chat.message.created", "chat.message.deleted", "chat.dm.created", "chat.moderation.action"],
++ "eventsConsumed": "live.release.deployed (the deploy card, folded by subject.id with the bridge's deployNotice: one card per head whichever comes first; register C-84) and network.module.updated (drops a cached chat.preferences copy), POST /internal/events, signature v2, openvibe-sdk inbox; subscriptions created at boot. In code only, not deployed; needs CHAT_EVENTS_SECRET in chat.env and the Network grant chat events.subscription.manage openvibe.events",
"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",
"preferences": "Chat preferences in the Network user module chat.preferences (server/prefs/: GET/PUT /api/chat/preferences with Chat's service token, cached 60 s, If-Match on every write; scripts/migrate-chat-preferences.js moves the choices in Live's user_preferences). In code only, not deployed; needs Network with the chat.preferences handoff and the grants chat network.modules.read/write on chat.preferences, then the migration, then Live's chat.js calling /api/chat/preferences",
ddiff --git a/package-lock.json b/package-lock.json
iindex b3ec03c..7861e64 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -17,6 +17,7 @@
"express-rate-limit": "^7.5.0",
"multer": "^2.0.2",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.13.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.5.0",
"ws": "^8.18.0"
},
"engines": {
@@@ -969,6 +970,15 @@
"node": ">=20"
}
},
++ "node_modules/openvibe-sdk": {
++ "version": "0.5.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.5.0",
++ "integrity": "sha512-QVJkR5iazQX0ZwIrDp2A88ZeYYBSXEmSzhpgURPUSnzezIM3jz0HBhVKowyhkvg2aBO61XNf77eb4zJGkMjZiw==",
++ "license": "MIT",
++ "engines": {
++ "node": ">=20"
++ }
++ },
"node_modules/parseurl": {
"version": "1.3.3",
"resolved": "https://registry.npmjs.org/parseurl/-/parseurl-1.3.3.tgz",
ddiff --git a/package.json b/package.json
iindex ef01855..278dd41 100644
---- a/package.json
++++ b/package.json
@@@ -28,6 +28,7 @@
"express-rate-limit": "^7.5.0",
"multer": "^2.0.2",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.13.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.5.0",
"ws": "^8.18.0"
},
"engines": {
ddiff --git a/scripts/subscribe-events.js b/scripts/subscribe-events.js
nnew file mode 100644
iindex 0000000..ad72722
---- /dev/null
++++ b/scripts/subscribe-events.js
@@@ -0,0 +1,46 @@
++#!/usr/bin/env node
++/**
++ * Chat's OpenVibe.Events subscriptions (server/events/subscriptions.js): live.release.deployed and
++ * network.module.updated → POST /internal/events. Boot already creates missing ones; this lists
++ * them and is the rollback switch. Uses the service's env file (0600, so run it as root):
++ *
++ * sudo node --env-file=/etc/openvibe/chat.env scripts/subscribe-events.js --dry-run # list, change nothing
++ * sudo node --env-file=/etc/openvibe/chat.env scripts/subscribe-events.js # create what is missing
++ * sudo node --env-file=/etc/openvibe/chat.env scripts/subscribe-events.js --disable # rollback: stop deliveries
++ * sudo node --env-file=/etc/openvibe/chat.env scripts/subscribe-events.js --enable # undo a --disable
++ *
++ * [--endpoint http://127.0.0.1:4400/internal/events] [--topic live.release.deployed] (repeatable)
++ *
++ * A disabled subscription stays disabled across Chat restarts (boot never re-enables one). Nothing
++ * secret is printed.
++ */
++'use strict';
++
++const config = require('../server/config');
++const subscriptions = require('../server/events/subscriptions');
++
++function parseArgs(argv) {
++ const o = { action: 'create', endpoint: subscriptions.endpointFor(config), topics: [] };
++ for (let i = 0; i < argv.length; i++) {
++ const a = argv[i];
++ if (a === '--dry-run') o.action = 'list';
++ else if (a === '--disable') o.action = 'disable';
++ else if (a === '--enable') o.action = 'enable';
++ else if (a === '--endpoint') o.endpoint = argv[++i];
++ else if (a === '--topic') o.topics.push(argv[++i]);
++ else throw new Error(`unknown argument ${a}`);
++ }
++ for (const t of o.topics) if (!subscriptions.TOPICS.includes(t)) throw new Error(`--topic ${t} is not one of ${subscriptions.TOPICS.join(', ')}`);
++ if (!o.endpoint) throw new Error('--endpoint needs a value');
++ return o;
++}
++
++(async () => {
++ const o = parseArgs(process.argv.slice(2));
++ const results = await subscriptions.ensure({ config, endpoint: o.endpoint, action: o.action, topics: o.topics.length ? o.topics : subscriptions.TOPICS });
++ for (const r of results) {
++ const state = r.enabled === false ? 'disabled' : r.enabled === true ? 'enabled' : 'state unknown';
++ console.log(`${r.result.padEnd(8)} ${r.id || '-'} (${r.topic} → ${o.endpoint}${r.id ? `, ${state}` : ''})`);
++ }
++ if (o.action === 'list' && results.some((r) => r.result === 'missing')) console.log('missing ones are created at Chat\'s next boot, or by running this without --dry-run');
++})().catch((err) => { console.error(`subscribe-events failed: ${err.message}`); process.exit(1); });
ddiff --git a/server/app.js b/server/app.js
iindex 4f5427e..ed996a4 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -58,7 +58,7 @@ function getAllowedOrigins() {
return allowed;
}
--function createApp({ chatServer, bridge, mirror, relay }) {
++function createApp({ chatServer, bridge, mirror, relay, events = null }) {
const allowedOrigins = getAllowedOrigins();
const app = express();
app.disable('x-powered-by');
@@@ -74,6 +74,9 @@ function createApp({ chatServer, bridge, mirror, relay }) {
// ── Internal (loopback; never routed by nginx) ─────────────
// Alert sounds travel as base64 in broadcasts (uploads up to MAX_SOUND_SIZE_KB): room for them.
app.use('/internal/live', express.json({ limit: '16mb' }), bridge);
++ // OpenVibe.Events deliveries (live.release.deployed, network.module.updated): signature v2 over
++ // the raw body, so this router reads the body itself (server/events/consumer.js).
++ if (events) app.use('/internal/events', events.router);
app.get('/health', (req, res) => res.json({ ok: true, service: 'chat' }));
// Readiness in the openvibe-shared/ready shape (status ready/degraded/not_ready, named checks):
// 503 only when the required check fails. `db` is Chat's own database, which it cannot serve
@@@ -121,7 +124,10 @@ function createApp({ chatServer, bridge, mirror, relay }) {
connections: chatServer.getTotalConnections(),
live: { last_sync_at: ctx.stats.lastSyncAt, last_success_at: ctx.stats.lastSuccessAt, failures: ctx.stats.failures, last_error: ctx.stats.lastError },
mirror: mirror ? { enabled: config.live.mirror, pending, last_error: mirror.lastError() } : null,
-- events: { enabled: !!config.events.url },
++ events: {
++ enabled: !!config.events.url,
++ consumer: events ? { enabled: events.enabled, ...events.stats } : null,
++ },
});
});
ddiff --git a/server/chat/deploy-notice.js b/server/chat/deploy-notice.js
iindex 08d02d6..7fa741a 100644
---- a/server/chat/deploy-notice.js
++++ b/server/chat/deploy-notice.js
@@@ -3,11 +3,22 @@
// Deploy notices in chat — one tidy, rolling message instead of a line per restart.
// (Moved from OpenVibe.Live server/chat/deploy-notice.js, W6.)
//
--// Live still decides WHAT shipped — its git HEAD and the `deploy_last_announced` site setting —
--// and hands the new commits to Chat over the bridge (bridge op deployNotice). Chat keeps the part
--// that is chat:
--// • Consecutive deploys fold into ONE stored message: while the newest global chat row is a deploy
--// notice (nobody has spoken since) and it is under 12 hours old, that row is updated in place.
++// Live still decides WHAT shipped — its git HEAD and the `deploy_last_announced` site setting — and
++// tells Chat two ways while the Chat bridge is being retired (compatibility register C-84):
++// • the bridge op deployNotice (Live server/chat/chat-remote.js), commits newest first, so the
++// first one is the head Live deployed → announceCommits();
++// • the durable OpenVibe.Events event live.release.deployed (Live server/events/release-events.js),
++// subject { type: 'release', id: <head> } → applyReleaseEvent(), from ../events/consumer.js.
++// Both claim the release by its head commit in deploy_releases, in the SAME transaction that stores
++// or folds the card: the first to arrive makes the card, the other finds the head claimed and only
++// records when it saw it. So the same head never makes two cards, whichever path comes first, and a
++// repeat of either (a bridge retry, an Events redelivery, Live re-announcing after a crash) is one
++// card. Once the event path has carried a clean deploy, the bridge op can go.
++//
++// Chat keeps the part that is chat:
++// • Consecutive deploys fold into ONE stored message: while the newest chat row in ANY room is a
++// deploy notice (nobody has spoken since) and its first deploy is under 3 hours old, that row is
++// updated in place.
// • The row stores data, not prose: metadata { kind: 'deploy', commits[], deploys, first_at,
// updated_at }. Clients render times from ISO timestamps, so they are always right.
// • The live broadcast carries the row id; clients replace the card with that id instead of
@@@ -19,6 +30,10 @@
const FOLD_WINDOW_MS = 3 * 60 * 60 * 1000;
const MAX_COMMITS = 40;
const ATTEMPTS_MS = [5000, 20000, 45000]; // clients reconnect with backoff after a restart
++// A live.release.deployed older than this announces nothing (an operator replay, or Events catching
++// up after a long outage): the card would sit under hours of chat with an old time.
++const RELEASE_MAX_AGE_MS = 6 * 60 * 60 * 1000;
++const HASH_RE = /^[0-9a-f]{40}$/;
// The notice for the announcing boot, kept for a while so clients that reconnect late still get it exactly once.
const REPLAY_MS = 15 * 60 * 1000;
@@@ -27,7 +42,12 @@ let _live = null; // { payload, until, sent: WeakSet }
const parseLog = (raw) => raw.trim().split('\n').filter(Boolean).map((line) => {
const [hash, short, date, ...rest] = line.split('\x1f');
return { hash, short, date, subject: rest.join(' ').trim().slice(0, 200) };
--}).filter(c => /^[0-9a-f]{40}$/.test(c.hash || ''));
++}).filter(c => HASH_RE.test(c.hash || ''));
++
++/** Live's commit list as stored on the card: full hashes only, fields capped (both paths). */
++const cleanCommits = (commits) => (Array.isArray(commits) ? commits : []).filter((c) => c && HASH_RE.test(c.hash || '')).map((c) => ({
++ hash: c.hash, short: String(c.short || c.hash.slice(0, 7)).slice(0, 12), date: String(c.date || ''), subject: String(c.subject || '').slice(0, 200),
++}));
const plainText = (meta) => `🚀 ${meta.commits.length} update${meta.commits.length === 1 ? '' : 's'} shipped: ${meta.commits.slice(0, 3).map(c => c.subject).join(' · ')}${meta.commits.length > 3 ? ` · and ${meta.commits.length - 3} more` : ''}`;
@@@ -49,27 +69,35 @@ function persist(db, commits) {
const merged = commits.concat((prev.commits || []).filter(c => !seen.has(c.hash))).slice(0, MAX_COMMITS);
const meta = { kind: 'deploy', commits: merged, deploys: (prev.deploys || 1) + 1, first_at: prev.first_at, updated_at: nowIso };
db.run('UPDATE chat_messages SET message = ?, metadata = ? WHERE id = ?', [plainText(meta), JSON.stringify(meta), newest.id]);
-- return { id: newest.id, meta };
++ return { id: newest.id, meta, folded: true };
}
const meta = { kind: 'deploy', commits: commits.slice(0, MAX_COMMITS), deploys: 1, first_at: nowIso, updated_at: nowIso };
const res = db.saveChatMessage({ stream_id: null, user_id: null, anon_id: null, username: 'OpenVibe.Live', message: plainText(meta), message_type: 'system', is_global: true, metadata: meta });
-- return { id: res && (res.lastInsertRowid || res.lastID || res.id) || null, meta };
++ return { id: Number(res && (res.lastInsertRowid || res.lastID || res.id)) || null, meta, folded: false };
}
/**
-- * Announce commits Live just shipped (Live filtered them against deploy_last_announced).
-- * @returns {{ announced: number }}
++ * Claim the release (its head commit) and store or fold its card, in ONE transaction. The first path
++ * to arrive makes the card; the other finds the head claimed, notes when it saw it, and changes
++ * nothing else. Returns { duplicate: true, id } or { duplicate: false, saved }.
*/
--function announceCommits({ db, chatServer, commits, log = console }) {
-- const list = (Array.isArray(commits) ? commits : []).filter((c) => c && /^[0-9a-f]{40}$/.test(c.hash || '')).map((c) => ({
-- hash: c.hash, short: String(c.short || c.hash.slice(0, 7)).slice(0, 12), date: String(c.date || ''), subject: String(c.subject || '').slice(0, 200),
-- }));
-- if (!list.length) return { announced: 0 };
--
-- let saved;
-- try { saved = persist(db, list); }
-- catch (err) { log.warn('[Deploy notice] not saved:', err.message); return { announced: 0 }; }
++function claimRelease(db, { head, commits, via, eventId = null, now = Date.now() }) {
++ const seenCol = via === 'events' ? 'event_at' : 'bridge_at';
++ return db.transaction(() => {
++ const claimed = db.get('SELECT message_id FROM deploy_releases WHERE head = ?', [head]);
++ if (claimed) {
++ db.run(`UPDATE deploy_releases SET ${seenCol} = COALESCE(${seenCol}, ?), event_id = COALESCE(event_id, ?) WHERE head = ?`, [now, eventId, head]);
++ return { duplicate: true, id: claimed.message_id };
++ }
++ const saved = persist(db, commits);
++ db.run(`INSERT INTO deploy_releases (head, message_id, first_via, ${seenCol}, event_id, commit_count, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`,
++ [head, saved.id, via, now, eventId, commits.length, now]);
++ return { duplicate: false, saved };
++ });
++}
++/** Push the card to every connected client (three passes: clients reconnect with backoff) and keep it for late joiners. */
++function broadcast({ chatServer, saved, list, log = console }) {
const payload = JSON.stringify({ type: 'update', kind: 'deploy', id: saved.id, fresh: list.map(c => c.hash), url: '/updates', timestamp: saved.meta.updated_at, ...saved.meta });
const sent = new WeakSet();
_live = { payload, until: Date.now() + REPLAY_MS, sent };
@@@ -82,7 +110,58 @@ function announceCommits({ db, chatServer, commits, log = console }) {
return n;
};
ATTEMPTS_MS.forEach((ms, i) => { const t = setTimeout(() => { const n = push(); if (n) log.log(`[Deploy notice] ${list.length} commit(s) → ${n} client(s) (pass ${i + 1})`); }, ms); if (t.unref) t.unref(); });
-- return { announced: list.length, id: saved.id };
++}
++
++/**
++ * The bridge path: commits Live just shipped (Live filtered them against deploy_last_announced),
++ * newest first, so the first is the head. A head already announced (by the event, or a retried
++ * bridge call) announces nothing.
++ * @returns {{ announced: number, id?: number, duplicate?: boolean }}
++ */
++function announceCommits({ db, chatServer, commits, log = console }) {
++ const list = cleanCommits(commits);
++ if (!list.length) return { announced: 0 };
++ const head = list[0].hash;
++
++ let r;
++ try { r = claimRelease(db, { head, commits: list, via: 'bridge' }); }
++ catch (err) { log.warn('[Deploy notice] not saved:', err.message); return { announced: 0 }; }
++ if (r.duplicate) {
++ log.log(`[Deploy notice] ${head.slice(0, 7)} already announced (card ${r.id}); the bridge copy changes nothing`);
++ return { announced: 0, id: r.id, duplicate: true };
++ }
++ broadcast({ chatServer, saved: r.saved, list, log });
++ return { announced: list.length, id: r.saved.id };
++}
++
++/** The release a live.release.deployed envelope announces ({ head, commits }), or an 'ignored:*' outcome. No I/O. */
++function releaseFrom(event, { now = Date.now(), maxAgeMs = RELEASE_MAX_AGE_MS } = {}) {
++ if (!event || event.event_type !== 'live.release.deployed') return 'ignored:type';
++ if (event.source !== 'live') return 'ignored:source';
++ const s = event.subject && typeof event.subject === 'object' ? event.subject : {};
++ if (s.type !== 'release' || !HASH_RE.test(String(s.id || ''))) return 'ignored:subject';
++ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
++ if (p.redacted === true) return 'ignored:redacted';
++ const commits = cleanCommits(p.commits);
++ if (!commits.length) return 'ignored:payload';
++ const at = Date.parse(p.deployed_at || event.timestamp || '');
++ if (Number.isFinite(at) && now - at > maxAgeMs) return 'ignored:stale';
++ return { head: s.id, commits };
++}
++
++/**
++ * The Events path, called inside the consumer's inbox transaction (synchronous). Returns an
++ * 'ignored:*' outcome, or { outcome, detail, after } where after() broadcasts once the transaction
++ * has committed. outcome: announced (a new card) | folded (into the last card) | duplicate:release
++ * (the bridge, or an earlier event with this head, made the card already).
++ */
++function applyReleaseEvent({ db, chatServer, event, now = Date.now(), maxAgeMs = RELEASE_MAX_AGE_MS, log = console }) {
++ const rel = releaseFrom(event, { now, maxAgeMs });
++ if (typeof rel === 'string') return rel;
++ const r = claimRelease(db, { head: rel.head, commits: rel.commits, via: 'events', eventId: event.event_id, now });
++ const detail = { head: rel.head, message_id: r.duplicate ? r.id : r.saved.id, commits: rel.commits.length };
++ if (r.duplicate) return { outcome: 'duplicate:release', detail };
++ return { outcome: r.saved.folded ? 'folded' : 'announced', detail, after: () => broadcast({ chatServer, saved: r.saved, list: rel.commits, log }) };
}
/**
@@@ -97,4 +176,4 @@ function replayTo(ws) {
return false;
}
--module.exports = { replayTo, announceCommits, persist, plainText, parseLog };
++module.exports = { replayTo, announceCommits, applyReleaseEvent, releaseFrom, claimRelease, persist, plainText, parseLog, cleanCommits, RELEASE_MAX_AGE_MS };
ddiff --git a/server/config.js b/server/config.js
iindex 2407679..18bb5c6 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -84,7 +84,8 @@ module.exports = {
// (server/prefs/). Read and written with Chat's service token (grants network.modules.read and
// network.modules.write on chat.preferences, audience openvibe.network). Cached per person for
// ttlMs: Chat's own writes update the cache at once, a change made through Network directly shows
-- // within ttlMs (at once when a network.module.updated event reaches prefs.handleEvent()).
++ // at once when its network.module.updated reaches the Events consumer (prefs.handleEvent()), and
++ // within ttlMs without it.
prefs: {
enabled: bool(process.env.CHAT_PREFS_ENABLED, true),
ttlMs: int(process.env.CHAT_PREFS_TTL_MS, 60_000),
@@@ -96,6 +97,17 @@ module.exports = {
events: {
url: strip(process.env.EVENTS_URL || ''),
intervalMs: int(process.env.EVENTS_RELAY_INTERVAL_MS, 5000),
++ // The consumer (server/events/consumer.js): POST /internal/events, deliveries of Chat's
++ // subscriptions (live.release.deployed, network.module.updated), signature v2 under
++ // CHAT_EVENTS_SECRET (comma-separated for rotation; each 32+ characters, the first is the one
++ // handed to Events). Unset = the route answers 503 and no subscription is made.
++ secrets: String(process.env.CHAT_EVENTS_SECRET || '').split(',').map((s) => s.trim()).filter((s) => s.length >= 32),
++ // Create the subscriptions at boot when missing (idempotent; an existing one, even disabled,
++ // is left as it is). Needs EVENTS_URL, the secret, OV_OAUTH_CLIENT_SECRET and the Network
++ // grant chat events.subscription.manage on openvibe.events.
++ subscribe: bool(process.env.CHAT_EVENTS_SUBSCRIBE, true),
++ // Where Events delivers. Default http://127.0.0.1:<PORT>/internal/events.
++ endpoint: strip(process.env.CHAT_EVENTS_ENDPOINT || ''),
},
// Channel sound clips live on disk. At cutover this is Live's sounds directory: stored rows
ddiff --git a/server/db/schema.sql b/server/db/schema.sql
iindex e2488ba..73087a2 100644
---- a/server/db/schema.sql
++++ b/server/db/schema.sql
@@@ -464,3 +464,29 @@ CREATE TABLE IF NOT EXISTS chat_meta (
key TEXT PRIMARY KEY,
value TEXT
);
++
++-- Deploys announced in chat, one row per release: Live's head commit (live.release.deployed
++-- subject.id; the first commit of the bridge's deployNotice). The bridge and the Events consumer
++-- both claim the head here in the transaction that stores or folds the card
++-- (server/chat/deploy-notice.js), so a head never makes two cards whichever path comes first.
++-- bridge_at / event_at: when each path delivered it (ms), to see that both carry every deploy.
++CREATE TABLE IF NOT EXISTS deploy_releases (
++ head TEXT PRIMARY KEY,
++ message_id INTEGER,
++ first_via TEXT NOT NULL CHECK(first_via IN ('bridge', 'events')),
++ bridge_at INTEGER,
++ event_at INTEGER,
++ event_id TEXT,
++ commit_count INTEGER NOT NULL DEFAULT 0,
++ created_at INTEGER NOT NULL
++);
++
++-- OpenVibe.Events deliveries already applied (openvibe-sdk inbox: consumer + event id), claimed in
++-- the transaction that applies the event (server/events/consumer.js). Pruned after 35 days
++-- (Events keeps events 30).
++CREATE TABLE IF NOT EXISTS chat_event_inbox (
++ consumer TEXT NOT NULL,
++ event_id TEXT NOT NULL,
++ processed_at INTEGER NOT NULL,
++ PRIMARY KEY (consumer, event_id)
++);
ddiff --git a/server/events/consumer.js b/server/events/consumer.js
nnew file mode 100644
iindex 0000000..b7d3aca
---- /dev/null
++++ b/server/events/consumer.js
@@@ -0,0 +1,144 @@
++/**
++ * OpenVibe.Events → Chat: POST /internal/events, the endpoint of Chat's Events subscriptions
++ * (consumer `chat`, created at boot by ./subscriptions.js; scripts/subscribe-events.js lists,
++ * disables and enables them).
++ *
++ * live.release.deployed Live shipped new commits (Live server/events/release-events.js). The deploy
++ * card in global chat is stored or folded exactly as the bridge's deployNotice
++ * does (../chat/deploy-notice.js), keyed by subject.id, the head commit: both
++ * paths claim the head in deploy_releases, so while both run (compatibility
++ * register C-84) a deploy makes one card whichever arrives first. Older than
++ * 6 hours = ignored:stale.
++ * network.module.updated a user-module record changed at Network. Only chat.preferences matters: a
++ * 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.
++ *
++ * 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
++ * to browsers run only after that transaction has committed.
++ *
++ * Signature: v2 only (openvibe-sdk parseDelivery with requireV2): HMAC over "<t>.<raw body>" with t
++ * within ±300 s, under CHAT_EVENTS_SECRET (comma-separated for rotation, each 32+ characters). A
++ * v1-only, stale, forged or unsigned delivery is 401. Unset secret = 503: the route is inert until
++ * the operator sets it. Loopback only, like Chat's other internal routes: a request that came through
++ * nginx/Cloudflare (it carries a forwarding header) is refused whatever it presents.
++ */
++'use strict';
++
++const express = require('express');
++const { http } = require('openvibe-contracts');
++const { parseDelivery, createInbox } = require('openvibe-sdk/events');
++const db = require('../db/database');
++const { viaProxy } = require('../net/service-auth');
++const deployNotice = require('../chat/deploy-notice');
++const prefs = require('../prefs/chat-preferences');
++
++const CONSUMER = 'chat';
++const INBOX_TABLE = 'chat_event_inbox';
++const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated']);
++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
++
++/**
++ * @param {object} o
++ * @param {object} o.chatServer
++ * @param {string[]} o.secrets CHAT_EVENTS_SECRET values (32+ characters each)
++ */
++function createEventsConsumer({ chatServer, secrets = [], now = () => Date.now(), releaseMaxAgeMs = deployNotice.RELEASE_MAX_AGE_MS, log = console } = {}) {
++ const keys = (secrets || []).filter((s) => typeof s === 'string' && s.length >= 32);
++ const inbox = createInbox(db.getDb(), { table: INBOX_TABLE, now });
++ inbox.ensureSchema();
++ const stats = { received: 0, applied: 0, duplicates: 0, ignored: 0, refused: 0, failed: 0, last_at: null, last_type: null, last_outcome: null, last_error: null };
++
++ /**
++ * What an envelope needs, decided before the inbox: an 'ignored:*' outcome (acknowledged, not
++ * recorded), or the synchronous work to run inside the inbox transaction.
++ */
++ function plan(event) {
++ if (event.event_type === 'live.release.deployed') {
++ const rel = deployNotice.releaseFrom(event, { now: now(), maxAgeMs: releaseMaxAgeMs });
++ if (typeof rel === 'string') return rel;
++ return () => deployNotice.applyReleaseEvent({ db, chatServer, event, now: now(), maxAgeMs: releaseMaxAgeMs, log });
++ }
++ if (event.event_type === 'network.module.updated') {
++ if (event.source !== 'network') return 'ignored:source';
++ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
++ if (p.namespace !== prefs.NAMESPACE) return 'ignored:namespace';
++ return () => (prefs.handleEvent(event) ? 'invalidated' : 'unchanged');
++ }
++ return 'ignored:type';
++ }
++
++ /** Apply one envelope. Returns { duplicate, outcome, detail? }. Throws only on a storage failure. */
++ function apply(event) {
++ const work = plan(event);
++ if (typeof work === 'string') return { duplicate: false, outcome: work };
++ let after = null;
++ const r = inbox.once(CONSUMER, event.event_id, () => {
++ const out = work();
++ if (typeof out === 'string') return { outcome: out };
++ after = out.after || null;
++ return { outcome: out.outcome, detail: out.detail };
++ });
++ if (r.duplicate) return { duplicate: true, outcome: null };
++ // Side effects outside the database run only after the commit, once per event.
++ if (after) { try { after(); } catch (err) { log.warn('[Events consumer] after-commit step failed:', err.message); } }
++ return { duplicate: false, outcome: r.result.outcome, ...(r.result.detail ? { detail: r.result.detail } : {}) };
++ }
++
++ const router = express.Router();
++ router.post('/', express.raw({ type: () => true, limit: '256kb' }), (req, res) => {
++ const ctx = http.requestContext(req.headers);
++ const problem = (status, code, detail) => http.sendProblem(res, status, code, { detail, ctx });
++ if (viaProxy(req)) return problem(403, 'chat.internal_only', 'internal route');
++ if (!keys.length) return problem(503, 'chat.webhook_disabled', 'CHAT_EVENTS_SECRET is not set');
++ const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
++ let delivery = null;
++ // Signature v2 only: a v1-only (v2 stripped), stale or forged delivery is refused.
++ for (const s of keys) { delivery = parseDelivery(raw, req.headers, s, { requireV2: true, now: now() }); if (delivery) break; }
++ if (!delivery) { stats.refused++; return problem(401, 'chat.bad_signature', 'X-OpenVibe-Signature-V2 does not verify or is outside the replay window'); }
++ const event = delivery.event;
++ if (!event || typeof event !== 'object' || typeof event.event_id !== 'string' || !EVENT_ID_RE.test(event.event_id) || typeof event.event_type !== 'string') {
++ stats.refused++;
++ return problem(400, 'chat.bad_delivery', 'body must be { event: <envelope>, seq }');
++ }
++ stats.received++;
++ stats.last_at = new Date(now()).toISOString();
++ stats.last_type = event.event_type;
++ let out;
++ try {
++ out = apply(event);
++ } catch (err) {
++ // Not acknowledged: the inbox claim rolled back with the change, and Events retries.
++ stats.failed++;
++ stats.last_error = String(err.message || err).slice(0, 200);
++ log.error(`[Events consumer] ${event.event_id} (${event.event_type}) failed:`, err.message);
++ return problem(500, 'chat.event_failed', 'processing failed; it will be retried');
++ }
++ if (out.duplicate) stats.duplicates++;
++ else if (String(out.outcome).startsWith('ignored:')) stats.ignored++;
++ else stats.applied++;
++ stats.last_outcome = out.duplicate ? 'duplicate' : out.outcome;
++ if (event.event_type === 'live.release.deployed' && !out.duplicate) {
++ log.log(`[Events consumer] ${event.event_id} live.release.deployed ${String(event.subject && event.subject.id).slice(0, 7)}: ${out.outcome}`);
++ }
++ res.status(200).json({ event_id: event.event_id, duplicate: out.duplicate, outcome: out.outcome, ...(out.detail ? { detail: out.detail } : {}) });
++ });
++
++ /** Drop inbox receipts older than Events' retention (nothing older can be redelivered). */
++ function prune() {
++ return db.run(`DELETE FROM ${INBOX_TABLE} WHERE processed_at < ?`, [now() - INBOX_KEEP_MS]).changes;
++ }
++ let timer = null;
++ function start() {
++ if (timer) return;
++ timer = setInterval(() => { try { prune(); } catch (err) { log.warn('[Events consumer] inbox prune:', err.message); } }, 6 * 3600 * 1000);
++ if (timer.unref) timer.unref();
++ }
++ function stop() { if (timer) clearInterval(timer); timer = null; }
++
++ return { router, apply, prune, start, stop, stats, enabled: keys.length > 0 };
++}
++
++module.exports = { createEventsConsumer, CONSUMER, TOPICS, INBOX_TABLE };
ddiff --git a/server/events/subscriptions.js b/server/events/subscriptions.js
nnew file mode 100644
iindex 0000000..8607387
---- /dev/null
++++ b/server/events/subscriptions.js
@@@ -0,0 +1,133 @@
++/**
++ * Chat's OpenVibe.Events subscriptions: one per topic of ./consumer.js TOPICS (live.release.deployed,
++ * network.module.updated), all delivering to POST /internal/events, signed with the first
++ * CHAT_EVENTS_SECRET. Events names the consumer after the calling service (`chat`) and refuses a
++ * second subscription with the same topic and endpoint (409 with its id), so creating is idempotent.
++ *
++ * ensure({ action: 'create' }) what boot runs (startAtBoot): lists Chat's subscriptions and creates
++ * the missing ones. An existing one is left as it is, even disabled,
++ * so an operator's --disable (the rollback) survives restarts.
++ * ensure({ action: 'list' }) report only (scripts/subscribe-events.js --dry-run)
++ * ensure({ action: 'disable' | 'enable' }) the rollback switch and its undo
++ *
++ * Needs EVENTS_URL, OV_OAUTH_CLIENT_SECRET and the Network grant chat events.subscription.manage on
++ * openvibe.events. Nothing secret is logged. A new subscription gets no history (Events replays with
++ * POST /api/v1/deliveries/replay, events.delivery.admin; the inbox makes a replayed event a no-op).
++ */
++'use strict';
++
++const { serviceAuth } = require('openvibe-contracts');
++const { TOPICS } = require('./consumer');
++
++const AUDIENCE = 'openvibe.events';
++const SCOPE = 'events.subscription.manage';
++const BOOT_RETRY_MS = [0, 10_000, 60_000, 5 * 60_000, 15 * 60_000];
++
++const normalize = (u) => { try { return new URL(String(u)).toString(); } catch { return String(u || ''); } };
++
++/** The endpoint Events delivers to: CHAT_EVENTS_ENDPOINT, else loopback on the port Chat listens on. */
++function endpointFor(config, port = config.port) {
++ return config.events.endpoint || `http://127.0.0.1:${port}/internal/events`;
++}
++
++async function readJson(res) {
++ const text = await res.text().catch(() => '');
++ try { return text ? JSON.parse(text) : {}; } catch { return { detail: text.slice(0, 200) }; }
++}
++
++/**
++ * @returns {Promise<Array<{ topic, id, result: 'exists'|'created'|'missing'|'disabled'|'enabled', enabled }>>}
++ */
++async function ensure({ config, endpoint, action = 'create', topics = TOPICS, fetchImpl = globalThis.fetch, tokenClient = null }) {
++ const base = config.events.url;
++ const secret = config.events.secrets[0];
++ if (!base) throw new Error('EVENTS_URL is not set');
++ if (action === 'create' && !secret) throw new Error('CHAT_EVENTS_SECRET must be set (32+ characters) before subscribing');
++ if (!config.oauth.clientSecret) throw new Error('OV_OAUTH_CLIENT_SECRET is not set');
++ const tokens = tokenClient || serviceAuth.createTokenClient({
++ tokenUrl: `${config.networkInternalUrl}/oauth/token`,
++ clientId: config.oauth.clientId,
++ clientSecret: config.oauth.clientSecret,
++ audience: AUDIENCE,
++ scope: SCOPE,
++ fetchImpl,
++ });
++ const target = normalize(endpoint);
++ const call = async (method, path, body) => {
++ const res = await fetchImpl(`${base}${path}`, {
++ method,
++ headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await tokens.authHeaders()) },
++ body: body ? JSON.stringify(body) : undefined,
++ signal: AbortSignal.timeout(15000),
++ });
++ if (res.status === 401) tokens.invalidate();
++ return { status: res.status, ok: res.ok, body: await readJson(res) };
++ };
++
++ const listed = await call('GET', '/api/v1/subscriptions');
++ if (!listed.ok) throw new Error(`Events answered ${listed.status} listing subscriptions: ${listed.body.code || ''} ${listed.body.detail || ''}`.trim());
++ const mine = (listed.body.subscriptions || []).filter((s) => normalize(s.endpoint) === target);
++
++ const out = [];
++ for (const topic of topics) {
++ const existing = mine.find((s) => s.topic_pattern === topic) || null;
++ if (action === 'list') {
++ out.push({ topic, id: existing ? existing.id : null, result: existing ? 'exists' : 'missing', enabled: existing ? existing.enabled !== false : null });
++ continue;
++ }
++ if (action === 'disable' || action === 'enable') {
++ if (!existing) { out.push({ topic, id: null, result: 'missing', enabled: null }); continue; }
++ const r = await call('POST', `/api/v1/subscriptions/${encodeURIComponent(existing.id)}/${action}`);
++ if (!r.ok) throw new Error(`Events answered ${r.status} to ${action} ${existing.id}: ${r.body.code || ''} ${r.body.detail || ''}`.trim());
++ out.push({ topic, id: existing.id, result: `${action}d`, enabled: action === 'enable' });
++ continue;
++ }
++ if (existing) { out.push({ topic, id: existing.id, result: 'exists', enabled: existing.enabled !== false }); continue; }
++ const r = await call('POST', '/api/v1/subscriptions', { topic_pattern: topic, endpoint: target, secret });
++ if (r.status === 409 && r.body.subscription_id) { out.push({ topic, id: r.body.subscription_id, result: 'exists', enabled: null }); continue; }
++ if (!r.ok) throw new Error(`Events answered ${r.status} for ${topic}: ${r.body.code || ''} ${r.body.detail || ''}`.trim());
++ out.push({ topic, id: r.body.id, result: 'created', enabled: r.body.enabled !== false });
++ }
++ return out;
++}
++
++/**
++ * At boot, in the background: make sure the subscriptions exist, retrying while Network or Events
++ * cannot answer (a restart of either, the grant not seeded yet). Returns { done: Promise, stop() }.
++ * Does nothing (done resolves null) unless EVENTS_URL, CHAT_EVENTS_SECRET and the client secret are
++ * set and CHAT_EVENTS_SUBSCRIBE is on.
++ */
++function startAtBoot({ config, port, fetchImpl, log = console, retryMs = BOOT_RETRY_MS }) {
++ let stopped = false;
++ let timer = null;
++ const why = !config.events.url ? 'EVENTS_URL unset' : !config.events.secrets.length ? 'CHAT_EVENTS_SECRET unset'
++ : !config.oauth.clientSecret ? 'OV_OAUTH_CLIENT_SECRET unset' : !config.events.subscribe ? 'CHAT_EVENTS_SUBSCRIBE=0' : null;
++ if (why) {
++ log.log(`[Events] subscriptions not checked (${why})`);
++ return { done: Promise.resolve(null), stop() {} };
++ }
++ const endpoint = endpointFor(config, port);
++ let finish = () => {};
++ const done = new Promise((resolve) => {
++ finish = resolve;
++ const attempt = async (i) => {
++ if (stopped) return resolve(null);
++ try {
++ const results = await ensure({ config, endpoint, action: 'create', fetchImpl });
++ for (const r of results) log.log(`[Events] subscription ${r.result}: ${r.id} (${r.topic} → ${endpoint}${r.enabled === false ? ', DISABLED' : ''})`);
++ return resolve(results);
++ } catch (err) {
++ const next = retryMs[i + 1];
++ log.warn(`[Events] subscriptions not ensured: ${err.message}${next != null ? `; retrying in ${Math.round(next / 1000)} s` : '; giving up until the next restart (or run scripts/subscribe-events.js)'}`);
++ if (next == null || stopped) return resolve(null);
++ timer = setTimeout(() => attempt(i + 1), next);
++ if (timer.unref) timer.unref();
++ }
++ };
++ timer = setTimeout(() => attempt(0), retryMs[0] || 0);
++ if (timer.unref) timer.unref();
++ });
++ return { done, stop() { stopped = true; if (timer) clearTimeout(timer); finish(null); } };
++}
++
++module.exports = { ensure, startAtBoot, endpointFor, TOPICS, AUDIENCE, SCOPE };
ddiff --git a/server/index.js b/server/index.js
iindex 64fe4e9..8edfd36 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -4,8 +4,10 @@
* node server/index.js (PORT 4400; systemd unit deploy/systemd/openvibe-chat.service)
*
* Boot: database → Live projections (first full sync; Chat serves with whatever it has if Live is
-- * slow) → chat WebSocket server → HTTP app → background loops (Live sync, events outbox relay,
-- * Live mirror relay). SIGTERM tells clients the chat is restarting, closes sockets and exits.
++ * slow) → chat WebSocket server → HTTP app (with the Events consumer, POST /internal/events) →
++ * background loops (Live sync, events outbox relay, Live mirror relay) → once listening, Chat's
++ * Events subscriptions are checked and created if missing (in the background, retried). SIGTERM
++ * tells clients the chat is restarting, closes sockets and exits.
*/
'use strict';
@@@ -17,10 +19,12 @@ const chatServer = require('./chat/chat-server');
const { createBridge } = require('./bridge/live-bridge');
const { createMirror } = require('./bridge/live-mirror');
const { createRelay } = require('./events/outbox');
++const { createEventsConsumer } = require('./events/consumer');
++const subscriptions = require('./events/subscriptions');
const { createApp } = require('./app');
async function start() {
-- console.log(`[Chat] OpenVibe.Chat starting (db ${config.dbPath}, Live ${config.live.internalUrl}, mirror ${config.live.mirror ? 'on' : 'off'}, events ${config.events.url || 'off'})`);
++ console.log(`[Chat] OpenVibe.Chat starting (db ${config.dbPath}, Live ${config.live.internalUrl}, mirror ${config.live.mirror ? 'on' : 'off'}, events ${config.events.url || 'off'}, consumer ${config.events.secrets.length ? 'on' : 'off'})`);
// Mirror capture only when the mirror is on: before the cutover (rehearsals) nothing queues.
db.initDb({ captureMirror: config.live.mirror });
@@@ -33,14 +37,17 @@ async function start() {
const bridge = createBridge({ chatServer });
const mirror = createMirror({ config });
const relay = createRelay({ config });
-- const { app, handleUpgrade } = createApp({ chatServer, bridge, mirror, relay });
++ const events = createEventsConsumer({ chatServer, secrets: config.events.secrets });
++ const { app, handleUpgrade } = createApp({ chatServer, bridge, mirror, relay, events });
server.on('request', app);
server.on('upgrade', (req, socket, head) => { handleUpgrade(req, socket, head).catch(() => { try { socket.destroy(); } catch { /* */ } }); });
mirror.start();
relay.start();
++ events.start();
await new Promise((resolve) => server.listen(config.port, config.host, resolve));
console.log(`[Chat] listening on http://${config.host}:${server.address().port} (ws /ws/chat)`);
++ const subs = subscriptions.startAtBoot({ config, port: server.address().port });
let stopping = false;
const shutdown = async () => {
@@@ -57,6 +64,8 @@ async function start() {
ctx.stop();
relay.stop();
mirror.stop();
++ events.stop();
++ subs.stop();
try { await Promise.race([mirror.flush(), new Promise((r) => setTimeout(r, 3000))]); } catch { /* */ }
try { chatServer.close(); } catch { /* */ }
server.close(() => { db.close(); process.exit(0); });
@@@ -64,7 +73,7 @@ async function start() {
};
process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);
-- return { server, mirror, relay, bridge, shutdown };
++ return { server, mirror, relay, bridge, events, subscriptions: subs, shutdown };
}
if (require.main === module) {
ddiff --git a/server/prefs/chat-preferences.js b/server/prefs/chat-preferences.js
iindex 9ab880e..0ceec46 100644
---- a/server/prefs/chat-preferences.js
++++ b/server/prefs/chat-preferences.js
@@@ -13,8 +13,9 @@
* strict instead: 412 when the record moved. A patch that changes nothing
* writes nothing.
* handleEvent(envelope) network.module.updated for chat.preferences newer than the cached copy
-- * drops it at once (for when Events deliveries reach Chat; until then the TTL
-- * bounds how long a change made outside Chat takes to show).
++ * drops it at once. Called by the Events consumer (server/events/consumer.js,
++ * Chat's subscription to network.module.updated); without that subscription
++ * the TTL bounds how long a change made outside Chat takes to show.
*
* Only usr_ subjects: Network keeps modules for accounts, and Chat learns a person's subject from Live.
*/
ddiff --git a/test/events-consumer.test.js b/test/events-consumer.test.js
nnew file mode 100644
iindex 0000000..315643a
---- /dev/null
++++ b/test/events-consumer.test.js
@@@ -0,0 +1,367 @@
++'use strict';
++/**
++ * OpenVibe.Events → Chat (server/events/consumer.js, subscriptions.js; compatibility register C-84):
++ *
++ * - POST /internal/events takes signature v2 only: a bad signature, a timestamp outside ±300 s and a
++ * v1-only delivery are 401; no secret is 503; a forwarded request is 403
++ * - live.release.deployed makes the deploy card exactly as the bridge's deployNotice does; the inbox
++ * makes a redelivered event a no-op (one card)
++ * - a bridge notice and an event for the same head are one card, whichever comes first
++ * - folding as before: a later deploy folds into the last card, a message in a stream room in
++ * between starts a new card; the two paths fold into each other's cards
++ * - network.module.updated drops a cached chat.preferences copy when newer, ignores an older revision
++ * - boot creates the subscriptions once (idempotent), never re-enables a disabled one, and the
++ * secret it hands Events is the one the consumer verifies
++ *
++ * A stub OpenVibe.Events (below) checks Chat's token (audience openvibe.events,
++ * events.subscription.manage) and keeps subscriptions the way Events does (409 on a duplicate).
++ */
++const assert = require('assert');
++const http = require('http');
++const { ids } = require('openvibe-contracts');
++const { signDeliveryHeaders } = require('openvibe-sdk/events');
++const { boot, suite } = require('./helpers');
++
++const t = suite('events-consumer');
++const SECRET = 'e'.repeat(64);
++const OTHER = 'f'.repeat(64);
++const ANN = 'usr_01J9ANN0000000000000000AAA';
++let h, BRIDGE, streamer, viewer, streamId, prefs;
++let seq = 0;
++let n = 0;
++
++// ── Stub OpenVibe.Events: subscriptions (token-checked, 409 on a duplicate) and publishing ──
++const stubEvents = { subs: [], posts: 0 };
++/** Chat's token: audience openvibe.events, events.subscription.manage (null = accepted). */
++stubEvents.verify = (req) => {
++ if (!h) return { status: 503, code: 'not_ready' };
++ const { serviceAuth } = require('openvibe-contracts');
++ const auth = String(req.headers.authorization || '');
++ const v = auth.startsWith('Bearer ') ? serviceAuth.verifyServiceToken(auth.slice(7), { publicKey: h.keys.publicKey, issuer: h.ISS, audience: 'openvibe.events' }) : { ok: false, code: 'token.missing' };
++ if (!v.ok) return { status: 401, code: v.code };
++ if (!(v.claims.cap || []).includes('events.subscription.manage')) return { status: 403, code: 'capability.denied' };
++ return null;
++};
++const eventsServer = http.createServer(async (req, res) => {
++ const raw = await new Promise((r) => { let s = ''; req.on('data', (c) => { s += c; }); req.on('end', () => r(s)); });
++ const send = (status, body) => { res.statusCode = status; res.setHeader('Content-Type', 'application/json'); res.end(JSON.stringify(body)); };
++ const url = new URL(req.url, 'http://events');
++ if (req.method === 'POST' && url.pathname === '/api/v1/events') return send(200, { results: [] }); // the outbox relay
++ const bad = stubEvents.verify(req);
++ if (bad) return send(bad.status, { code: bad.code });
++ let m;
++ if (req.method === 'GET' && url.pathname === '/api/v1/subscriptions') {
++ return send(200, { subscriptions: stubEvents.subs.map(({ secret, ...s }) => s) });
++ }
++ if (req.method === 'POST' && url.pathname === '/api/v1/subscriptions') {
++ stubEvents.posts++;
++ const b = JSON.parse(raw || '{}');
++ const endpoint = new URL(b.endpoint).toString();
++ const dup = stubEvents.subs.find((s) => s.topic_pattern === b.topic_pattern && s.endpoint === endpoint);
++ if (dup) return send(409, { code: 'events.subscription_exists', subscription_id: dup.id });
++ const sub = { id: `sub_${ids.ulid()}`, consumer: 'chat', topic_pattern: b.topic_pattern, endpoint, enabled: true, secret: b.secret };
++ stubEvents.subs.push(sub);
++ return send(201, { ...sub });
++ }
++ if (req.method === 'POST' && (m = /^\/api\/v1\/subscriptions\/([^/]+)\/(disable|enable)$/.exec(url.pathname))) {
++ const sub = stubEvents.subs.find((s) => s.id === m[1]);
++ if (!sub) return send(404, { code: 'events.not_found' });
++ sub.enabled = m[2] === 'enable';
++ const { secret, ...view } = sub;
++ return send(200, view);
++ }
++ send(404, { code: 'not_found' });
++});
++
++const commit = (tag, subject = `change ${tag}`) => ({ hash: tag.repeat(40).slice(0, 40), short: tag.repeat(7).slice(0, 7), date: '2026-09-23T10:00:00Z', subject });
++/** A live.release.deployed envelope as Live's server/events/release-events.js builds it. */
++function release(commits, { eventId = ids.newId('event'), deployedAt = new Date().toISOString() } = {}) {
++ const head = commits[0].hash;
++ return {
++ event_id: eventId, event_type: 'live.release.deployed', version: 1, source: 'live',
++ actor: { type: 'service', id: 'live' }, timestamp: deployedAt, priority: 'low', visibility: 'internal',
++ subject: { type: 'release', id: head },
++ payload: {
++ service: 'live', release: commits[0].short, commit: head, previous: null, commit_count: commits.length,
++ commits: commits.map((c) => ({ hash: c.hash, short: c.short, subject: c.subject, date: c.date })),
++ deployed_at: deployedAt, notes_url: 'https://openvibe.live/updates',
++ },
++ };
++}
++function moduleUpdated(revision, { namespace = 'chat.preferences', owner = ANN, source = 'network', eventId = ids.newId('event') } = {}) {
++ return {
++ event_id: eventId, event_type: 'network.module.updated', version: 1, source, actor: { type: 'user', id: owner },
++ timestamp: new Date().toISOString(), priority: 'important', visibility: 'internal',
++ subject: { type: 'user_module', id: `${owner}:${namespace}`, revision },
++ payload: { owner: { type: 'user', id: owner }, namespace, namespace_owner: 'chat', schema_version: 1, revision, change: 'updated', reason: 'write', keys: ['timestamps'] },
++ };
++}
++
++/** POST a delivery the way Events does (all three signature headers), or with changes. */
++function deliver(event, { secret = SECRET, now = Date.now(), headers = {}, strip = [] } = {}) {
++ const raw = JSON.stringify({ event, seq: ++seq });
++ const signed = signDeliveryHeaders(raw, secret, { now });
++ for (const k of strip) delete signed[k];
++ return h.http('POST', '/internal/events', { raw, headers: { 'Content-Type': 'application/json', 'X-OpenVibe-Event-Id': event.event_id, ...signed, ...headers } });
++}
++const bridge = (commits) => h.http('POST', '/internal/live/calls', { token: BRIDGE, body: { boot: 'live-boot-1', ops: [{ seq: ++n, op: 'deployNotice', args: [commits] }] } });
++const cards = () => h.db.all("SELECT id, metadata FROM chat_messages WHERE message_type = 'system' AND metadata LIKE '%\"kind\":\"deploy\"%' ORDER BY id").map((r) => ({ id: r.id, ...JSON.parse(r.metadata) }));
++const card = (id) => cards().find((c) => c.id === id);
++const speak = (text) => h.db.saveChatMessage({ stream_id: streamId, user_id: viewer.id, username: viewer.username, message: text, message_type: 'chat', is_global: false });
++const releaseRow = (head) => h.db.get('SELECT * FROM deploy_releases WHERE head = ?', [head]);
++
++t('boot with the consumer on and a stub Events', async () => {
++ const eventsPort = await new Promise((r) => eventsServer.listen(0, '127.0.0.1', () => r(eventsServer.address().port)));
++ h = await boot({ env: { CHAT_EVENTS_SECRET: `${SECRET}, short-is-ignored`, EVENTS_URL: `http://127.0.0.1:${eventsPort}`, CHAT_PREFS_TTL_MS: '60000' } });
++ BRIDGE = h.serviceToken(['chat.live_bridge.write', 'chat.message.send']);
++ streamer = h.addUser('streamer', { role: 'streamer' });
++ viewer = h.addUser('viewer');
++ streamId = h.addStream(streamer.id, h.addChannel(streamer.id));
++ await h.ctx.sync();
++ prefs = require('../server/prefs/chat-preferences');
++});
++
++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);
++ 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');
++ }
++ const tok = h.tokenRequests.find((x) => x.audience === 'openvibe.events');
++ assert.deepStrictEqual(tok, { audience: 'openvibe.events', scope: 'events.subscription.manage', client: 'chat' });
++});
++
++t('a second boot changes nothing, and never re-enables a subscription an operator disabled', async () => {
++ const config = require('../server/config');
++ const subs = require('../server/events/subscriptions');
++ const 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.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.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']);
++});
++
++t('signature v2 only: bad signature, stale or future timestamp, v1-only, unsigned → 401', async () => {
++ const ev = release([commit('0')]);
++ let r = await deliver(ev, { secret: OTHER });
++ assert.strictEqual(r.status, 401, r.text);
++ assert.strictEqual(r.body.code, 'chat.bad_signature');
++ r = await deliver(ev, { now: Date.now() - 6 * 60 * 1000 });
++ assert.strictEqual(r.status, 401, 'a timestamp older than 300 s is refused (a replayed capture)');
++ r = await deliver(ev, { now: Date.now() + 6 * 60 * 1000 });
++ assert.strictEqual(r.status, 401, 'and one from the future');
++ r = await deliver(ev, { strip: ['X-OpenVibe-Signature-V2'] });
++ assert.strictEqual(r.status, 401, 'v1 alone is not enough');
++ r = await h.http('POST', '/internal/events', { raw: JSON.stringify({ event: ev, seq: 1 }), headers: { 'Content-Type': 'application/json' } });
++ assert.strictEqual(r.status, 401);
++ // A v2 header that names another time than X-OpenVibe-Timestamp.
++ r = await deliver(ev, { headers: { 'X-OpenVibe-Timestamp': String(Math.floor(Date.now() / 1000) - 5) } });
++ assert.strictEqual(r.status, 401);
++ assert.strictEqual(cards().length, 0, 'nothing was applied');
++ assert.strictEqual(h.db.get('SELECT COUNT(*) AS n FROM chat_event_inbox').n, 0);
++ // Signed but not an envelope.
++ r = await deliver({ event_type: 'live.release.deployed' });
++ assert.strictEqual(r.status, 400);
++ // Through nginx: refused whatever it carries.
++ r = await deliver(ev, { headers: { 'X-Forwarded-For': '203.0.113.9' } });
++ assert.strictEqual(r.status, 403);
++ // A timestamp within the window verifies.
++ r = await deliver(release([commit('8')]), { now: Date.now() - 200 * 1000 });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.body.outcome, 'announced');
++});
++
++t('no CHAT_EVENTS_SECRET: the route answers 503', async () => {
++ const express = require('express');
++ const { createEventsConsumer } = require('../server/events/consumer');
++ const off = createEventsConsumer({ chatServer: h.chatServer, secrets: ['too-short'] });
++ assert.strictEqual(off.enabled, false);
++ const app = express();
++ app.use('/internal/events', off.router);
++ const srv = await new Promise((r) => { const s = app.listen(0, '127.0.0.1', () => r(s)); });
++ const raw = JSON.stringify({ event: release([commit('9')]), seq: 1 });
++ const res = await fetch(`http://127.0.0.1:${srv.address().port}/internal/events`, { method: 'POST', headers: { 'Content-Type': 'application/json', ...signDeliveryHeaders(raw, 'too-short') }, body: raw });
++ assert.strictEqual(res.status, 503);
++ assert.strictEqual((await res.json()).code, 'chat.webhook_disabled');
++ srv.close();
++});
++
++t('live.release.deployed makes the deploy card; the same event again is one card (inbox)', async () => {
++ speak('before the first card');
++ const ev = release([commit('a', 'events: first'), commit('b', 'events: second')]);
++ let r = await deliver(ev);
++ assert.strictEqual(r.status, 200, r.text);
++ assert.deepStrictEqual([r.body.duplicate, r.body.outcome], [false, 'announced']);
++ const id = r.body.detail.message_id;
++ const c = card(id);
++ assert.deepStrictEqual(c.commits.map((x) => x.subject), ['events: first', 'events: second']);
++ assert.strictEqual(c.deploys, 1);
++ const row = h.db.getChatMessageById(id);
++ assert.strictEqual(row.message_type, 'system');
++ assert.strictEqual(row.is_global, 1);
++ assert.strictEqual(row.username, 'OpenVibe.Live');
++ assert.strictEqual(row.message, '🚀 2 updates shipped: events: first · events: second');
++ const before = cards().length;
++ r = await deliver(ev); // Events redelivers (a lost acknowledgement)
++ assert.deepStrictEqual([r.status, r.body.duplicate, r.body.outcome], [200, true, null]);
++ r = await deliver(ev, { now: Date.now() + 1000 }); // a later attempt, new signature
++ assert.strictEqual(r.body.duplicate, true);
++ assert.strictEqual(cards().length, before, 'still one card');
++ assert.strictEqual(card(id).deploys, 1, 'not folded into itself');
++ // Late joiners get this announcement once, keyed by the row id (as with the bridge).
++ const late = await h.ws({ ip: '198.51.100.70' });
++ late.sendJson({ type: 'join' });
++ const upd = await late.next((m) => m.type === 'update' && m.kind === 'deploy', 4000);
++ assert.strictEqual(upd.id, id);
++ assert.deepStrictEqual(upd.fresh, [commit('a').hash, commit('b').hash]);
++ late.close();
++ const rel = releaseRow(commit('a').hash);
++ assert.deepStrictEqual([rel.first_via, rel.message_id, rel.event_id, rel.bridge_at], ['events', id, ev.event_id, null]);
++});
++
++t('bridge first, then the event for the same head: one card', async () => {
++ speak('a stream message, so the next deploy is a new card');
++ const commits = [commit('c', 'bridge: head'), commit('d', 'bridge: parent')];
++ const b = await bridge(commits);
++ const id = b.body.results[0].result.id;
++ assert.strictEqual(b.body.results[0].result.announced, 2);
++ const before = cards().length;
++ const r = await deliver(release(commits));
++ assert.deepStrictEqual([r.status, r.body.duplicate, r.body.outcome], [200, false, 'duplicate:release']);
++ assert.strictEqual(r.body.detail.message_id, id);
++ assert.strictEqual(cards().length, before, 'no second card');
++ assert.strictEqual(card(id).deploys, 1, 'the card was not folded again');
++ const rel = releaseRow(commits[0].hash);
++ assert.strictEqual(rel.first_via, 'bridge');
++ assert.ok(rel.bridge_at && rel.event_at, 'both paths are recorded as having delivered it');
++});
++
++t('the event first, then the bridge for the same head: one card', async () => {
++ speak('another stream message');
++ const commits = [commit('e', 'event: head')];
++ const r = await deliver(release(commits));
++ assert.strictEqual(r.body.outcome, 'announced');
++ const id = r.body.detail.message_id;
++ const before = cards().length;
++ const b = await bridge(commits);
++ assert.deepStrictEqual(b.body.results[0].result, { announced: 0, id, duplicate: true });
++ // A retried bridge op (Live's durable queue after a lost response) is the same.
++ const again = await bridge(commits);
++ assert.strictEqual(again.body.results[0].result.duplicate, true);
++ assert.strictEqual(cards().length, before);
++ assert.strictEqual(card(id).deploys, 1);
++ const rel = releaseRow(commits[0].hash);
++ assert.deepStrictEqual([rel.first_via, !!rel.event_at, !!rel.bridge_at], ['events', true, true]);
++});
++
++t('folding as before: the next deploy folds into the last card; a stream message in between starts a new card', async () => {
++ speak('new card next');
++ let r = await deliver(release([commit('1', 'fold: one')]));
++ const first = r.body.detail.message_id;
++ assert.strictEqual(r.body.outcome, 'announced');
++ // Nobody spoke: the next deploy (from either path) folds into that card.
++ const b = await bridge([commit('2', 'fold: two')]);
++ assert.strictEqual(b.body.results[0].result.id, first, 'the bridge folds into the card the event made');
++ r = await deliver(release([commit('3', 'fold: three')]));
++ assert.deepStrictEqual([r.body.outcome, r.body.detail.message_id], ['folded', first]);
++ assert.strictEqual(card(first).deploys, 3);
++ assert.deepStrictEqual(card(first).commits.map((c) => c.subject), ['fold: three', 'fold: two', 'fold: one']);
++ r = await deliver(release([commit('2', 'fold: two')]));
++ assert.strictEqual(r.body.outcome, 'duplicate:release', 'the bridge already announced that head');
++ assert.strictEqual(card(first).deploys, 3);
++ // Someone speaks in a stream room (the global feed shows it): the next deploy is a new card.
++ speak('hello brother');
++ r = await deliver(release([commit('4', 'fold: four')]));
++ assert.strictEqual(r.body.outcome, 'announced');
++ const next = r.body.detail.message_id;
++ assert.notStrictEqual(next, first);
++ assert.strictEqual(card(next).deploys, 1);
++ assert.strictEqual(card(first).deploys, 3, 'the old card is left as it was');
++});
++
++t('what a release event must be: from Live, a release subject, commits, not stale', async () => {
++ const before = cards().length;
++ const ev = release([commit('5')]);
++ let r = await deliver({ ...ev, event_id: ids.newId('event'), source: 'media' });
++ assert.strictEqual(r.body.outcome, 'ignored:source');
++ r = await deliver({ ...ev, event_id: ids.newId('event'), subject: { type: 'release', id: 'not-a-sha' } });
++ assert.strictEqual(r.body.outcome, 'ignored:subject');
++ r = await deliver({ ...ev, event_id: ids.newId('event'), payload: { ...ev.payload, commits: [] } });
++ assert.strictEqual(r.body.outcome, 'ignored:payload');
++ r = await deliver(release([commit('6')], { deployedAt: new Date(Date.now() - 7 * 3600e3).toISOString() }));
++ assert.strictEqual(r.body.outcome, 'ignored:stale', 'an operator replay of an old deploy says nothing');
++ r = await deliver({ ...ev, event_id: ids.newId('event'), event_type: 'live.stream.started' });
++ assert.strictEqual(r.body.outcome, 'ignored:type');
++ assert.strictEqual(cards().length, before);
++ assert.strictEqual(releaseRow(commit('6').hash), undefined);
++});
++
++t('network.module.updated drops a cached chat.preferences copy when newer; an older revision is ignored', async () => {
++ const ann = h.addUser('ann', { subject: ANN });
++ h.netModules.subjects.add(ANN);
++ h.netModules.set('chat.preferences', ANN, { timestamps: true }); // revision 1
++ h.netModules.set('chat.preferences', ANN, { timestamps: true, compact: true }); // revision 2
++ let g = await h.http('GET', '/api/chat/preferences', { token: ann.token });
++ assert.deepStrictEqual([g.status, g.body.revision], [200, 2], g.text);
++ // Changed at Network directly (the account page): Chat still serves its cached copy...
++ h.netModules.set('chat.preferences', ANN, { timestamps: false, compact: true }); // revision 3
++ g = await h.http('GET', '/api/chat/preferences', { token: ann.token });
++ assert.strictEqual(g.body.revision, 2, 'cached for CHAT_PREFS_TTL_MS');
++ // ...an older or equal revision (a late or reordered delivery, or Chat's own write) changes nothing...
++ let r = await deliver(moduleUpdated(1));
++ assert.deepStrictEqual([r.status, r.body.outcome], [200, 'unchanged']);
++ r = await deliver(moduleUpdated(2));
++ assert.strictEqual(r.body.outcome, 'unchanged');
++ assert.strictEqual(prefs._cached(ANN).revision, 2);
++ // ...other namespaces and other sources are acknowledged and not recorded...
++ const inbox = h.db.get('SELECT COUNT(*) AS n FROM chat_event_inbox').n;
++ r = await deliver(moduleUpdated(9, { namespace: 'tools.usage' }));
++ assert.strictEqual(r.body.outcome, 'ignored:namespace');
++ r = await deliver(moduleUpdated(9, { source: 'live' }));
++ assert.strictEqual(r.body.outcome, 'ignored:source');
++ assert.strictEqual(h.db.get('SELECT COUNT(*) AS n FROM chat_event_inbox').n, inbox);
++ assert.strictEqual(prefs._cached(ANN).revision, 2);
++ // ...and the newer revision drops it at once: the next read is Network's.
++ const ev = moduleUpdated(3);
++ r = await deliver(ev);
++ assert.strictEqual(r.body.outcome, 'invalidated');
++ assert.strictEqual(prefs._cached(ANN), null);
++ g = await h.http('GET', '/api/chat/preferences', { token: ann.token });
++ assert.deepStrictEqual([g.body.revision, g.body.preferences], [3, { timestamps: false, compact: true }]);
++ r = await deliver(ev);
++ assert.strictEqual(r.body.duplicate, true, 'a redelivery is a no-op');
++ assert.strictEqual(prefs._cached(ANN).revision, 3);
++});
++
++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');
++ const r = await deliver(release([commit('7', 'through the subscription')]), { secret: sub.secret, headers: { 'X-OpenVibe-Subscription-Id': sub.id } });
++ assert.deepStrictEqual([r.status, r.body.outcome], [200, 'announced']);
++ const ready = await h.http('GET', '/ready');
++ const c = ready.body.events.consumer;
++ assert.strictEqual(c.enabled, true);
++ assert.ok(c.received >= 10 && c.applied >= 5 && c.duplicates >= 3 && c.refused >= 6, JSON.stringify(c));
++ assert.strictEqual(c.last_type, 'live.release.deployed');
++ assert.strictEqual(c.last_outcome, 'announced');
++});
++
++t('inbox receipts older than Events’ retention are pruned', async () => {
++ h.db.run("INSERT INTO chat_event_inbox (consumer, event_id, processed_at) VALUES ('chat', 'evt_OLD', ?)", [Date.now() - 40 * 86400e3]);
++ const kept = h.db.get('SELECT COUNT(*) AS n FROM chat_event_inbox').n - 1;
++ assert.strictEqual(h.eventsConsumer.prune(), 1);
++ assert.strictEqual(h.db.get('SELECT COUNT(*) AS n FROM chat_event_inbox').n, kept);
++});
++
++t.run(async () => { if (h) await h.close(); eventsServer.close(); });
ddiff --git a/test/helpers.js b/test/helpers.js
iindex 4dddd86..b5afcaa 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -320,6 +320,8 @@ async function boot({ env = {} } = {}) {
h.server = started.server;
h.mirrorRelay = started.mirror;
h.eventsRelay = started.relay;
++ h.eventsConsumer = started.events;
++ h.subscriptions = started.subscriptions;
h.port = started.server.address().port;
h.base = `http://127.0.0.1:${h.port}`;
h.db = require('../server/db/database');
@@@ -417,7 +419,7 @@ async function boot({ env = {} } = {}) {
h.close = async () => {
for (const c of h.children || []) { if (c.exitCode == null && c.signalCode == null) { c.kill('SIGKILL'); await c.exited; } }
try { h.chatServer.close(); } catch { /* */ }
-- try { h.ctx.stop(); h.mirrorRelay.stop(); h.eventsRelay.stop(); } catch { /* */ }
++ try { h.ctx.stop(); h.mirrorRelay.stop(); h.eventsRelay.stop(); h.eventsConsumer.stop(); h.subscriptions.stop(); } catch { /* */ }
await new Promise((r) => h.server.close(() => r()));
network.close(); liveServer.close();
try { h.db.close(); } catch { /* */ }