openvibers/openvibe.chat Commits

Guard ingress events, moderation and invalidate with livebridge.write (#17)

OP OpenVibers committed 4 Oct 2026 · 72214349b0 · 16 files, +935 −23

checks run by GitHub CI (test) · CI ✓ 3/3 · review SHIP by DeepSeek harness

https://github.com/OpenVibers/OpenVibe.Chat/pull/17

Changes

ddiff --git a/STATUS.json b/STATUS.json
iindex 2339bed..2167a28 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -8,7 +8,7 @@
"port": 4400,
"domain": "openvibe.chat (Chat's own site: global chat, messages, rooms, settings; the service manifest records it live since 2026-09-24) and openvibe.live (/ws/chat, /api/chat/, /api/dm/, /api/tts/, /api/sounds via an nginx include in openvibe.live's vhost)",
"contracts": "openvibe-contracts v0.71.0",
-- "capabilities": ["chat.message.send"],
++ "capabilities": ["chat.message.send", "chat.presence.read", "chat.live_bridge.write"],
"capabilitiesUsed": [
"live.chat_context.read",
"live.chat_effects.write",
ddiff --git a/docs/capabilities-proposal/chat.cache.invalidate.json b/docs/capabilities-proposal/chat.cache.invalidate.json
nnew file mode 100644
iindex 0000000..4f42b42
---- /dev/null
++++ b/docs/capabilities-proposal/chat.cache.invalidate.json
@@@ -0,0 +1,13 @@
++{
++ "id": "chat.cache.invalidate",
++ "version": "1.0",
++ "owner": "chat",
++ "status": "planned",
++ "visibility": "internal",
++ "description": "Invalidate Chat's user, channel, IP approval and ban cache entries after an external write.",
++ "permissions": ["chat:write"],
++ "resourceConstraints": ["none"],
++ "quotaClass": "internal-write",
++ "events": [],
++ "implementedBy": ["POST /internal/chat/invalidate"]
++}
ddiff --git a/docs/capabilities-proposal/chat.event.publish.json b/docs/capabilities-proposal/chat.event.publish.json
nnew file mode 100644
iindex 0000000..aa4e730
---- /dev/null
++++ b/docs/capabilities-proposal/chat.event.publish.json
@@@ -0,0 +1,13 @@
++{
++ "id": "chat.event.publish",
++ "version": "1.0",
++ "owner": "chat",
++ "status": "planned",
++ "visibility": "internal",
++ "description": "Publish validated transient chat cards, alerts, sounds, media-queue, redemption and vibe-coding frames to a typed room audience, and call invite/response frames to one user.",
++ "permissions": ["chat:write"],
++ "resourceConstraints": ["none"],
++ "quotaClass": "internal-write",
++ "events": [],
++ "implementedBy": ["POST /internal/chat/events"]
++}
ddiff --git a/docs/capabilities-proposal/chat.moderation.write.json b/docs/capabilities-proposal/chat.moderation.write.json
nnew file mode 100644
iindex 0000000..f1615a5
---- /dev/null
++++ b/docs/capabilities-proposal/chat.moderation.write.json
@@@ -0,0 +1,13 @@
++{
++ "id": "chat.moderation.write",
++ "version": "1.0",
++ "owner": "chat",
++ "status": "planned",
++ "visibility": "internal",
++ "description": "Write typed Chat-owned moderation actions and broadcast their results.",
++ "permissions": ["chat:write"],
++ "resourceConstraints": ["none"],
++ "quotaClass": "internal-write",
++ "events": ["chat.message.deleted", "chat.moderation.action"],
++ "implementedBy": ["POST /internal/chat/moderation"]
++}
ddiff --git a/docs/capabilities-proposal/chat.presence.read.json b/docs/capabilities-proposal/chat.presence.read.json
iindex 7afab02..c1f632d 100644
---- a/docs/capabilities-proposal/chat.presence.read.json
++++ b/docs/capabilities-proposal/chat.presence.read.json
@@@ -14,6 +14,7 @@
"quotaClass": "internal-read",
"events": [],
"implementedBy": [
-- "GET /internal/live/presence"
++ "GET /internal/live/presence",
++ "GET /internal/chat/presence"
]
}
ddiff --git a/docs/chat-ingress.md b/docs/chat-ingress.md
nnew file mode 100644
iindex 0000000..5524d80
---- /dev/null
++++ b/docs/chat-ingress.md
@@@ -0,0 +1,57 @@
++# Chat-owned internal ingress (T3 J2 prerequisite)
++
++These routes accept an RS256 service token for `openvibe.chat` on loopback only. The existing
++`/internal/live/*` bridge remains available during the Live rollout. All POST bodies require a
++stable `key` (1–160 letters, digits, `:._-`). A retry with the same principal, route and body
++returns the first result. Reusing the key with another body returns 409. If delivery fails after
++the database write, retry the same key: Chat retries the broadcast without writing another row.
++While another Chat process holds the delivery claim for the key (or one crashed holding it), a
++retry gets 503 "delivery in progress; retry with the same key", never a 200 for an undelivered
++broadcast. A crashed claim expires after five minutes; the next retry then delivers it, so the
++caller must keep retrying 503s past that lease. A crash between the broadcast and marking it
++delivered can still repeat that broadcast once (at least once, never lost). Bad input is 400 and
++a stream that does not exist is 404: do not retry a 4xx.
++
++| Route | Capability | Body and result |
++| --- | --- | --- |
++| `POST /internal/chat/messages` | `chat.message.send` | Chat message fields (`stream_id`, `channel_user_id`, `user_id`, `anon_id`, `username`, `message`, `message_type`, `is_global`, `reply_to_id`, `source_platform`, `metadata`), optional `mirror`, `tts:{voice,identity_key,key}`, `auto_delete_at`, and typed chat-frame fields in `frame` (`role`: `admin`/`global_mod`/`streamer`/`user`/`anon`/`external`, `profile_color`: hex, `avatar_url`, `is_ai`/`is_bot`/`filtered`: booleans, `core_username`, `display_name`); the broadcast carries `reply_to_id`, `is_global`, `auto_delete_at` and `metadata` from the body. Returns `{ok,id}`. With no TTS key, Chat uses `m<real message id>`. `dm:{to_user_id}` with `user_id` persists and delivers a direct message (the same `dm` frame as `POST /api/dm`, with the sender looked up by id), returning `{ok,id,conversation_id,to_user_id}`; a blank message is 400. Chat records the first chat in the channel (the welcome check) under `first_chat_key` when given (`user:<id>`, `anon:<anonId>` or `ext:<prefixed username>`), otherwise for a `chat` line only: `ext:<username>` with `source_platform`, else `user:<user_id>`, else `anon:<anon_id>`. |
++| `POST /internal/chat/events` | `chat.live_bridge.write` | `{key,target:{kind,id?},frame:{type,...}}`, returns `{ok:true}`. Target kinds: `stream`, `channel`, `owner-streams`, `user` (positive `id`), `global`, `all` (no `id`). Only the event frame types allowlisted in `internal-ingress.js` are accepted; `dm` and delete frames are refused, and `chat` only as the news card. |
++| `POST /internal/chat/moderation` | `chat.live_bridge.write` | `{key,action,...}`; actions: `delete-message`, `delete-user-messages`, `delete-anon-messages`, `delete-relay-messages`, `delete-by-range`, `review-pending-ip`, `approve-ip-messages`, `deny-ip-messages`, `relay-hide`, `relay-unhide`, `relay-record`, `tts-voice-override`, `disconnect`, `log`. Deletes return `ids` and broadcast `delete-messages` to every affected surface. |
++| `POST /internal/chat/invalidate` | `chat.live_bridge.write` | `{key,user?,user_data?,approvals?,bans?,channel?}`; a cache hint, returns `{ok:true}`. Include a typed `user_data` object when Live edited the account, so Chat updates sockets and stored message names. |
++| `GET /internal/chat/presence` | `chat.presence.read` | The same snapshot as `/internal/live/presence`. |
++
++`/internal/chat/events` is a narrow transient event ingress, not a generic ChatServer call.
++`live.release.deployed` continues through OpenVibe.Events; there is no deploy-notice endpoint.
++An `alert` targets its `streamerId` channel; a `channel-sound` targets its `streamId` stream.
++`voice-channels` targets `all`. `media_queue_update` (`state` object) and `media_now_playing`
++(`request` object or null) target `owner-streams` (the media owner's live streams).
++`redemption` (`username`, `reward_title`, numeric `cost`) and `vibe-coding` (`managed_stream_id`,
++`event` object) target a `stream`. The transient call frames `vc-call-invite` and
++`vc-call-response` (typed fields only: `channelId`, `channelName`, `fromUserId`, `fromUsername`,
++`fromDisplayName`, `fromAvatarUrl`, `createdAt`, plus `status` on a response) are the only frames
++that target a `user`; Chat sends them to that user's sockets and persists nothing. The news card
++(Live `news-service`) is a `chat` frame with `message_type: 'news'` and only `username`,
++`message`, `timestamp`, and optional `url` (≤ 500 characters), `news_source`,
++`source_platform: 'news'` and `system` (boolean); it targets a `stream` and is never
++saved.
++
++`events`, `moderation` and `invalidate` take `chat.live_bridge.write`, the capability Live
++already holds for the same broadcasts, moderation writes and cache hints through
++`/internal/live/calls`, so Live repoints with the grants it has today (`chat.message.send`,
++`chat.presence.read`, `chat.live_bridge.write`). Narrower capabilities (`chat.event.publish`,
++`chat.moderation.write`, `chat.cache.invalidate`; proposals in `docs/capabilities-proposal/`)
++need the separate **OpenVibe.Contracts** repository to register them and add them to
++`manifests/services/chat.json` first: Chat's CI `contracts` step (`openvibe-contracts-check`) refuses a guard on
++a capability Contracts does not define. Then Chat switches the three guards, and Network grants
++`live chat.event.publish openvibe.chat`, `live chat.moderation.write openvibe.chat` and
++`live chat.cache.invalidate openvibe.chat` before that Chat release deploys.
++
++Then Live J2 replaces its bridge writer with a typed client: repoint message producers and persisted DMs
++to `messages` (with `role`/`profile_color` etc. in `frame`), card, sound, media-queue, redemption
++and vibe-coding producers to `events`, the call invite/response `sendDm` calls to `events` with a
++`user` target, moderation writes and disconnects to
++`moderation`, cache hints to `invalidate`, and presence reads to `/internal/chat/presence`.
++Await the real message id; use one stable key per operation. Drain Live's bridge outbox before
++removing its sender. Live J4b moves arena-command replies into the Chat-effects response.
++For deploy notices, Live publishes `live.release.deployed` through Events. If Events outbox
++initialization fails, Live must leave `deploy_last_announced` unchanged so the next boot retries.
ddiff --git a/docs/cutover.md b/docs/cutover.md
iindex 16c442f..6000ea2 100644
---- a/docs/cutover.md
++++ b/docs/cutover.md
@@@ -10,9 +10,19 @@ until a later wave retires them.
The cutover itself ran on 2026-09-23; the rehearsal, flip and check steps below record that deploy.
++## Rehearsal
++
++`ov rehearse` loads this seed after main's migrations, then this branch's migrations:
++
++```rehearse
++# Chat's PostgreSQL seed; test/fixtures/live-chat-schema.sql is Live's SQLite schema for the import tests
++seed: test/rehearsal/seed.sql
++```
++
- [What is served where](#what-is-served-where)
- [Data authority](#data-authority)
- [Prerequisites](#prerequisites)
++- [Rehearsal](#rehearsal)
- [Rehearsal on a live.db snapshot](#rehearsal-on-a-livedb-snapshot)
- [Cutover](#cutover)
- [Rollback](#rollback)
@@@ -49,10 +59,14 @@ Stay on Live (do not route): `/api/chat-ai/*` (the trailing slash in `/api/chat/
`/ws/call` (until the calls cutover, `docs/calls-cutover.md`), `/ws/broadcast`, `/ws/control`, and
everything else.
--Loopback only, never routed: Chat `/internal/live/*`, `/health`, `/ready`; Live
++Loopback only, never routed: Chat `/internal/live/*`, `/internal/chat/*`, `/health`, `/ready`; Live
`/internal/chat-context/*`, `/internal/chat-effects/*` (both refuse anything that came through
nginx).
++The additive Chat-owned ingress at `/internal/chat/{messages,events,moderation,invalidate,presence}`
++is documented in [chat-ingress.md](chat-ingress.md). Live keeps using `/internal/live/*` until its
++writers are repointed; deploy the three new capability grants before that Live release.
++
## Data authority
| Table (Live baseline, target OpenVibe.Chat) | W6 authority | Notes |
ddiff --git a/migrations/0003_chat_ingress.sql b/migrations/0003_chat_ingress.sql
nnew file mode 100644
iindex 0000000..2049c01
---- /dev/null
++++ b/migrations/0003_chat_ingress.sql
@@@ -0,0 +1,14 @@
++-- phase: expand
++-- Durable results for the Chat-owned ingress. Keep separate from bridge_applied so J3 can
++-- retire the Live bridge without losing retry protection for the replacement routes.
++CREATE TABLE IF NOT EXISTS chat_ingress_applied (
++ principal text COLLATE "C" NOT NULL,
++ family text COLLATE "C" NOT NULL,
++ key text COLLATE "C" NOT NULL,
++ body_hash text COLLATE "C" NOT NULL,
++ result text COLLATE "C" NOT NULL,
++ delivered bigint NOT NULL DEFAULT 0,
++ delivery_claimed_at bigint,
++ applied_at bigint NOT NULL,
++ PRIMARY KEY (principal, family, key)
++);
ddiff --git a/server/app.js b/server/app.js
iindex 9ef5577..10f2101 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -19,6 +19,7 @@ const ctx = require('./live-context');
const db = require('./db/database');
const { extractWsToken, authenticateWs, requestUser } = require('./auth/auth');
const { trustProxy } = require('./net/client-ip');
++const { createInternalIngress } = require('./chat/internal-ingress');
function normalizeOrigin(origin) {
if (!origin || typeof origin !== 'string') return null;
@@@ -102,6 +103,7 @@ function createApp({ chatServer, bridge, mirror, relay, events = null, callServe
// ── 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);
++ app.use('/internal/chat', express.json({ limit: '64kb' }), createInternalIngress({ chatServer }));
// 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);
ddiff --git a/server/bridge/live-bridge.js b/server/bridge/live-bridge.js
iindex 1be077a..7f73d9d 100644
---- a/server/bridge/live-bridge.js
++++ b/server/bridge/live-bridge.js
@@@ -257,23 +257,7 @@ function createBridge({ chatServer, mirror = null, config = require('../config')
});
router.get('/presence', serviceAuth.guard('chat.presence.read'), async (req, res) => {
-- const streams = {};
-- const users = [];
-- const anons = [];
-- for (const [, c] of chatServer.clients) {
-- if (c.streamId) streams[c.streamId] = null;
-- if (c.user) users.push({ user_id: c.user.id, ip: c.ip, stream_id: c.streamId || null });
-- else if (c.anonId) anons.push({ anon_id: c.anonId, ip: c.ip, stream_id: c.streamId || null });
-- }
-- for (const sid of Object.keys(streams)) streams[sid] = chatServer.getStreamViewerCount(Number(sid));
-- res.json({
-- at: new Date().toISOString(),
-- total: chatServer.getTotalConnections(),
-- streams,
-- slow_mode: Object.fromEntries(await chatServer.slowModeByStream()),
-- users,
-- anons,
-- });
++ res.json(await require('../chat/presence').snapshot(chatServer));
});
router._mapRefs = mapRefs;
ddiff --git a/server/chat/internal-ingress.js b/server/chat/internal-ingress.js
nnew file mode 100644
iindex 0000000..6b53563
---- /dev/null
++++ b/server/chat/internal-ingress.js
@@@ -0,0 +1,379 @@
++/** Typed, service-token ingress for writes that Live still originates. */
++'use strict';
++
++const crypto = require('crypto');
++const express = require('express');
++const { isIP } = require('net');
++const db = require('../db/database');
++const ctx = require('../live-context');
++const dm = require('./dm');
++const serviceAuth = require('../net/service-auth');
++
++const locks = new Map();
++const int = (v) => Number.isSafeInteger(v) && v > 0;
++const str = (v, max = 2000) => typeof v === 'string' && v.length > 0 && v.length <= max;
++const obj = (v) => !!v && typeof v === 'object' && !Array.isArray(v);
++const optionalId = (v) => v == null || int(v);
++const bool = (v) => v == null || typeof v === 'boolean';
++const bad = (message) => { const e = new Error(message); e.status = 400; throw e; };
++const missing = (message) => { const e = new Error(message); e.status = 404; throw e; };
++function fields(v, allowed) {
++ if (!obj(v) || Object.keys(v).some((k) => !allowed.includes(k))) bad('Invalid fields');
++}
++function stable(v) {
++ if (Array.isArray(v)) return v.map(stable);
++ if (obj(v)) return Object.fromEntries(Object.keys(v).sort().map((k) => [k, stable(v[k])]));
++ return v;
++}
++function hash(v) { return crypto.createHash('sha256').update(JSON.stringify(stable(v))).digest('hex'); }
++
++// Reserve the key before the domain write, in the same transaction. A concurrent process waits
++// on the unique index and returns the committed result; a failed write rolls the key back.
++async function once(req, family, body, write, deliver) {
++ const principal = req.principal.sub;
++ const key = body.key;
++ const bodyHash = hash(body);
++ const lockKey = `${principal}|${family}|${key}`;
++ while (locks.has(lockKey)) await locks.get(lockKey);
++ let unlock;
++ locks.set(lockKey, new Promise((resolve) => { unlock = resolve; }));
++ try {
++ const result = await db.tx(async () => {
++ const inserted = await db.run(
++ `INSERT INTO chat_ingress_applied (principal, family, key, body_hash, result, applied_at)
++ VALUES (?, ?, ?, ?, '{}', ?) ON CONFLICT DO NOTHING`,
++ [principal, family, key, bodyHash, Date.now()]
++ );
++ if (!inserted.changes) {
++ const old = await db.get('SELECT body_hash, result FROM chat_ingress_applied WHERE principal = ? AND family = ? AND key = ?', [principal, family, key]);
++ if (!old || old.body_hash !== bodyHash) { const e = new Error('Idempotency key reused with a different body'); e.status = 409; throw e; }
++ return JSON.parse(old.result);
++ }
++ const value = await write();
++ await db.run('UPDATE chat_ingress_applied SET result = ? WHERE principal = ? AND family = ? AND key = ?', [JSON.stringify(value), principal, family, key]);
++ return value;
++ });
++ // The claim prevents two Chat processes from broadcasting the same retry at once. A
++ // crashed process leaves a lease that another retry can reclaim after five minutes.
++ const claimed = await db.run(
++ `UPDATE chat_ingress_applied SET delivered = 2, delivery_claimed_at = ?
++ WHERE principal = ? AND family = ? AND key = ?
++ AND (delivered = 0 OR (delivered = 2 AND delivery_claimed_at < ?))`,
++ [Date.now(), principal, family, key, Date.now() - 5 * 60_000]
++ );
++ if (claimed.changes) {
++ try {
++ await deliver(result);
++ await db.run('UPDATE chat_ingress_applied SET delivered = 1 WHERE principal = ? AND family = ? AND key = ?', [principal, family, key]);
++ } catch (err) {
++ await db.run('UPDATE chat_ingress_applied SET delivered = 0 WHERE principal = ? AND family = ? AND key = ? AND delivered = 2', [principal, family, key]).catch(() => {});
++ throw err;
++ }
++ } else {
++ // Another process holds the claim, or one crashed holding it: the caller must retry
++ // rather than take a 200 for a broadcast that may never happen.
++ const row = await db.get('SELECT delivered FROM chat_ingress_applied WHERE principal = ? AND family = ? AND key = ?', [principal, family, key]);
++ if (Number(row?.delivered) !== 1) { const e = new Error('Delivery in progress; retry with the same key'); e.status = 503; throw e; }
++ }
++ return result;
++ } finally { locks.delete(lockKey); unlock(); }
++}
++
++function endpoint(family, validate, write, deliver) {
++ return async (req, res) => {
++ try {
++ const body = req.body;
++ if (!obj(body) || !str(body.key, 160) || !/^[A-Za-z0-9:._-]+$/.test(body.key)) bad('A valid idempotency key is required');
++ validate(body);
++ const result = await once(req, family, body, () => write(body, req), (saved) => deliver(body, saved, req));
++ const { rows, ...publicResult } = result;
++ res.json({ ok: true, ...publicResult });
++ } catch (err) {
++ if (err.status) return res.status(err.status).json({ ok: false, error: err.message });
++ console.error(`[ChatIngress] ${family}:`, err);
++ res.status(503).json({ ok: false, error: 'Chat ingress unavailable; retry with the same key' });
++ }
++ };
++}
++
++const ROLES = ['admin', 'global_mod', 'streamer', 'user', 'anon', 'external'];
++function validateMessage(b) {
++ fields(b, ['key', 'stream_id', 'channel_user_id', 'user_id', 'anon_id', 'username', 'message', 'message_type', 'is_global', 'reply_to_id', 'source_platform', 'auto_delete_at', 'metadata', 'mirror', 'dm', 'tts', 'frame', 'first_chat_key']);
++ if (!str(b.message) || !b.message.trim() || !str(b.username, 120)) bad('Invalid message or username');
++ if (b.auto_delete_at != null && (!str(b.auto_delete_at, 40) || !Number.isFinite(Date.parse(b.auto_delete_at)))) bad('Invalid expiry');
++ if (b.dm) {
++ fields(b.dm, ['to_user_id']);
++ if (!int(b.user_id) || !int(b.dm.to_user_id) || b.stream_id != null || b.channel_user_id != null || b.mirror || b.tts || b.frame) bad('Invalid DM');
++ } else {
++ if (!optionalId(b.stream_id) || !optionalId(b.channel_user_id) || !optionalId(b.user_id) || !optionalId(b.reply_to_id)) bad('Invalid message ID');
++ if (b.anon_id != null && !str(b.anon_id, 80)) bad('Invalid anonymous ID');
++ if (b.message_type != null && !['chat', 'system', 'donation', 'command', 'tts', 'channel-sound', 'soundboard', 'clip'].includes(b.message_type)) bad('Invalid message type');
++ if (b.mirror != null && typeof b.mirror !== 'boolean') bad('Invalid mirror flag');
++ if (b.is_global != null && typeof b.is_global !== 'boolean') bad('Invalid global flag');
++ if (b.source_platform != null && !str(b.source_platform, 80)) bad('Invalid source platform');
++ if (b.first_chat_key != null && (!str(b.first_chat_key, 160) || !/^(?:user|anon|ext):./.test(b.first_chat_key))) bad('Invalid first-chat key');
++ if (b.metadata != null && (!obj(b.metadata) || JSON.stringify(b.metadata).length > 8192)) bad('Invalid metadata');
++ if (b.frame != null) {
++ const f = b.frame;
++ fields(f, ['role', 'profile_color', 'avatar_url', 'is_ai', 'is_bot', 'filtered', 'core_username', 'display_name']);
++ if ((f.role != null && !ROLES.includes(f.role)) || (f.profile_color != null && !/^#[0-9a-fA-F]{3,8}$/.test(f.profile_color))
++ || (f.avatar_url != null && !str(f.avatar_url, 500)) || !bool(f.is_ai) || !bool(f.is_bot) || !bool(f.filtered)
++ || (f.core_username != null && !str(f.core_username, 120)) || (f.display_name != null && !str(f.display_name, 120))) bad('Invalid frame');
++ }
++ if (b.tts != null) {
++ fields(b.tts, ['voice', 'identity_key', 'key']);
++ if (b.tts.voice != null && !str(b.tts.voice, 80)) bad('Invalid TTS voice');
++ if (b.tts.identity_key != null && !str(b.tts.identity_key, 120)) bad('Invalid TTS identity');
++ if (b.tts.key != null && !str(b.tts.key, 160)) bad('Invalid TTS key');
++ if (!b.stream_id && !b.channel_user_id) bad('TTS needs a room');
++ }
++ }
++}
++
++// The keys chat-server's welcome check reads (`user:`/`anon:`) and Live's relay saves (`ext:` +
++// the prefixed username it stores relay lines under). Only chat lines use up a welcome.
++function firstChatKey(b) {
++ if (b.first_chat_key) return b.first_chat_key;
++ if ((b.message_type || 'chat') !== 'chat') return null;
++ if (b.source_platform) return `ext:${b.username}`;
++ if (b.user_id) return `user:${b.user_id}`;
++ return b.anon_id ? `anon:${b.anon_id}` : null;
++}
++
++async function writeMessage(b) {
++ if (b.dm) {
++ if (await dm.isBlockedEither(b.user_id, b.dm.to_user_id)) { const e = new Error('DM participants are blocked'); e.status = 403; throw e; }
++ if (!await ctx.getUserById(b.user_id) || !await ctx.getUserById(b.dm.to_user_id)) { const e = new Error('DM participant not found'); e.status = 404; throw e; }
++ const conversationId = await dm.getOrCreateDirect(b.user_id, b.dm.to_user_id);
++ const message = await dm.sendMessage(conversationId, b.user_id, b.message);
++ if (!message) bad('Invalid message');
++ return { id: Number(message.id), conversation_id: Number(conversationId), to_user_id: b.dm.to_user_id };
++ }
++ if (b.stream_id && !await ctx.ensureStream(b.stream_id)) missing('Stream not found');
++ const saved = await db.saveChatMessage({ ...b, is_global: b.is_global ?? (!b.stream_id && !b.channel_user_id) });
++ const id = Number(saved.lastInsertRowid);
++ const channelUserId = b.channel_user_id || (b.stream_id ? (await ctx.getStreamById(b.stream_id))?.user_id : null) || null;
++ const chatterKey = firstChatKey(b);
++ if (channelUserId && chatterKey) await db.recordFirstChat(chatterKey, channelUserId);
++ return { id, stream_id: b.stream_id || null, channel_user_id: channelUserId };
++}
++
++async function deliverMessage(chatServer, b, result) {
++ if (b.dm) {
++ // The same message shape as dm-routes: the saved row plus the sender's account fields.
++ // A retry after the row was deleted has nothing left to deliver.
++ const row = await db.get('SELECT * FROM dm_messages WHERE id = ?', [result.id]);
++ if (!row) return;
++ const { sender_subject_id, ...message } = row;
++ const sender = await ctx.getUserById(b.user_id);
++ Object.assign(message, { username: sender?.username, display_name: sender?.display_name, avatar_url: sender?.avatar_url, profile_color: sender?.profile_color });
++ const frame = { type: 'dm', conversation_id: result.conversation_id, message };
++ await chatServer.sendDm(b.dm.to_user_id, frame);
++ await chatServer.sendDm(b.user_id, frame);
++ return;
++ }
++ const frame = { type: 'chat', id: result.id, stream_id: result.stream_id, channel_user_id: result.channel_user_id,
++ user_id: b.user_id || null, anon_id: b.anon_id || null, username: b.username, message: b.message,
++ message_type: b.message_type || 'chat', is_global: b.is_global ?? (!result.stream_id && !result.channel_user_id),
++ reply_to_id: b.reply_to_id || null, auto_delete_at: b.auto_delete_at || null, metadata: b.metadata || null,
++ source_platform: b.source_platform || null, timestamp: new Date().toISOString(), ...(b.frame || {}) };
++ if (result.channel_user_id || result.stream_id) await chatServer.broadcastToChannelRoom(result.channel_user_id, result.stream_id, frame);
++ else await chatServer.broadcastGlobal(frame);
++ if (b.mirror && result.stream_id) {
++ await chatServer.forwardToGlobal(result.stream_id, frame);
++ await chatServer.forwardToStreamerRooms(result.stream_id, frame);
++ } else if (b.mirror && result.channel_user_id) await chatServer.forwardToGlobalByChannel(result.channel_user_id, frame);
++ if (b.tts) await chatServer.synthesizeAndBroadcastTTS(result.stream_id, b.username, b.message, b.tts.voice || null, b.source_platform || null, b.tts.identity_key || null, result.channel_user_id, b.tts.key || `m${result.id}`);
++}
++
++const TARGETS = new Set(['stream', 'channel', 'owner-streams', 'global', 'all', 'user']);
++const FRAMES = new Set(['chat', 'system', 'donation', 'goal-update', 'goal-reached', 'update', 'voice-channels', 'alert', 'channel-sound', 'server_restart',
++ 'media_queue_update', 'media_now_playing', 'redemption', 'vibe-coding', 'vc-call-invite', 'vc-call-response']);
++// Transient call frames for one user's sockets; nothing is persisted.
++const CALL_FRAME = ['type', 'channelId', 'channelName', 'fromUserId', 'fromUsername', 'fromDisplayName', 'fromAvatarUrl', 'createdAt'];
++function validateCallFrame(f) {
++ fields(f, f.type === 'vc-call-response' ? [...CALL_FRAME, 'status'] : CALL_FRAME);
++ if (!str(f.channelId, 120) || (f.channelName != null && !str(f.channelName, 120)) || !int(f.fromUserId) || !str(f.fromUsername, 120)
++ || (f.fromDisplayName != null && !str(f.fromDisplayName, 120)) || (f.fromAvatarUrl != null && !str(f.fromAvatarUrl, 500))
++ || !int(f.createdAt)) bad('Invalid call frame');
++ if (f.type === 'vc-call-response' && !['accepted', 'declined', 'busy', 'no-answer', 'canceled'].includes(f.status)) bad('Invalid call response');
++}
++function validateEvent(b) {
++ fields(b, ['key', 'target', 'frame']);
++ fields(b.target, ['kind', 'id']);
++ if (!TARGETS.has(b.target.kind) || (['stream', 'channel', 'owner-streams', 'user'].includes(b.target.kind) ? !int(b.target.id) : b.target.id != null)) bad('Invalid target');
++ if (!obj(b.frame) || !FRAMES.has(b.frame.type) || JSON.stringify(b.frame).length > 16384 || Object.keys(b.frame).some((k) => /^(?:__proto__|constructor|prototype)$/.test(k))) bad('Invalid frame');
++ if (['system', 'server_restart'].includes(b.frame.type) && !str(b.frame.message, 4000)) bad('Invalid frame message');
++ // The transient news card (Live news-service): a chat frame that is never saved, so only news.
++ if (b.frame.type === 'chat') {
++ fields(b.frame, ['type', 'message_type', 'username', 'message', 'url', 'news_source', 'timestamp', 'source_platform', 'system']);
++ if (b.frame.message_type !== 'news' || !str(b.frame.username, 120) || !str(b.frame.message, 4000) || (b.frame.url != null && !str(b.frame.url, 500))
++ || (b.frame.news_source != null && !str(b.frame.news_source, 120)) || (b.frame.source_platform != null && b.frame.source_platform !== 'news')
++ || !bool(b.frame.system) || !str(b.frame.timestamp, 40) || !Number.isFinite(Date.parse(b.frame.timestamp))) bad('Invalid news card');
++ if (b.target.kind !== 'stream') bad('News target must be a stream');
++ }
++ if (b.frame.type === 'alert' && (!int(b.frame.streamerId) || !optionalId(b.frame.streamId) || !['goal', 'donation'].includes(b.frame.kind))) bad('Invalid alert');
++ if (b.frame.type === 'voice-channels' && (!Array.isArray(b.frame.channels) || b.frame.channels.length > 128)) bad('Invalid voice channels');
++ if (b.frame.type === 'channel-sound' && (!int(b.frame.streamId) || !str(b.frame.command, 120) || (b.frame.userId != null && !int(b.frame.userId)) || (b.frame.anonId != null && !str(b.frame.anonId, 80)) || (b.frame.args != null && (!Array.isArray(b.frame.args) || b.frame.args.length > 20 || b.frame.args.some((a) => !str(a, 120)))) || (b.frame.relay != null && (!obj(b.frame.relay) || !str(b.frame.relay.username, 120))))) bad('Invalid channel sound');
++ if (b.frame.type === 'alert' && (b.target.kind !== 'channel' || b.target.id !== b.frame.streamerId)) bad('Alert target must be its channel owner');
++ if (b.frame.type === 'channel-sound' && (b.target.kind !== 'stream' || b.target.id !== b.frame.streamId)) bad('Sound target must be its stream');
++ if (b.frame.type === 'voice-channels' && b.target.kind !== 'all') bad('Voice channels target must be all');
++ if (b.frame.type === 'media_queue_update' && !obj(b.frame.state)) bad('Invalid media queue');
++ if (b.frame.type === 'media_now_playing' && b.frame.request !== null && !obj(b.frame.request)) bad('Invalid media request');
++ if (b.frame.type.startsWith('media_') && b.target.kind !== 'owner-streams') bad('Media target must be the owner streams');
++ if (b.frame.type === 'redemption' && (!str(b.frame.username, 120) || !str(b.frame.reward_title, 200) || !Number.isFinite(b.frame.cost) || (b.frame.user_input != null && typeof b.frame.user_input !== 'string'))) bad('Invalid redemption');
++ if (b.frame.type === 'vibe-coding' && (!int(b.frame.managed_stream_id) || !obj(b.frame.event))) bad('Invalid vibe-coding event');
++ if (['redemption', 'vibe-coding'].includes(b.frame.type) && b.target.kind !== 'stream') bad('Stream event target must be a stream');
++ const call = b.frame.type.startsWith('vc-call-');
++ if (call) validateCallFrame(b.frame);
++ if (call !== (b.target.kind === 'user')) bad('Only call frames target a user');
++}
++async function deliverEvent(chatServer, b) {
++ const { kind, id } = b.target;
++ const f = b.frame;
++ if (f.type === 'alert') return require('./alert-sounds').playAlertSound(chatServer, f.streamerId, f.streamId || null, f.kind);
++ if (f.type === 'channel-sound') {
++ const stream = await ctx.ensureStream(f.streamId);
++ if (!stream) missing('Stream not found');
++ const client = { streamId: f.streamId, user: f.userId ? await ctx.getUserById(f.userId) : null, anonId: f.anonId || null };
++ return chatServer.triggerChannelSound(null, client, stream, f.command, Array.isArray(f.args) ? f.args : [], f.relay || null);
++ }
++ if (kind === 'stream') { if (!await ctx.ensureStream(id)) missing('Stream not found'); return chatServer.broadcastToStream(id, f); }
++ if (kind === 'channel') return chatServer.broadcastToChannelRoom(id, null, f);
++ if (kind === 'owner-streams') {
++ for (const stream of await ctx.getLiveStreamsByUserId(id) || []) await chatServer.broadcastToStream(stream.id, f);
++ return;
++ }
++ if (kind === 'user') return chatServer.sendDm(id, f);
++ if (kind === 'global') return chatServer.broadcastGlobal(f);
++ return chatServer.broadcastAll(f);
++}
++
++async function broadcastDeletes(chatServer, rows) {
++ const rooms = new Map();
++ for (const row of rows) {
++ const streamId = row.stream_id || null;
++ const channelUserId = row.channel_user_id || (streamId ? (await ctx.getStreamById(streamId))?.user_id : null) || null;
++ const key = `${streamId || ''}|${channelUserId || ''}`;
++ if (!rooms.has(key)) rooms.set(key, { streamId, channelUserId, ids: [] });
++ rooms.get(key).ids.push(Number(row.id));
++ }
++ for (const { streamId, channelUserId, ids } of rooms.values()) {
++ const frame = { type: 'delete-messages', ids };
++ if (streamId || channelUserId) await chatServer.broadcastToChannelRoom(channelUserId, streamId, frame);
++ if (streamId) await chatServer.forwardToGlobal(streamId, frame);
++ else if (channelUserId) await chatServer.forwardToGlobalByChannel(channelUserId, frame);
++ else await chatServer.broadcastGlobal(frame);
++ }
++}
++
++const MOD_FIELDS = {
++ 'delete-message': ['id', 'deleted_by'],
++ 'delete-user-messages': ['user_id', 'stream_id', 'deleted_by'],
++ 'delete-anon-messages': ['anon_id', 'stream_id', 'deleted_by'],
++ 'delete-relay-messages': ['username', 'stream_id', 'deleted_by'],
++ 'delete-by-range': ['stream_id', 'from', 'to', 'deleted_by'],
++ 'review-pending-ip': ['id', 'status', 'reviewed_by', 'channel_id'],
++ 'approve-ip-messages': ['channel_id', 'ip', 'reviewed_by'],
++ 'deny-ip-messages': ['channel_id', 'ip', 'reviewed_by'],
++ 'relay-hide': ['channel_id', 'platform', 'external_username', 'reason', 'created_by', 'mode'],
++ 'relay-unhide': ['id', 'channel_id', 'platform', 'external_username'],
++ 'relay-record': ['platform', 'username'],
++ 'tts-voice-override': ['identity_key', 'params', 'set_by'],
++ disconnect: ['user_id', 'ip', 'stream_id'],
++ log: ['scope_type', 'scope_id', 'actor_user_id', 'target_user_id', 'action_type', 'details'],
++};
++function validateModeration(b) {
++ const allowed = MOD_FIELDS[b.action];
++ if (!allowed) bad('Unknown moderation action');
++ fields(b, ['key', 'action', ...allowed]);
++ if (['delete-message', 'review-pending-ip', 'relay-unhide'].includes(b.action) && b.id != null && !int(b.id)) bad('Invalid ID');
++ for (const name of ['user_id', 'stream_id', 'channel_id', 'reviewed_by', 'actor_user_id', 'target_user_id', 'created_by', 'set_by']) if (b[name] != null && !int(b[name])) bad(`Invalid ${name}`);
++ // chat_messages.deleted_by is text: Live sends the moderator's id or username.
++ if (b.deleted_by != null && !int(b.deleted_by) && !str(b.deleted_by, 120)) bad('Invalid deleted_by');
++ if (b.action === 'relay-hide' && b.reason != null && !str(b.reason, 500)) bad('Invalid reason');
++ if (b.action.startsWith('delete-') && b.action !== 'delete-by-range' && !b.id && !b.user_id && !str(b.anon_id, 80) && !str(b.username, 120)) bad('Missing delete subject');
++ if (b.action === 'delete-by-range' && (!str(b.from, 40) || !str(b.to, 40) || !Number.isFinite(Date.parse(b.from)) || !Number.isFinite(Date.parse(b.to)))) bad('Invalid time range');
++ if (b.action === 'review-pending-ip' && (!int(b.id) || !['approved', 'denied'].includes(b.status))) bad('Invalid review');
++ if (['approve-ip-messages', 'deny-ip-messages'].includes(b.action) && (!int(b.channel_id) || !isIP(b.ip))) bad('Invalid IP review');
++ if (b.action.startsWith('relay-') && b.action !== 'relay-unhide' && (!str(b.platform, 80) || !str(b.external_username || b.username, 120))) bad('Invalid relay identity');
++ if (b.action === 'relay-hide' && b.mode != null && !['hide', 'ban'].includes(b.mode)) bad('Invalid relay action');
++ if (b.action === 'relay-unhide' && !int(b.id) && (!str(b.platform, 80) || !str(b.external_username, 120))) bad('Invalid relay identity');
++ if (b.action === 'tts-voice-override' && (!str(b.identity_key, 120) || (b.params != null && (!obj(b.params) || Object.keys(b.params).some((k) => !['voice', 'pitch', 'speed', 'gap'].includes(k)) || !str(b.params.voice, 80) || ['pitch', 'speed', 'gap'].some((k) => b.params[k] != null && (!Number.isFinite(b.params[k]) || b.params[k] < 0 || b.params[k] > 10)))))) bad('Invalid TTS override');
++ if (b.action === 'disconnect' && ((!int(b.user_id) && !isIP(b.ip)) || (b.ip != null && !isIP(b.ip)))) bad('Invalid disconnect target');
++ if (b.action === 'log' && (!str(b.action_type, 120) || (b.scope_type != null && !['site', 'channel', 'stream', 'room'].includes(b.scope_type)) || (b.scope_id != null && !(b.scope_type === 'room' ? str(b.scope_id, 120) : int(b.scope_id))) || (b.details != null && (!obj(b.details) || JSON.stringify(b.details).length > 8192)))) bad('Invalid moderation log');
++}
++
++async function writeModeration(b) {
++ let rows = [];
++ const deletedBy = b.deleted_by == null ? null : String(b.deleted_by);
++ if (b.action.startsWith('delete-')) {
++ if (b.action === 'delete-message') rows = await db.all('SELECT id, stream_id, channel_user_id FROM chat_messages WHERE id = ? AND is_deleted = 0', [b.id]);
++ else if (b.action === 'delete-user-messages') rows = await db.all(`SELECT id, stream_id, channel_user_id FROM chat_messages WHERE user_id = ? AND is_deleted = 0${b.stream_id ? ' AND stream_id = ?' : ''}`, b.stream_id ? [b.user_id, b.stream_id] : [b.user_id]);
++ else if (b.action === 'delete-anon-messages') rows = await db.all(`SELECT id, stream_id, channel_user_id FROM chat_messages WHERE anon_id = ? AND is_deleted = 0${b.stream_id ? ' AND stream_id = ?' : ''}`, b.stream_id ? [b.anon_id, b.stream_id] : [b.anon_id]);
++ else if (b.action === 'delete-relay-messages') rows = await db.all(`SELECT id, stream_id, channel_user_id FROM chat_messages WHERE username = ? AND is_deleted = 0${b.stream_id ? ' AND stream_id = ?' : ''}`, b.stream_id ? [b.username, b.stream_id] : [b.username]);
++ else if (b.action === 'delete-by-range') {
++ const where = b.stream_id ? 'stream_id = ?' : 'is_global = 1';
++ rows = await db.all(`SELECT id, stream_id, channel_user_id FROM chat_messages WHERE ${where} AND timestamp >= datetime(?) AND timestamp <= datetime(?) AND is_deleted = 0`, b.stream_id ? [b.stream_id, b.from, b.to] : [b.from, b.to]);
++ }
++ if (b.action === 'delete-message' && rows.length) await db.deleteChatMessage(b.id, deletedBy);
++ if (b.action === 'delete-user-messages') await db.deleteUserChatMessages(b.user_id, { streamId: b.stream_id, deletedBy });
++ if (b.action === 'delete-anon-messages') await db.deleteAnonChatMessages(b.anon_id, { streamId: b.stream_id, deletedBy });
++ if (b.action === 'delete-relay-messages') await db.deleteRelayUserMessages(b.username, { streamId: b.stream_id, deletedBy });
++ if (b.action === 'delete-by-range') await db.deleteChatMessagesByTimeRange(b.stream_id || null, b.from, b.to, deletedBy);
++ return { ids: rows.map((r) => Number(r.id)), rows };
++ }
++ let result;
++ if (b.action === 'review-pending-ip') result = await db.reviewPendingIpMessage(b.id, { status: b.status, reviewedBy: b.reviewed_by, channelId: b.channel_id });
++ else if (b.action === 'approve-ip-messages') result = await db.approveAllFromIp(b.channel_id, b.ip, b.reviewed_by);
++ else if (b.action === 'deny-ip-messages') result = await db.denyAllFromIp(b.channel_id, b.ip, b.reviewed_by);
++ else if (b.action === 'relay-hide') result = await db.hideRelayUser({ channelId: b.channel_id, platform: b.platform, externalUsername: b.external_username, action: b.mode || 'hide', reason: b.reason, createdBy: b.created_by });
++ else if (b.action === 'relay-unhide') result = b.id ? await db.unhideRelayUser(b.id) : await db.unhideRelayUserByIdentity(b.channel_id, b.platform, b.external_username);
++ else if (b.action === 'relay-record') result = await db.recordRelayUser(b.platform, b.username);
++ else if (b.action === 'tts-voice-override') result = b.params ? await db.setTtsVoiceOverride(b.identity_key, b.params, b.set_by) : await db.deleteTtsVoiceOverride(b.identity_key);
++ else if (b.action === 'log') result = await db.logModerationAction(b);
++ return { changes: result?.changes || 0, id: result?.lastInsertRowid ? Number(result.lastInsertRowid) : null };
++}
++
++async function deliverModeration(chatServer, b, result) {
++ if (result.rows) return broadcastDeletes(chatServer, result.rows);
++ if (b.action === 'disconnect') { await ctx.invalidateBans(); return chatServer.disconnectUser({ userId: b.user_id, ip: b.ip, streamId: b.stream_id }); }
++ if (['approve-ip-messages', 'deny-ip-messages', 'review-pending-ip'].includes(b.action) && b.channel_id) ctx.invalidateApprovals(b.channel_id);
++}
++
++function validateInvalidate(b) {
++ fields(b, ['key', 'user', 'user_data', 'approvals', 'bans', 'channel']);
++ if (b.user == null && b.approvals == null && b.bans !== true && b.channel == null) bad('No cache hint');
++ if (!optionalId(b.user) || !optionalId(b.approvals) || !optionalId(b.channel) || (b.bans != null && typeof b.bans !== 'boolean')) bad('Invalid cache hint');
++ if (b.user_data != null) {
++ fields(b.user_data, ['id', 'username', 'display_name', 'role', 'avatar_url', 'profile_color']);
++ if (!int(b.user) || b.user_data.id !== b.user || !str(b.user_data.username, 120) || (b.user_data.display_name != null && !str(b.user_data.display_name, 120))) bad('Invalid user update');
++ }
++}
++async function deliverInvalidate(chatServer, b) {
++ if (b.user_data) await chatServer.sendUserUpdate(b.user, b.user_data);
++ else if (b.user) { ctx.invalidateUser(b.user); await ctx.ensureUsers([b.user]).catch(() => {}); }
++ if (b.channel) await ctx.invalidateChannel(b.channel);
++ if (b.approvals) ctx.invalidateApprovals(b.approvals);
++ if (b.bans) await ctx.invalidateBans();
++}
++
++// Until OpenVibe.Contracts registers chat.event.publish, chat.moderation.write and
++// chat.cache.invalidate (proposals in docs/capabilities-proposal/), events, moderation and cache
++// hints take chat.live_bridge.write: the capability Live already holds for the same broadcasts,
++// moderation writes and invalidations through /internal/live/calls, so the ingress grants nothing new.
++const LIVE_WRITE = 'chat.live_bridge.write';
++
++function createInternalIngress({ chatServer }) {
++ const r = express.Router();
++ r.post('/messages', serviceAuth.guard('chat.message.send'), endpoint('messages', validateMessage, writeMessage, (b, result) => deliverMessage(chatServer, b, result)));
++ r.post('/events', serviceAuth.guard(LIVE_WRITE), endpoint('events', validateEvent, async () => ({}), (b) => deliverEvent(chatServer, b)));
++ r.post('/moderation', serviceAuth.guard(LIVE_WRITE), endpoint('moderation', validateModeration, writeModeration, (b, result) => deliverModeration(chatServer, b, result)));
++ r.post('/invalidate', serviceAuth.guard(LIVE_WRITE), endpoint('invalidate', validateInvalidate, async () => ({}), (b) => deliverInvalidate(chatServer, b)));
++ r.get('/presence', serviceAuth.guard('chat.presence.read'), async (req, res) => {
++ res.json(await require('./presence').snapshot(chatServer));
++ });
++ return r;
++}
++
++module.exports = { createInternalIngress };
ddiff --git a/server/chat/presence.js b/server/chat/presence.js
nnew file mode 100644
iindex 0000000..27a7eda
---- /dev/null
++++ b/server/chat/presence.js
@@@ -0,0 +1,23 @@
++'use strict';
++
++async function snapshot(chatServer) {
++ const streams = {};
++ const users = [];
++ const anons = [];
++ for (const [, c] of chatServer.clients) {
++ if (c.streamId) streams[c.streamId] = null;
++ if (c.user) users.push({ user_id: c.user.id, ip: c.ip, stream_id: c.streamId || null });
++ else if (c.anonId) anons.push({ anon_id: c.anonId, ip: c.ip, stream_id: c.streamId || null });
++ }
++ for (const sid of Object.keys(streams)) streams[sid] = chatServer.getStreamViewerCount(Number(sid));
++ return {
++ at: new Date().toISOString(),
++ total: chatServer.getTotalConnections(),
++ streams,
++ slow_mode: Object.fromEntries(await chatServer.slowModeByStream()),
++ users,
++ anons,
++ };
++}
++
++module.exports = { snapshot };
ddiff --git a/test/chat-ingress.test.js b/test/chat-ingress.test.js
nnew file mode 100644
iindex 0000000..a5e8c41
---- /dev/null
++++ b/test/chat-ingress.test.js
@@@ -0,0 +1,320 @@
++'use strict';
++
++const assert = require('assert');
++const { boot, suite } = require('./helpers');
++
++const t = suite('chat-ingress');
++let h, streamer, viewer, streamId, channelId, room, globalRoom;
++const CAPS = {
++ messages: 'chat.message.send', events: 'chat.live_bridge.write', moderation: 'chat.live_bridge.write',
++ invalidate: 'chat.live_bridge.write', presence: 'chat.presence.read',
++};
++const post = (family, body, token = h.serviceToken([CAPS[family]]), headers) =>
++ h.http('POST', `/internal/chat/${family}`, { body, token, headers });
++
++t('boot', async () => {
++ h = await boot();
++ streamer = h.addUser('ingress-streamer', { role: 'streamer' });
++ viewer = h.addUser('ingress-viewer');
++ channelId = h.addChannel(streamer.id);
++ streamId = h.addStream(streamer.id, channelId);
++ await h.ctx.sync();
++ room = await h.ws({ token: viewer.token, stream: streamId });
++ room.sendJson({ type: 'join', streamId, token: viewer.token });
++ await room.next((m) => m.type === 'auth');
++ globalRoom = await h.ws();
++ globalRoom.sendJson({ type: 'join' });
++ await globalRoom.next((m) => m.type === 'auth');
++});
++
++t('each route requires its own service capability and loopback', async () => {
++ for (const family of ['messages', 'events', 'moderation', 'invalidate']) {
++ assert.strictEqual((await h.http('POST', `/internal/chat/${family}`, { body: {} })).status, 401);
++ assert.strictEqual((await post(family, {}, h.serviceToken(['chat.presence.read']))).status, 403);
++ assert.strictEqual((await post(family, {}, h.serviceToken([CAPS[family]], { aud: 'openvibe.live' }))).status, 401);
++ assert.strictEqual((await post(family, {}, h.serviceToken([CAPS[family]]), { 'X-Forwarded-For': '1.2.3.4' })).status, 403);
++ }
++ assert.strictEqual((await h.http('GET', '/internal/chat/presence')).status, 401);
++ assert.strictEqual((await h.http('GET', '/internal/chat/presence', { token: h.serviceToken(['chat.message.send']) })).status, 403);
++ // A message sender cannot publish events, moderate or invalidate, nor a bridge writer send lines.
++ for (const family of ['events', 'moderation', 'invalidate']) {
++ assert.strictEqual((await post(family, {}, h.serviceToken(['chat.message.send']))).status, 403);
++ }
++ assert.strictEqual((await post('messages', {}, h.serviceToken(['chat.live_bridge.write']))).status, 403);
++});
++
++let messageId;
++t('messages persist once, broadcast with the real id, and mirror', async () => {
++ const body = { key: 'ingress:line:1', stream_id: streamId, user_id: viewer.id, username: viewer.username, message: 'owned ingress line', mirror: true };
++ const r = await post('messages', body);
++ assert.strictEqual(r.status, 200, r.text);
++ messageId = r.body.id;
++ assert.ok(messageId > 0);
++ assert.strictEqual((await room.next((m) => m.type === 'chat' && m.message === body.message)).id, messageId);
++ assert.strictEqual((await globalRoom.next((m) => m.type === 'chat' && m.message === body.message)).id, messageId);
++ assert.strictEqual((await h.db.getChatMessageById(messageId)).message, body.message);
++ const received = room.all.filter((m) => m.type === 'chat' && m.id === messageId).length;
++ const retry = await post('messages', body);
++ assert.strictEqual(retry.body.id, messageId);
++ assert.strictEqual((await post('messages', { message: body.message, username: body.username, user_id: body.user_id, stream_id: body.stream_id, mirror: true, key: body.key })).body.id, messageId);
++ assert.strictEqual((await h.db.get('SELECT COUNT(*) AS n FROM chat_messages WHERE message = ?', [body.message])).n, 1);
++ await h.sleep(300);
++ assert.strictEqual(room.all.filter((m) => m.type === 'chat' && m.id === messageId).length, received);
++ assert.strictEqual((await post('messages', { ...body, message: 'changed' })).status, 409);
++ assert.strictEqual((await post('messages', { ...body, key: 'ingress:bad:1', frame: { type: 'delete-messages' } })).status, 400);
++ assert.strictEqual((await post('messages', { ...body, key: 'ingress:bad:2', frame: { profile_color: 'red;}' } })).status, 400);
++ assert.strictEqual((await post('messages', { ...body, key: 'ingress:bad:3', frame: { role: 'superuser' } })).status, 400);
++ assert.strictEqual((await post('messages', { ...body, key: 'ingress:bad:4', frame: { is_ai: 'yes' } })).status, 400);
++});
++
++t('relayed and AI lines keep their colour, role and threading', async () => {
++ const body = { key: 'ingress:relay-line:1', stream_id: streamId, username: 'twitch:outside', message: 'relayed reply',
++ source_platform: 'twitch', reply_to_id: messageId, is_global: false, auto_delete_at: new Date(Date.now() + 600e3).toISOString(),
++ frame: { role: 'external', profile_color: '#33aaff', avatar_url: 'https://example.test/a.png', is_ai: false, filtered: false, core_username: 'outside' } };
++ const r = await post('messages', body);
++ assert.strictEqual(r.status, 200, r.text);
++ const frame = await room.next((m) => m.type === 'chat' && m.id === r.body.id);
++ assert.strictEqual(frame.role, 'external');
++ assert.strictEqual(frame.profile_color, '#33aaff');
++ assert.strictEqual(frame.reply_to_id, messageId);
++ assert.strictEqual(frame.is_global, false);
++ assert.strictEqual(frame.auto_delete_at, body.auto_delete_at);
++ assert.strictEqual(frame.core_username, 'outside');
++});
++
++t('first-chat records use the keys the welcome checks read; only chat lines use one up', async () => {
++ const keys = async () => (await h.db.all('SELECT chatter_key FROM stream_first_chats WHERE channel_user_id = ?', [streamer.id])).map((r) => r.chatter_key);
++ assert.ok((await keys()).includes(`user:${viewer.id}`));
++ assert.ok((await keys()).includes('ext:twitch:outside'));
++ assert.strictEqual(await h.db.isFirstChatInChannel(`user:${viewer.id}`, streamer.id), false);
++ const donor = h.addUser('ingress-donor');
++ await h.ctx.sync();
++ const donation = await post('messages', { key: 'ingress:first:donation', stream_id: streamId, user_id: donor.id, username: donor.username, message: 'donated 5', message_type: 'donation' });
++ assert.strictEqual(donation.status, 200, donation.text);
++ assert.ok(!(await keys()).includes(`user:${donor.id}`));
++ const anon = await post('messages', { key: 'ingress:first:anon', stream_id: streamId, anon_id: 'anon-first-1', username: 'anon-first-1', message: 'hi' });
++ assert.strictEqual(anon.status, 200, anon.text);
++ assert.ok((await keys()).includes('anon:anon-first-1'));
++ const explicit = await post('messages', { key: 'ingress:first:explicit', stream_id: streamId, username: 'Bot', message: 'hello', message_type: 'system', first_chat_key: 'ext:kick:someone' });
++ assert.strictEqual(explicit.status, 200, explicit.text);
++ assert.ok((await keys()).includes('ext:kick:someone'));
++ assert.strictEqual((await post('messages', { key: 'ingress:first:bad', stream_id: streamId, username: 'Bot', message: 'x', first_chat_key: '42' })).status, 400);
++});
++
++t('a retry while another process holds the delivery claim is told to retry', async () => {
++ const body = { key: 'ingress:claim:1', stream_id: streamId, username: 'ClaimBot', message: 'claimed line' };
++ const r = await post('messages', body);
++ assert.strictEqual(r.status, 200, r.text);
++ await h.db.run('UPDATE chat_ingress_applied SET delivered = 2, delivery_claimed_at = ? WHERE key = ?', [Date.now(), body.key]);
++ assert.strictEqual((await post('messages', body)).status, 503);
++ await h.db.run('UPDATE chat_ingress_applied SET delivery_claimed_at = ? WHERE key = ?', [Date.now() - 10 * 60_000, body.key]);
++ const reclaimed = await post('messages', body);
++ assert.strictEqual(reclaimed.status, 200, reclaimed.text);
++ assert.strictEqual(reclaimed.body.id, r.body.id);
++ await h.sleep(300);
++ assert.strictEqual(room.all.filter((m) => m.type === 'chat' && m.id === r.body.id).length, 2);
++ assert.strictEqual(Number((await h.db.get('SELECT delivered FROM chat_ingress_applied WHERE key = ?', [body.key])).delivered), 1);
++});
++
++t('DM mode persists and delivers to the recipient', async () => {
++ const body = { key: 'ingress:dm:1', user_id: streamer.id, dm: { to_user_id: viewer.id }, username: streamer.username, message: 'private ingress line' };
++ const r = await post('messages', body);
++ assert.strictEqual(r.status, 200, r.text);
++ const frame = await room.next((m) => m.type === 'dm' && m.message?.message === body.message);
++ assert.strictEqual(frame.message.id, r.body.id);
++ assert.strictEqual(frame.message.sender_id, streamer.id);
++ assert.strictEqual(frame.message.username, streamer.username);
++ assert.ok(frame.message.created_at);
++ assert.ok('profile_color' in frame.message && !('sender_subject_id' in frame.message));
++ assert.strictEqual((await post('messages', { ...body, key: 'ingress:dm:blank', message: ' ' })).status, 400);
++ assert.strictEqual((await h.db.get('SELECT message FROM dm_messages WHERE id = ?', [r.body.id])).message, body.message);
++ assert.strictEqual((await post('messages', body)).body.id, r.body.id);
++ await h.db.run('INSERT INTO dm_blocks (blocker_id, blocked_id) VALUES (?, ?)', [viewer.id, streamer.id]);
++ assert.strictEqual((await post('messages', { ...body, key: 'ingress:dm:blocked' })).status, 403);
++ assert.strictEqual((await h.db.get('SELECT COUNT(*) AS n FROM dm_messages WHERE message = ?', [body.message])).n, 1);
++ await h.db.run('DELETE FROM dm_blocks WHERE blocker_id = ? AND blocked_id = ?', [viewer.id, streamer.id]);
++});
++
++t('optional TTS uses the persisted message id as its queue key', async () => {
++ const original = h.chatServer.synthesizeAndBroadcastTTS;
++ let args;
++ h.chatServer.synthesizeAndBroadcastTTS = async (...received) => { args = received; };
++ try {
++ const r = await post('messages', { key: 'ingress:tts:1', stream_id: streamId, username: 'TTSBot', message: 'read this line', tts: { voice: 'en-US' } });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(args[0], streamId);
++ assert.strictEqual(args[2], 'read this line');
++ assert.strictEqual(args[7], `m${r.body.id}`);
++ assert.strictEqual((await post('messages', { key: 'ingress:tts:1', stream_id: streamId, username: 'TTSBot', message: 'read this line', tts: { voice: 'en-US' } })).body.id, r.body.id);
++ } finally { h.chatServer.synthesizeAndBroadcastTTS = original; }
++});
++
++t('events validate targets and frame types, then publish once', async () => {
++ const body = { key: 'ingress:event:1', target: { kind: 'stream', id: streamId }, frame: { type: 'system', message: 'ingress notice' } };
++ assert.strictEqual((await post('events', body)).status, 200);
++ await room.next((m) => m.type === 'system' && m.message === body.frame.message);
++ const received = room.all.filter((m) => m.type === 'system' && m.message === body.frame.message).length;
++ assert.strictEqual((await post('events', body)).status, 200);
++ await h.sleep(300);
++ assert.strictEqual(room.all.filter((m) => m.type === 'system' && m.message === body.frame.message).length, received);
++ assert.strictEqual((await post('events', { ...body, key: 'ingress:event:bad', frame: { type: 'chat', message: 'injected' } })).status, 400);
++ assert.strictEqual((await post('events', { ...body, key: 'ingress:event:bad2', target: { kind: 'stream' } })).status, 400);
++});
++
++t('the transient news card reaches a stream and is never saved', async () => {
++ const frame = { type: 'chat', message_type: 'news', username: 'News', message: 'ingress headline', url: 'https://example.test/n',
++ news_source: 'wire', timestamp: new Date().toISOString(), source_platform: 'news', system: true };
++ const r = await post('events', { key: 'ingress:news:1', target: { kind: 'stream', id: streamId }, frame });
++ assert.strictEqual(r.status, 200, r.text);
++ const got = await room.next((m) => m.type === 'chat' && m.message === frame.message);
++ assert.strictEqual(got.message_type, 'news');
++ assert.strictEqual(got.url, frame.url);
++ assert.strictEqual((await h.db.get('SELECT COUNT(*) AS n FROM chat_messages WHERE message = ?', [frame.message])).n, 0);
++ for (const [key, bad] of [
++ ['global', { target: { kind: 'global' }, frame }],
++ ['type', { target: { kind: 'stream', id: streamId }, frame: { ...frame, message_type: 'chat' } }],
++ ['url', { target: { kind: 'stream', id: streamId }, frame: { ...frame, url: 'x'.repeat(501) } }],
++ ['extra', { target: { kind: 'stream', id: streamId }, frame: { ...frame, user_id: viewer.id } }],
++ ['platform', { target: { kind: 'stream', id: streamId }, frame: { ...frame, source_platform: 'twitch' } }],
++ ['time', { target: { kind: 'stream', id: streamId }, frame: { ...frame, timestamp: 'soon' } }],
++ ]) assert.strictEqual((await post('events', { key: `ingress:news:bad:${key}`, ...bad })).status, 400, key);
++});
++
++t('bad input and missing rows are refused with a 4xx, never a retryable 503', async () => {
++ assert.strictEqual((await post('events', { key: 'ingress:4xx:stream', target: { kind: 'stream', id: 987654 }, frame: { type: 'system', message: 'nobody' } })).status, 404);
++ assert.strictEqual((await post('events', { key: 'ingress:4xx:alert', target: { kind: 'channel', id: streamer.id }, frame: { type: 'alert', streamerId: streamer.id, streamId: 'x', kind: 'goal' } })).status, 400);
++ assert.strictEqual((await post('messages', { key: 'ingress:4xx:line', stream_id: 987654, username: 'Bot', message: 'nowhere' })).status, 404);
++ assert.strictEqual((await post('moderation', { key: 'ingress:4xx:by', action: 'delete-message', id: 1, deleted_by: { id: 1 } })).status, 400);
++ assert.strictEqual((await post('moderation', { key: 'ingress:4xx:reason', action: 'relay-hide', channel_id: channelId, platform: 'twitch', external_username: 'x', reason: 5 })).status, 400);
++ const dmBody = { key: 'ingress:4xx:dm', user_id: streamer.id, dm: { to_user_id: viewer.id }, username: streamer.username, message: 'soon gone' };
++ const sent = await post('messages', dmBody);
++ assert.strictEqual(sent.status, 200, sent.text);
++ await h.db.run('UPDATE chat_ingress_applied SET delivered = 0 WHERE key = ?', [dmBody.key]);
++ await h.db.run('DELETE FROM dm_messages WHERE id = ?', [sent.body.id]);
++ const retry = await post('messages', dmBody);
++ assert.strictEqual(retry.status, 200, retry.text);
++ assert.strictEqual(retry.body.id, sent.body.id);
++});
++
++t('media, redemption and vibe-coding frames reach their streams; call frames reach one user', async () => {
++ const media = { key: 'ingress:media:1', target: { kind: 'owner-streams', id: streamer.id }, frame: { type: 'media_queue_update', state: { queue: [] }, timestamp: new Date().toISOString() } };
++ assert.strictEqual((await post('events', media)).status, 200);
++ await room.next((m) => m.type === 'media_queue_update' && Array.isArray(m.state.queue));
++ assert.strictEqual((await post('events', { ...media, key: 'ingress:media:bad', target: { kind: 'stream', id: streamId } })).status, 400);
++ assert.strictEqual((await post('events', { ...media, key: 'ingress:media:bad2', frame: { type: 'media_now_playing', request: 'x' } })).status, 400);
++ const redemption = { key: 'ingress:redeem:1', target: { kind: 'stream', id: streamId }, frame: { type: 'redemption', username: viewer.username, reward_title: 'Hydrate', reward_icon: null, reward_color: '#fff', cost: 100, user_input: '' } };
++ assert.strictEqual((await post('events', redemption)).status, 200);
++ await room.next((m) => m.type === 'redemption' && m.reward_title === 'Hydrate');
++ const vibe = { key: 'ingress:vibe:1', target: { kind: 'stream', id: streamId }, frame: { type: 'vibe-coding', managed_stream_id: 7, slot_slug: 'main', delay_ms: 0, event: { kind: 'edit' } } };
++ assert.strictEqual((await post('events', vibe)).status, 200);
++ await room.next((m) => m.type === 'vibe-coding' && m.event.kind === 'edit');
++
++ const invite = { key: 'ingress:call:1', target: { kind: 'user', id: viewer.id }, frame: { type: 'vc-call-invite', channelId: 'room-abc', channelName: 'Call',
++ fromUserId: streamer.id, fromUsername: streamer.username, fromDisplayName: 'Streamer', fromAvatarUrl: null, createdAt: Date.now() } };
++ assert.strictEqual((await post('events', invite)).status, 200);
++ await room.next((m) => m.type === 'vc-call-invite' && m.channelId === 'room-abc');
++ await h.sleep(200);
++ assert.ok(!globalRoom.all.some((m) => m.type === 'vc-call-invite'));
++ const response = { ...invite, key: 'ingress:call:2', frame: { ...invite.frame, type: 'vc-call-response', status: 'accepted' } };
++ assert.strictEqual((await post('events', response)).status, 200);
++ await room.next((m) => m.type === 'vc-call-response' && m.status === 'accepted');
++ assert.strictEqual((await post('events', { ...response, key: 'ingress:call:bad', frame: { ...response.frame, status: 'maybe' } })).status, 400);
++ assert.strictEqual((await post('events', { ...invite, key: 'ingress:call:bad2', target: { kind: 'stream', id: streamId } })).status, 400);
++ assert.strictEqual((await post('events', { key: 'ingress:call:bad3', target: { kind: 'user', id: viewer.id }, frame: { type: 'system', message: 'hi' } })).status, 400);
++ assert.strictEqual((await post('events', { ...invite, key: 'ingress:call:bad4', frame: { ...invite.frame, conversation_id: 1 } })).status, 400);
++});
++
++t('sound events pass a typed stream client to the Chat sound queue', async () => {
++ const original = h.chatServer.triggerChannelSound;
++ let call;
++ h.chatServer.triggerChannelSound = async (...args) => { call = args; };
++ try {
++ const frame = { type: 'channel-sound', streamId, command: 'honk', args: ['0.5'], relay: { username: 'external' } };
++ const r = await post('events', { key: 'ingress:sound:1', target: { kind: 'stream', id: streamId }, frame });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(call[1].streamId, streamId);
++ assert.strictEqual(call[2].id, streamId);
++ assert.strictEqual(call[3], 'honk');
++ assert.strictEqual(call[5].username, 'external');
++ assert.strictEqual((await post('events', { key: 'ingress:sound:bad', target: { kind: 'all' }, frame })).status, 400);
++ } finally { h.chatServer.triggerChannelSound = original; }
++});
++
++t('moderation deletes once, announces deleted ids, and writes a log', async () => {
++ const body = { key: 'ingress:delete:1', action: 'delete-message', id: messageId, deleted_by: streamer.username };
++ const r = await post('moderation', body);
++ assert.strictEqual(r.status, 200, r.text);
++ assert.deepStrictEqual(r.body.ids, [messageId]);
++ assert.strictEqual((await room.next((m) => m.type === 'delete-messages' && m.ids.includes(messageId))).type, 'delete-messages');
++ assert.strictEqual((await h.db.getChatMessageById(messageId)).is_deleted, 1);
++ const received = room.all.filter((m) => m.type === 'delete-messages' && m.ids.includes(messageId)).length;
++ assert.deepStrictEqual((await post('moderation', body)).body.ids, [messageId]);
++ await h.sleep(300);
++ assert.strictEqual(room.all.filter((m) => m.type === 'delete-messages' && m.ids.includes(messageId)).length, received);
++ const log = await post('moderation', { key: 'ingress:log:1', action: 'log', action_type: 'ingress_test', actor_user_id: streamer.id, details: { case: 'test' } });
++ assert.strictEqual(log.status, 200, log.text);
++ assert.ok((await h.db.get('SELECT id FROM moderation_actions WHERE id = ?', [log.body.id])).id);
++});
++
++t('moderation handles bulk deletes, pending IP decisions, relay identities and voice overrides', async () => {
++ const a = await post('messages', { key: 'ingress:bulk:a', stream_id: streamId, user_id: viewer.id, username: viewer.username, message: 'bulk A' });
++ const b = await post('messages', { key: 'ingress:bulk:b', stream_id: streamId, user_id: viewer.id, username: viewer.username, message: 'bulk B' });
++ assert.strictEqual(a.status, 200, a.text);
++ assert.strictEqual(b.status, 200, b.text);
++ const gone = await post('moderation', { key: 'ingress:bulk:delete', action: 'delete-user-messages', user_id: viewer.id, stream_id: streamId, deleted_by: streamer.username });
++ assert.strictEqual(gone.status, 200, gone.text);
++ assert.ok(gone.body.ids.includes(a.body.id) && gone.body.ids.includes(b.body.id));
++ await room.next((m) => m.type === 'delete-messages' && m.ids.includes(a.body.id));
++
++ const ip = '198.51.100.80';
++ const pending = await h.db.run('INSERT INTO pending_ip_messages (channel_id, stream_id, ip_address, username, message) VALUES (?, ?, ?, ?, ?)', [channelId, streamId, ip, 'waiting', 'let me in']);
++ const review = await post('moderation', { key: 'ingress:review:1', action: 'review-pending-ip', id: Number(pending.lastInsertRowid), status: 'denied', channel_id: channelId, reviewed_by: streamer.id });
++ assert.strictEqual(review.status, 200, review.text);
++ assert.strictEqual((await h.db.get('SELECT status FROM pending_ip_messages WHERE id = ?', [pending.lastInsertRowid])).status, 'denied');
++ const pending2 = await h.db.run('INSERT INTO pending_ip_messages (channel_id, stream_id, ip_address, username, message) VALUES (?, ?, ?, ?, ?)', [channelId, streamId, ip, 'waiting', 'try again']);
++ assert.strictEqual((await post('moderation', { key: 'ingress:approve:1', action: 'approve-ip-messages', channel_id: channelId, ip, reviewed_by: streamer.id })).status, 200);
++ assert.strictEqual((await h.db.get('SELECT status FROM pending_ip_messages WHERE id = ?', [pending2.lastInsertRowid])).status, 'approved');
++
++ const hide = await post('moderation', { key: 'ingress:relay:hide', action: 'relay-hide', channel_id: channelId, platform: 'twitch', external_username: 'outside', created_by: streamer.id });
++ assert.strictEqual(hide.status, 200, hide.text);
++ assert.strictEqual((await h.db.get('SELECT COUNT(*) AS n FROM hidden_relay_users WHERE external_username = ?', ['outside'])).n, 1);
++ assert.strictEqual((await post('moderation', { key: 'ingress:relay:unhide', action: 'relay-unhide', id: hide.body.id })).status, 200);
++ assert.strictEqual((await h.db.get('SELECT COUNT(*) AS n FROM hidden_relay_users WHERE external_username = ?', ['outside'])).n, 0);
++ const voice = await post('moderation', { key: 'ingress:voice:set', action: 'tts-voice-override', identity_key: 'viewer', params: { voice: 'en-US', pitch: 1, speed: 1, gap: 0 }, set_by: streamer.id });
++ assert.strictEqual(voice.status, 200, voice.text);
++ assert.strictEqual((await h.db.getTtsVoiceOverride('viewer')).voice, 'en-US');
++ assert.strictEqual((await post('moderation', { key: 'ingress:voice:clear', action: 'tts-voice-override', identity_key: 'viewer' })).status, 200);
++ assert.strictEqual(await h.db.getTtsVoiceOverride('viewer'), null);
++});
++
++t('invalidate accepts typed hints; presence retains the bridge snapshot shape', async () => {
++ const hint = { key: 'ingress:hint:1', user: viewer.id, approvals: channelId, bans: true };
++ assert.strictEqual((await post('invalidate', hint)).status, 200);
++ assert.strictEqual((await post('invalidate', hint)).status, 200);
++ const changed = await post('invalidate', { key: 'ingress:hint:user-update', user: viewer.id,
++ user_data: { id: viewer.id, username: viewer.username, display_name: 'Renamed Viewer' } });
++ assert.strictEqual(changed.status, 200, changed.text);
++ assert.strictEqual((await room.next((m) => m.type === 'user-updated' && m.user.display_name === 'Renamed Viewer')).user.id, viewer.id);
++ assert.strictEqual((await h.db.get('SELECT username FROM chat_messages WHERE id = ?', [messageId])).username, 'Renamed Viewer');
++ assert.strictEqual((await post('invalidate', { key: 'ingress:hint:bad', nonsense: true })).status, 400);
++ const r = await h.http('GET', '/internal/chat/presence', { token: h.serviceToken(['chat.presence.read']) });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.ok(r.body.total >= 2);
++ assert.strictEqual(r.body.streams[streamId], 1);
++ assert.ok(r.body.users.some((u) => u.user_id === viewer.id));
++ assert.strictEqual((await h.http('GET', '/internal/live/presence', { token: h.serviceToken(['chat.presence.read']) })).status, 200);
++});
++
++t('moderation disconnect closes only the selected user socket', async () => {
++ const closed = new Promise((resolve) => room.on('close', resolve));
++ const r = await post('moderation', { key: 'ingress:disconnect:1', action: 'disconnect', user_id: viewer.id, stream_id: streamId });
++ assert.strictEqual(r.status, 200, r.text);
++ await Promise.race([closed, h.sleep(3000).then(() => { throw new Error('socket stayed open'); })]);
++ assert.strictEqual(globalRoom.readyState, 1);
++});
++
++t.run(async () => {
++ if (room) room.close();
++ if (globalRoom) globalRoom.close();
++ if (h) await h.close();
++});
ddiff --git a/test/pg-schema.test.js b/test/pg-schema.test.js
iindex 4a89b94..0acc648 100644
---- a/test/pg-schema.test.js
++++ b/test/pg-schema.test.js
@@@ -18,14 +18,13 @@ let db, close;
async function all(sql, params) { return await db.prepare(sql).all(...(params || [])); }
async function get(sql, params) { return await db.prepare(sql).get(...(params || [])); }
--// Every table Chat owns: 34 in schema.sql plus the seven modules created at runtime (rooms*, token
--// revocations, network blocks, account-data events) — 41, and the bookkeeping ones among them.
++// Every table Chat owns, including the Chat ingress retry ledger and runtime-created modules.
const TABLES = [
'chat_messages', 'dm_conversations', 'dm_participants', 'dm_messages', 'dm_blocks', 'tts_voice_overrides',
'channel_sounds', 'relay_users', 'hidden_relay_users', 'pending_ip_messages', 'stream_first_chats', 'moderation_actions',
'channel_moderators', 'channel_moderation_settings', 'emotes', 'user_tags', 'chat_ai_summaries', 'chat_timeline_events',
'ctx_users', 'ctx_streams', 'ctx_managed_streams', 'ctx_channels', 'ctx_sync', 'events_outbox', 'live_mirror_outbox',
-- 'bridge_applied', 'audio_requests', 'bridge_refs', 'import_hold', 'import_runs', 'chat_meta', 'deploy_releases',
++ 'bridge_applied', 'chat_ingress_applied', 'audio_requests', 'bridge_refs', 'import_hold', 'import_runs', 'chat_meta', 'deploy_releases',
'chat_event_inbox', 'calls', 'rooms', 'room_members', 'room_messages', 'room_attachments', 'token_revocations',
'network_blocks', 'account_data_events',
];
ddiff --git a/test/rehearsal.test.js b/test/rehearsal.test.js
nnew file mode 100644
iindex 0000000..4775c2d
---- /dev/null
++++ b/test/rehearsal.test.js
@@@ -0,0 +1,51 @@
++'use strict';
++/**
++ * The cutover rehearsal's seed (docs/cutover.md's ```rehearse block → test/rehearsal/seed.sql).
++ * `ov rehearse` loads the declared seed on a fresh database after main's migrations and before the
++ * branch's, so the seed must be PostgreSQL and must only name columns of migrations/0001 and 0002 —
++ * not chat_ingress_applied, which 0003 creates. Without the declaration the harness would load every
++ * test/fixtures/*.sql instead, and test/fixtures/live-chat-schema.sql is Live's SQLite schema.
++ * This reproduces that exact order on an in-process PostgreSQL (PGlite).
++ */
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const { createDb } = require('openvibe-sdk/db');
++const { suite } = require('./helpers');
++
++const t = suite('rehearsal');
++const ROOT = path.join(__dirname, '..');
++const MIGRATIONS = path.join(ROOT, 'migrations');
++const RUNBOOK = path.join(ROOT, 'docs', 'cutover.md');
++const SEED = path.join(ROOT, 'test', 'rehearsal', 'seed.sql');
++
++let db;
++t('the runbook declares only the PostgreSQL seed', () => {
++ const block = /^```rehearse[ \t]*\n([\s\S]*?)^```/gm.exec(fs.readFileSync(RUNBOOK, 'utf8'));
++ assert.ok(block, "docs/cutover.md has a ```rehearse block");
++ const seeds = block[1].split('\n').map((l) => l.trim()).filter((l) => l.startsWith('seed:'))
++ .map((l) => l.slice(5).trim());
++ assert.deepStrictEqual(seeds, ['test/rehearsal/seed.sql'], 'the only seed is test/rehearsal/seed.sql');
++});
++
++t('the seed applies after 0001–0002 and before 0003', async () => {
++ const base = fs.mkdtempSync(path.join(os.tmpdir(), 'rh-base-'));
++ for (const f of ['0001_initial.sql', '0002_audio_claim_owner.sql']) fs.copyFileSync(path.join(MIGRATIONS, f), path.join(base, f));
++ db = createDb({ pglite: true, service: 'rehearsal-test', log: { log() {}, warn() {}, error() {} } });
++ const first = await db.migrate({ dir: base, log: { log() {} } });
++ assert.deepStrictEqual(first.applied.map((a) => a.id), ['0001', '0002'], 'main\'s migrations applied');
++ await db.query(fs.readFileSync(SEED, 'utf8'));
++ for (const [table, n] of [['chat_messages', 3], ['dm_conversations', 1], ['dm_participants', 2], ['dm_messages', 2], ['bridge_applied', 1]]) {
++ const row = await db.prepare(`SELECT COUNT(*) AS n FROM ${table}`).get();
++ assert.strictEqual(row.n, n, `${table} has ${n} seeded row(s)`);
++ }
++ assert.strictEqual((await db.prepare(`SELECT to_regclass('chat_ingress_applied') AS t`).get()).t, null, '0003 has not run yet');
++ const second = await db.migrate({ dir: MIGRATIONS, log: { log() {} } });
++ assert.strictEqual(second.held.length, 0, 'the branch migration is not held');
++ assert.deepStrictEqual(second.applied.map((a) => a.id), ['0003'], 'only the branch\'s migration applies after the seed');
++});
++
++t('cleanup', async () => { await db.close(); });
++
++t.run();
ddiff --git a/test/rehearsal/seed.sql b/test/rehearsal/seed.sql
nnew file mode 100644
iindex 0000000..3d12467
---- /dev/null
++++ b/test/rehearsal/seed.sql
@@@ -0,0 +1,29 @@
++-- PostgreSQL seed for `ov rehearse` (declared in docs/cutover.md's rehearse block); test/fixtures/*.sql are SQLite fixtures for the import tests, not seeds.
++--
++-- Loaded on a fresh scratch database *after* main's migrations (migrations/0001_initial.sql,
++-- 0002_audio_claim_owner.sql) and *before* this branch's migrations (0003_chat_ingress.sql), so it
++-- may only name columns those two create and must not touch chat_ingress_applied (0003's table).
++-- Plain PostgreSQL INSERTs, no SQLite syntax. The database is empty, so identity columns are left
++-- to their defaults and the first dm_conversations row is id 1, which the child rows reference.
++
++-- A stream's chat: two regular lines and a system notice.
++INSERT INTO chat_messages (stream_id, user_id, anon_id, username, message, message_type, source_platform, timestamp, channel_user_id, subject_id) VALUES
++ (9001, 42, NULL, 'alice', 'first!', 'chat', 'openvibe', '2026-10-01 12:00:00', 42, 'usr_alice'),
++ (9001, NULL, 'anon1', 'guest', 'hi from a guest', 'chat', 'openvibe', '2026-10-01 12:00:05', 42, NULL),
++ (9001, NULL, NULL, NULL, 'guest joined the chat', 'system', 'openvibe', '2026-10-01 12:00:06', 42, NULL);
++
++-- One direct conversation between alice and bob (id 1), with its participants and two lines.
++INSERT INTO dm_conversations (name, is_group, created_by, created_at, updated_at) VALUES
++ (NULL, 0, 42, '2026-10-01 12:01:00', '2026-10-01 12:01:45');
++
++INSERT INTO dm_participants (conversation_id, user_id, last_read_at, joined_at, subject_id) VALUES
++ (1, 42, '1970-01-01 00:00:00', '2026-10-01 12:01:00', 'usr_alice'),
++ (1, 43, '1970-01-01 00:00:00', '2026-10-01 12:01:00', 'usr_bob');
++
++INSERT INTO dm_messages (conversation_id, sender_id, message, created_at, sender_subject_id) VALUES
++ (1, 42, 'hey bob', '2026-10-01 12:01:30', 'usr_alice'),
++ (1, 43, 'hey alice', '2026-10-01 12:01:45', 'usr_bob');
++
++-- A bridge delivery already applied once (its idempotency key would return this result on a retry).
++INSERT INTO bridge_applied (key, result, applied_at) VALUES
++ ('rehearsal:seed:1', '{"ok":true}', 1788220800000);