openvibers/openvibe.live Commits

Deliver chat only through Chat's ingress and drop the bridge outbox (#21)

OP OpenVibers committed 4 Oct 2026 · c1bd342ca5 · 29 files, +695 −582

local checks ✓ 3/3 · CI ✓ 3/3 · review SHIP by Claude opus (QUALITY=high) · 3 fix rounds

https://github.com/OpenVibers/OpenVibe.Live/pull/21

Changes

ddiff --git a/.env.example b/.env.example
iindex 0182294e..2c1f3853 100644
---- a/.env.example
++++ b/.env.example
@@@ -165,17 +165,14 @@ OV_COMMUNITY_URL=https://openvibe.community
OV_COMMUNITY_INTERNAL_URL=http://127.0.0.1:4200
# Chat runs in OpenVibe.Chat when this is "chat" (roadmap W6; server/chat/chat-authority.js): Live
--# stops serving /ws/chat and /api/chat|dm|tts|sounds (nginx sends them to Chat), forwards its own
--# chat calls to Chat and keeps its chat tables as Chat's read mirror. Unset = Live runs chat.
++# stops serving /ws/chat and /api/chat|dm|tts|sounds (nginx sends them to Chat), delivers its own
++# chat producers (AI viewer lines, donations and alerts, relay and RobotStreamer chat, cards, moderation,
++# cache hints, presence) through Chat's typed ingress /internal/chat/* (server/chat/chat-delivery.js; Live's
++# principal needs chat.message.send, chat.event.publish, chat.moderation.write, chat.cache.invalidate and
++# chat.presence.read for audience openvibe.chat) and keeps its chat tables as Chat's read mirror.
++# Unset = Live runs chat.
#CHAT_AUTHORITY=chat
OV_CHAT_INTERNAL_URL=http://127.0.0.1:4400
--# With CHAT_AUTHORITY=chat: "1" sends Live's chat producers (AI viewer lines, donations and alerts, relay
--# and RobotStreamer chat, cards, moderation, cache hints, presence) to Chat's typed ingress
--# /internal/chat/* (server/chat/chat-delivery.js) instead of the chat-remote.js bridge. A Chat 4xx/5xx is
--# logged and counted, never thrown and never handed to the bridge. Turn-on order: Network grants Live
--# chat.event.publish, chat.moderation.write and chat.cache.invalidate (audience openvibe.chat) → Chat's
--# ingress is deployed → set this → watch chat_bridge_outbox stop growing → then the bridge removal.
--#LIVE_CHAT_INGRESS=1
# Bot panel embed (server/bot/): "1" lets a channel owner bind the channel to an OpenVibe.Bot robot
# (PUT /api/streams/channel/:username/bot) and adds bot_embed { enabled, robot_id, url } to the channel
# JSON; off, the route answers 404 and the JSON is unchanged. Live never calls Bot to bind: the robot
ddiff --git a/README.md b/README.md
iindex b8d52d1d..c281e685 100644
---- a/README.md
++++ b/README.md
@@@ -96,20 +96,14 @@ Live's own principal (`live`) holds grants to call:
- **Network**: `identity.subject.resolve`, `network.follows.read|write`, `network.modules.read|write`, `network.notifications.push`, `network.coins.credit|debit`, `network.analytics.creator.read`, `network.account.export.contribute`, `network.account.deletion.confirm`.
- **AI**: `ai.run.create|read` (namespaces `live.*`, `network.site_copy`, `media.analyze`), `ai.credential.manage`, `ai.quota.attribution.manage`.
--- **Chat**: `chat.live_bridge.write`, `chat.message.send`, `chat.presence.read`; with `LIVE_CHAT_INGRESS=1` also `chat.event.publish`, `chat.moderation.write`, `chat.cache.invalidate`.
++- **Chat**: `chat.message.send`, `chat.event.publish`, `chat.moderation.write`, `chat.cache.invalidate`, `chat.presence.read`; `chat.live_bridge.write` for the call channels (`server/streaming/calls-authority.js`) and the one-off outbox delivery (`scripts/chat-bridge-outbox-drop.js`).
- **Community**: `community.paste.*`, `community.comment.*`, `community.pulse.write`.
- **Events**: `events.event.publish|read`, `events.subscription.manage`.
- **OpenRe**: `openre.stream.*`, `openre.key.rotate`, `openre.session.read`.
- **Billing**: `billing.*` (intents, transfers, subscriptions, cash-outs, balances, entitlements).
- **Elsewhere**: `tips.interaction.record`, `vip.entitlement.check`, `tools.tool.run`, `tools.job.read`.
--**Chat ingress (`LIVE_CHAT_INGRESS`, off by default).** With `CHAT_AUTHORITY=chat` and `LIVE_CHAT_INGRESS=1`, every chat producer in Live goes through one seam, `server/chat/chat-delivery.js`, to OpenVibe.Chat's typed service-token ingress (`server/chat/chat-client.js`: `POST /internal/chat/messages|events|moderation|invalidate`, `GET /internal/chat/presence`) instead of the `chat-remote.js` bridge. Each operation carries one idempotency key, reused on every retry. A Chat 4xx is logged, counted and dropped; a 5xx or timeout is retried with the same key past Chat's five-minute delivery lease, then logged, counted and dropped. Neither is thrown into the caller, and neither falls back to the bridge (that would deliver twice). Turn it on in this order:
--
--1. Network grants Live `chat.event.publish`, `chat.moderation.write` and `chat.cache.invalidate` (audience `openvibe.chat`).
--2. Chat's ingress is deployed.
--3. Set `LIVE_CHAT_INGRESS=1` and restart Live.
--4. Watch `chat_bridge_outbox` stop growing.
--5. Then the removal PR (the bridge writers and the outbox drop) and Chat's J3.
++**Chat ingress.** With `CHAT_AUTHORITY=chat`, every chat producer in Live goes through one seam, `server/chat/chat-delivery.js`, to OpenVibe.Chat's typed service-token ingress (`server/chat/chat-client.js`: `POST /internal/chat/messages|events|moderation|invalidate`, `GET /internal/chat/presence`) — the only path; the old `chat-remote.js` bridge is gone (T3 J2). Each operation carries one idempotency key, reused on every retry. A Chat 4xx is logged, counted and dropped; a 5xx or timeout is retried with the same key past Chat's five-minute delivery lease, then logged, counted and dropped. Neither is thrown into the caller. The old bridge's `chat_bridge_outbox` is dropped by an operator step once the release before is out of rollback range: `node scripts/chat-bridge-outbox-drop.js` (dry run), `--deliver` (hands Chat every queued write it never acknowledged), then `--apply` (backup, then `op_002_drop_chat_bridge_outbox`, which refuses while a chat write is still queued).
API writes are limited per person (`server/net/actor-limits.js`, `LIVE_LIMITS_MINUTE` / `_HOUR`).
ddiff --git a/docs/chat-system.md b/docs/chat-system.md
iindex 4abc7830..0ccf3a92 100644
---- a/docs/chat-system.md
++++ b/docs/chat-system.md
@@@ -16,20 +16,27 @@ with the same WebSocket protocol and REST paths. `CHAT_AUTHORITY=chat` switches
- nginx sends `/ws/chat`, `/api/chat/`, `/api/dm/`, `/api/tts/` and `/api/sounds` to Chat; Live answers
them with 503 if one still arrives, and `/ws/chat` upgrades are refused.
--- `require('./chat/chat-server')` returns `chat-remote.js`: the same methods, forwarded in order to
-- Chat's `POST /internal/live/calls` (broadcasts, DMs, TTS, channel sounds, disconnects, user updates).
-- Chat-table writes Live's other modules make (AI viewers, relays, donations, `/api/mod`) are forwarded
-- too; inserts return a placeholder id (≤ -2^40) Chat maps to the real one. Unacknowledged writes wait
-- in `chat_bridge_outbox` (and are applied here if Live boots without the flag again — rollback).
--- Synchronous reads (`getTotalConnections`, viewer counts, slow modes, a connected user's IP) come from
-- Chat's presence snapshot, polled every 3 s.
--- **The bridge writers stay** (T3 J2 deferred). Live still generates the messages Chat delivers —
-- AI viewer replies, relays, donations, `/api/mod` deletes, deploy notices — and alert sounds are
-- requested from Live (`playAlertSound`). Chat does not yet consume those from an event or generate
-- them itself, so `chat-remote.js` and its `chat_bridge_outbox` must keep forwarding them: removing
-- the writers first sends nothing to Chat and viewers lose the messages (review 2026-10-02, PR #12).
-- Retire them only once Chat owns each path, and drop the outbox only after any unacknowledged
-- `op = 'db'` rows are delivered (`test/chat-context.test.js` locks the paths).
++- Live's chat producers (AI viewer lines, relayed and RobotStreamer chat, donations and alert sounds,
++ cards, `/api/mod` deletes and moderation, cache hints) go through one seam,
++ [server/chat/chat-delivery.js](../server/chat/chat-delivery.js), to Chat's typed service-token ingress
++ ([server/chat/chat-client.js](../server/chat/chat-client.js): `POST /internal/chat/messages|events|moderation|invalidate`,
++ `GET /internal/chat/presence`). Chat persists, broadcasts, mirrors and speaks (`tts`) in one call per
++ operation; each carries one idempotency key reused on every retry. A Chat 4xx is logged and dropped; a 5xx
++ or timeout is retried past Chat's five-minute delivery lease, then logged and dropped; neither is thrown.
++- `require('./chat/chat-server')` returns a `RemoteChatServer` that never listens: synchronous reads
++ (`getTotalConnections`, viewer counts, slow modes, a connected user's IP) come from Chat's presence
++ snapshot, polled every 3 s; Live's IP-approval and ban writes send Chat a cache hint; a push a module still
++ makes on it goes to the same ingress, and TTS is never synthesised in Live. The arena commands
++ (`/internal/chat-effects/arena-command`) answer the sender in the response's `replies`.
++- Deploy notices reach Chat only as the `live.release.deployed` event: with Events publishing off the commits
++ stay unannounced and the next boot tries again.
++- **The old ordered-calls bridge is gone** (T3 J2): no `chat-remote.js`, no `POST /internal/live/calls`, no
++ placeholder ids, and nothing under `server/` reads or writes `chat_bridge_outbox`
++ (`test/chat-bridge-removed.test.js`). The table is dropped by the operator migration
++ `op_002_drop_chat_bridge_outbox` through `scripts/chat-bridge-outbox-drop.js`, once the release before is
++ out of rollback range: `--deliver` first hands Chat (its bridge receiver, until Chat's J3) every chat write
++ that release queued and Chat never acknowledged, then `--apply` takes a backup and drops the table; the drop
++ refuses while any `op = 'db'` row remains.
- Chat reads Live data and asks for side effects on `/internal/chat-context/*` and
`/internal/chat-effects/*` (`server/chat/live-context-routes.js`), with Network service tokens
(`live.chat_context.read`, `live.chat_effects.write`, `live.chat_mirror.write`,
@@@ -42,8 +49,8 @@ with the same WebSocket protocol and REST paths. `CHAT_AUTHORITY=chat` switches
only through [server/chat/moderation-client.js](../server/chat/moderation-client.js) — Chat's internal
read API (`GET /internal/moderation/...`, capability `chat.moderation.read`, cached 30 s, off under
`LIVE_DRILL`). The dashboard, channel page and upload UI call Chat directly (`/api/chat/channels/:id/…`,
-- `/api/emotes`, `/api/chat/ai/…`); alert sounds are Chat's, played on Live's request through the bridge op
-- `playAlertSound [streamerId, streamId, kind]` ([server/monetization/alerts.js](../server/monetization/alerts.js)).
++ `/api/emotes`, `/api/chat/ai/…`); alert sounds are Chat's, played on Live's request as an `alert`
++ event `{ streamerId, streamId, kind }` on Chat's ingress ([server/monetization/alerts.js](../server/monetization/alerts.js)).
Live's staged-table machinery (the write relay, dual read and handoff in `chat-tables*.js`), Live's
chat-AI summary job and the old emote/channel-moderation routes were deleted in the N+1 release; Live's
copies of the seven supporting tables were dropped in N+2, and the unread `emotes` copy in N+3 (ADR-016).
ddiff --git a/docs/cutover-pr-openvibe-live-21.md b/docs/cutover-pr-openvibe-live-21.md
nnew file mode 100644
iindex 00000000..0a7cd901
---- /dev/null
++++ b/docs/cutover-pr-openvibe-live-21.md
@@@ -0,0 +1,94 @@
++# Cutover: drop Live's `chat_bridge_outbox` (T3 J2, operator migration)
++
++This PR ("Deliver chat only through Chat's ingress and drop the bridge outbox") deletes Live's old ordered-calls
++chat bridge ([server/chat/chat-remote.js](../server/chat/chat-remote.js), `POST /internal/live/calls`) and leaves
++[server/chat/chat-delivery.js](../server/chat/chat-delivery.js) as the only path into OpenVibe.Chat, through Chat's
++typed ingress. It adds the operator migration **`op_002_drop_chat_bridge_outbox`**
++([server/db/migrations.js](../server/db/migrations.js)) and the script that runs it,
++[scripts/chat-bridge-outbox-drop.js](../scripts/chat-bridge-outbox-drop.js).
++
++Unlike the boot migrations (`001…006`), `op_002` is **not** applied on boot: the release before this one still
++writes `chat_bridge_outbox`, so the table may only be dropped once that release is out of rollback range
++(ADR-028). The operator runs the script by hand after this release has been stable for the rollback window.
++
++`chat_bridge_outbox` is Live's own SQLite table, so this cutover is on Live's `live.db`, not on the shared
++PostgreSQL. The script has three modes:
++
++- **(default) dry run** — prints the queued rows by op and boot; changes nothing.
++- **`--deliver`** — hands every queued row to Chat's bridge receiver (`POST /internal/live/calls`, Live's
++ service token for audience `openvibe.chat`, capability `chat.live_bridge.write`), in the order and boot the
++ old bridge wrote them, each with idempotency key `live:<row id>` so Chat applies a write at most once.
++ Acknowledged rows are deleted; a row Chat refuses is left for the operator; a transport error stops the run
++ to be rerun later.
++- **`--apply`** — takes an online backup of `live.db`, then runs `op_002`, which
++ **refuses while any unacknowledged chat write (`op = 'db'`) is still queued**, so a write the old release queued is never lost.
++
++## Order
++
++1. Confirm this release is the running one and has been through the rollback window:
++ `ov access run openvibe-ovh health live` answers `ready` and `ov access run openvibe-ovh releases live`
++ shows the new release as current (the old writer no longer in rollback range).
++2. Stop nothing; `--deliver` and `--apply` run against the live database. From the release directory as the
++ service user: `node scripts/chat-bridge-outbox-drop.js` (dry run) — note the counts.
++3. If the dry run shows queued chat writes: `node scripts/chat-bridge-outbox-drop.js --deliver` and rerun the
++ dry run until it reports **0** chat writes (`op = 'db'`). Rows Chat refused stay and are reported with their
++ ids; take those to the operator (they are not deleted).
++4. `node scripts/chat-bridge-outbox-drop.js --apply` — writes
++ `<database dir>/backups/live-pre-chat-bridge-drop-<time>.db` (mode 0600) and runs the drop. The ledger
++ outcome is printed (`applied`, or `adopted` on a database that never had the table, `already` on a repeat).
++
++## Backup
++
++- The script takes its own `live.db` backup before it drops anything and prints the path (step 4). Restoring
++ that file is the way back to a database that still holds the table's rows.
++- A wider snapshot is also taken by the host: `ov access run openvibe-ovh db-backup` records Live's
++ `<backupDir>/live/<YYYYMMDD-HHMMSS>/live.db` with `ok: true` in `<stateDir>/backups/live.jsonl`. Either copy
++ is enough; keep the script's own file as it is the exact pre-drop state.
++
++## How to check success (verification)
++
++- **Delivery first.** After `--deliver`, the dry run prints `chat writes (op = 'db') Chat has not
++ acknowledged: 0`. The delivered messages appear in Chat (reload a watch page: they load from Chat).
++- **Drop applied.** `--apply` prints `op_002_drop_chat_bridge_outbox: applied`, and the read-only check on the
++ host prints nothing for the table and `applied` for the ledger row:
++
++```sh
++sqlite3 -readonly <Live's data dir>/live.db \
++ "SELECT name FROM sqlite_master WHERE name = 'chat_bridge_outbox'" # → nothing
++sqlite3 -readonly <Live's data dir>/live.db \
++ "SELECT id, mode FROM schema_migrations WHERE id = 'op_002_drop_chat_bridge_outbox'" # → op_002_…|applied
++```
++
++- **Nothing else broke.** `ov access run openvibe-ovh health live` answers `ready`; restarting Live does not
++ print the migration again, and its logs contain no `no such table: chat_bridge_outbox`. Live's chat still
++ delivers: send a chat message and see it in the Chat-served history, and the deploy-notice path still fires.
++- **Idempotent.** Rerunning `--apply` prints `already` and drops nothing further.
++
++## Rollback / restore
++
++Dropping the table is destructive and a code rollback does **not** bring it back — the previous release runs
++fine without the table (it is a chat send queue, not state). Only restore if a queued chat write turns out to
++have been needed:
++
++1. `ov access run openvibe-ovh rollback live` (back to the previous release).
++2. Stop Live and copy the script's backup (or the host `db-backup`) `live.db` over Live's database, as the
++ service user, with the original owner and mode and no stale `-wal`/`-shm` beside it; start Live.
++3. `ov access run openvibe-ovh health live`: `ready`, then `--deliver` to flush the restored queue.
++
++Restoring also discards every other Live write made since the backup, so prefer leaving the table dropped.
++Stop Live only for the copy; `--deliver`/`--apply` themselves run online.
++
++## Rehearsal
++
++The migration and the delivery/drop contract are exercised by [test/chat-bridge-removed.test.js](../test/chat-bridge-removed.test.js)
++against an in-memory SQLite database: it builds `chat_bridge_outbox`, checks that `op_002` refuses while a chat
++write is queued, that `deliver()` sends queued rows in order per boot with keys `live:<id>` and deletes only
++what Chat acknowledged, and that the drop then applies (and is `already` on a second run). The PR's PostgreSQL
++migrations (`migrations/`) are unchanged, so the harness base/migrate steps apply nothing new; the command below
++is the runbook's part.
++
++```rehearse
++seed: none
++migrations: migrations
++node test/chat-bridge-removed.test.js
++```
ddiff --git a/scripts/chat-bridge-outbox-drop.js b/scripts/chat-bridge-outbox-drop.js
nnew file mode 100644
iindex 00000000..cb7ad218
---- /dev/null
++++ b/scripts/chat-bridge-outbox-drop.js
@@@ -0,0 +1,172 @@
++#!/usr/bin/env node
++/**
++ * The contract step for Live's old chat bridge (roadmap T3 J2): Live now delivers to OpenVibe.Chat through
++ * its typed ingress only, and nothing under server/ reads or writes chat_bridge_outbox. Run it once the
++ * release before (which still wrote the outbox) is out of rollback range.
++ *
++ * node scripts/chat-bridge-outbox-drop.js [--db <live.db>]
++ * # dry run (the default): what is still queued, by op and boot; changes nothing
++ * node scripts/chat-bridge-outbox-drop.js --deliver [--db <live.db>]
++ * # hands every queued row to Chat's bridge receiver (POST /internal/live/calls, Live's service token
++ * # for audience openvibe.chat, capability chat.live_bridge.write — Chat serves it until its J3), in
++ * # the order and boot they were written, each with its idempotency key live:<row id> so Chat applies
++ * # a write at most once; acknowledged rows are deleted
++ * node scripts/chat-bridge-outbox-drop.js --apply [--backup <file.db>] [--db <live.db>]
++ * # an online backup first (default <database dir>/backups/live-pre-chat-bridge-drop-<time>.db, mode
++ * # 0600), then server/db/migrations.js operator migration op_002_drop_chat_bridge_outbox, which
++ * # refuses while any chat write (op = 'db') is still queued
++ *
++ * --deliver needs Live's environment (OV_CHAT_INTERNAL_URL, OV_NETWORK_INTERNAL_URL, OV_OAUTH_CLIENT_ID/
++ * SECRET). Printed: counts, ops and row ids only, never message text.
++ * Default database: $DB_PATH, else data/live.db under $DATA_DIR or the working directory (server/paths.js).
++ * In production run it from /opt/openvibe.live/current as the service user (data -> ../../shared/data).
++ */
++'use strict';
++const fs = require('fs');
++const path = require('path');
++
++const ID = 'op_002_drop_chat_bridge_outbox';
++const TABLE = 'chat_bridge_outbox';
++const BATCH = 200;
++
++function parseArgs(argv) {
++ const opts = { mode: 'dry', db: null, backup: null, help: false };
++ for (let i = 0; i < argv.length; i++) {
++ const a = argv[i];
++ const next = () => {
++ const v = argv[++i];
++ if (v === undefined || v.startsWith('--')) throw new Error(`${a} needs a value`);
++ return v;
++ };
++ if (a === '--apply') opts.mode = 'apply';
++ else if (a === '--deliver') opts.mode = 'deliver';
++ else if (a === '--dry-run') opts.mode = 'dry';
++ else if (a === '--db') opts.db = next();
++ else if (a === '--backup') opts.backup = next();
++ else if (a === '--help' || a === '-h') opts.help = true;
++ else throw new Error(`unknown option ${a}`);
++ }
++ return opts;
++}
++
++const hasTable = (db) => !!db.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?").get(TABLE);
++
++/** What is still queued: { exists, total, writes, byOp: { op: n }, boots }. Reads only. */
++function counts(db) {
++ if (!hasTable(db)) return { exists: false, total: 0, writes: 0, byOp: {}, boots: 0 };
++ const byOp = {};
++ for (const r of db.prepare(`SELECT op, COUNT(*) AS n FROM ${TABLE} GROUP BY op`).all()) byOp[r.op] = r.n;
++ const total = Object.values(byOp).reduce((a, b) => a + b, 0);
++ const boots = db.prepare(`SELECT COUNT(DISTINCT boot) AS n FROM ${TABLE}`).get().n;
++ return { exists: true, total, writes: byOp.db || 0, byOp, boots };
++}
++
++function report(c, log) {
++ if (!c.exists) { log(`${TABLE}: absent`); return; }
++ log(`${TABLE}: ${c.total} row(s) from ${c.boots} boot(s); chat writes (op = 'db') Chat has not acknowledged: ${c.writes}`);
++ for (const [op, n] of Object.entries(c.byOp)) log(` ${op}: ${n}`);
++}
++
++/**
++ * Deliver every queued row the way the old bridge did. post(path, body) → Chat's JSON answer (throws with
++ * err.status on a non-2xx). Stops at the first transport/5xx error (rerun later); a row Chat refuses on its
++ * own (400/413 for a one-row batch) can never be delivered and is left for the operator, not deleted.
++ * → { delivered, refused: [ids], error }
++ */
++async function deliver(db, post, log = () => {}) {
++ const out = { delivered: 0, refused: [], error: null };
++ // A batch Chat refuses whole is rebuilt row by row, so the same unparseable row can be seen twice.
++ const refused = new Set();
++ if (!hasTable(db)) return out;
++ const rows = db.prepare(`SELECT id, boot, ref, op, args FROM ${TABLE} ORDER BY id`).all();
++ const del = db.prepare(`DELETE FROM ${TABLE} WHERE id = ?`);
++ let i = 0;
++ let limit = BATCH;
++ while (i < rows.length) {
++ const boot = rows[i].boot;
++ const batch = [];
++ for (let j = i; j < rows.length && rows[j].boot === boot && batch.length < limit; j++) batch.push(rows[j]);
++ const ops = [];
++ for (const r of batch) {
++ let args;
++ try { args = JSON.parse(r.args); } catch { args = null; }
++ if (!Array.isArray(args)) { refused.add(r.id); continue; }
++ ops.push({ seq: r.id, op: r.op, args, ref: r.ref, key: `live:${r.id}` });
++ }
++ try {
++ const res = ops.length ? await post('/internal/live/calls', { boot, ops }) : { results: [] };
++ const refusedSeq = new Set(((res && res.results) || []).filter((x) => !x.ok).map((x) => x.seq));
++ db.transaction(() => {
++ for (const o of ops) {
++ if (refusedSeq.has(o.seq)) { refused.add(o.seq); continue; }
++ del.run(o.seq);
++ out.delivered++;
++ }
++ })();
++ i += batch.length;
++ limit = BATCH;
++ } catch (err) {
++ if ((err.status === 400 || err.status === 413) && batch.length > 1) { limit = 1; continue; }
++ if (err.status === 400 || err.status === 413) { refused.add(batch[0].id); i += 1; continue; }
++ out.error = err.message;
++ log(`stopped: ${err.message} (rerun --deliver later)`);
++ break;
++ }
++ }
++ out.refused = [...refused];
++ return out;
++}
++
++/** The old bridge's request: Live's service token for audience openvibe.chat, one retry on a 401. */
++function chatPost() {
++ const principal = require('../server/net/network-principal');
++ const { CHAT_URL } = require('../server/chat/chat-authority');
++ const post = async (p, body, retried = false) => {
++ const res = await fetch(`${CHAT_URL}${p}`, {
++ method: 'POST',
++ headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...(await principal.serviceHeaders('openvibe.chat')) },
++ body: JSON.stringify(body),
++ signal: AbortSignal.timeout(15000),
++ });
++ if (res.status === 401 && !retried) { principal.invalidate('openvibe.chat'); return post(p, body, true); }
++ if (!res.ok) { const err = new Error(`Chat ${res.status}`); err.status = res.status; throw err; }
++ return res.json();
++ };
++ return post;
++}
++
++async function backup(db, file) {
++ fs.mkdirSync(path.dirname(file), { recursive: true });
++ await db.backup(file);
++ fs.chmodSync(file, 0o600);
++ return file;
++}
++
++async function main(argv = process.argv.slice(2), log = console.log) {
++ const opts = parseArgs(argv);
++ if (opts.help) { log(fs.readFileSync(__filename, 'utf8').split('*/')[0]); return 0; }
++ const Database = require('better-sqlite3');
++ const DB_PATH = path.resolve(opts.db || require('../server/paths').dbPath());
++ const db = new Database(DB_PATH, { readonly: opts.mode === 'dry', fileMustExist: true });
++ try {
++ report(counts(db), log);
++ if (opts.mode === 'dry') return 0;
++ if (opts.mode === 'deliver') {
++ const r = await deliver(db, chatPost(), log);
++ log(`delivered ${r.delivered}; refused by Chat (left in place): ${r.refused.length ? r.refused.join(', ') : 'none'}`);
++ report(counts(db), log);
++ return r.error ? 1 : 0;
++ }
++ const file = opts.backup || path.join(path.dirname(DB_PATH), 'backups', `live-pre-chat-bridge-drop-${new Date().toISOString().replace(/[:.]/g, '-')}.db`);
++ log(`backup: ${await backup(db, file)}`);
++ const res = require('../server/db/migrations').runOperator(db, ID);
++ log(`${ID}: ${res.outcome}${res.error ? ` (${res.error})` : ''}`);
++ return res.outcome === 'failed' ? 1 : 0;
++ } finally { db.close(); }
++}
++
++if (require.main === module) {
++ main().then((code) => process.exit(code), (err) => { console.error(err.message); process.exit(1); });
++}
++
++module.exports = { ID, counts, deliver, parseArgs, main };
ddiff --git a/server/admin/mod-routes.js b/server/admin/mod-routes.js
iindex 4cbffa96..3e22bdf2 100644
---- a/server/admin/mod-routes.js
++++ b/server/admin/mod-routes.js
@@@ -386,7 +386,7 @@ router.post('/delete-message', async (req, res) => {
if (!allowed) return res.status(403).json({ error: 'You cannot moderate this stream' });
}
-- // LIVE_CHAT_INGRESS: Chat deletes the row and broadcasts the delete to every surface it reached.
++ // Chat ingress: Chat deletes the row and broadcasts the delete to every surface it reached.
if (delivery.ingress()) {
delivery.mirror('deleteChatMessage', parseInt(message_id), req.user.id);
await delivery.moderate('delete-message', { id: parseInt(message_id), deleted_by: req.user.id }, { key: `delete:${message_id}` });
@@@ -649,7 +649,7 @@ router.post('/ip-approval/:channelId/approve', async (req, res) => {
[channelId, ip]
);
-- // LIVE_CHAT_INGRESS: Chat approves its held rows; re-showing them live is Chat's (it holds them).
++ // Chat ingress: Chat approves its held rows; re-showing them live is Chat's (it holds them).
if (delivery.ingress()) {
delivery.mirror('approveAllFromIp', channelId, ip, req.user.id);
await delivery.moderate('approve-ip-messages', { channel_id: parseInt(channelId), ip, reviewed_by: req.user.id });
ddiff --git a/server/admin/routes.js b/server/admin/routes.js
iindex 4d79e4de..f500f72d 100644
---- a/server/admin/routes.js
++++ b/server/admin/routes.js
@@@ -246,7 +246,7 @@ router.put('/users/:id', (req, res) => {
// Push real-time update to the affected user's chat connections
if (updates.length > 0) {
const id = parseInt(req.params.id);
-- // LIVE_CHAT_INGRESS: a typed account hint; Chat updates the sockets and stored message names.
++ // Chat ingress: a typed account hint; Chat updates the sockets and stored message names.
delivery.invalidate({
user: id,
user_data: { id, username: safeUser.username, display_name: safeUser.display_name || null, role: safeUser.role || null, avatar_url: safeUser.avatar_url || null, profile_color: safeUser.profile_color || null },
ddiff --git a/server/arena/arena-chat.js b/server/arena/arena-chat.js
iindex c171c3c8..a37fb383 100644
---- a/server/arena/arena-chat.js
++++ b/server/arena/arena-chat.js
@@@ -49,7 +49,8 @@ function handle(chat, ws, client, cmd, parts) {
const stream = client.streamId ? db.getStreamById(client.streamId) : null;
const streamer = stream ? db.getUserById(stream.user_id) : null;
-- (async () => {
++ // A caller that answers once (Chat's /internal/chat-effects/arena-command) waits for the replies via chat.track.
++ const work = (async () => {
try {
if (cmd === '!hype') {
if (!streamer) return reply("!hype works inside a streamer's chat.");
@@@ -86,6 +87,7 @@ function handle(chat, ws, client, cmd, parts) {
reply(`Arena: ${err.message}`);
}
})();
++ if (typeof chat.track === 'function') chat.track(work);
return true;
}
ddiff --git a/server/auth/network-events.js b/server/auth/network-events.js
iindex 327288c6..a69c451a 100644
---- a/server/auth/network-events.js
++++ b/server/auth/network-events.js
@@@ -26,8 +26,7 @@ const stats = { received: 0, revoked: 0, unchanged: 0, profiles: 0, ignored: 0,
const notifyChat = (userId) => {
try {
const delivery = require('../chat/chat-delivery');
-- if (delivery.ingress()) { delivery.invalidate({ user: Number(userId) }); return; }
-- const cs = require('../chat/chat-server'); if (cs.remote) cs.userChanged(userId);
++ if (delivery.ingress()) delivery.invalidate({ user: Number(userId) });
} catch { /* non-critical */ }
};
ddiff --git a/server/chat/chat-authority.js b/server/chat/chat-authority.js
iindex 4867b0b5..c1d75e7e 100644
---- a/server/chat/chat-authority.js
++++ b/server/chat/chat-authority.js
@@@ -5,8 +5,9 @@
* CHAT_AUTHORITY=chat: OpenVibe.Chat (127.0.0.1:4400) serves /ws/chat and the chat REST routes
* (nginx routes them there) and owns the chat tables. Live then
* - does not start its chat WebSocket server or mount /api/chat, /api/dm, /api/tts, /api/sounds;
-- * - hands every chat call its other modules make to Chat (server/chat/chat-remote.js — what
-- * require('./chat/chat-server') returns in this mode);
++ * - delivers every chat call its other modules make through Chat's typed ingress
++ * (server/chat/chat-delivery.js → chat-client.js); require('./chat/chat-server') returns a
++ * RemoteChatServer that reads Chat's presence and never listens;
* - keeps its chat tables as a read mirror that Chat writes (POST /internal/chat-effects/mirror)
* so its own readers (home stats, recaps, AI context, /api/mod queues) keep working and a
* rollback loses nothing;
ddiff --git a/server/chat/chat-client.js b/server/chat/chat-client.js
iindex 539a8109..662c453f 100644
---- a/server/chat/chat-client.js
++++ b/server/chat/chat-client.js
@@@ -1,6 +1,6 @@
'use strict';
/**
-- * OpenVibe.Chat's typed service-token ingress (T3 J2), the replacement for the chat-remote.js bridge.
++ * OpenVibe.Chat's typed service-token ingress (T3 J2): Live's only delivery path to Chat.
*
* message(body) POST /internal/chat/messages chat.message.send persist + broadcast a line (or a DM)
* event(body) POST /internal/chat/events chat.event.publish a transient card, alert, sound, notice
@@@ -15,7 +15,7 @@
* Errors are logged and counted (`stats`), never thrown into the caller: every POST resolves to Chat's answer or
* null. A 4xx is final (Chat refused the body; retrying cannot help). A 5xx, a timeout or Chat being unreachable
* is retried with the same key on a back-off that outlasts Chat's five-minute delivery lease, then dropped. There
-- * is deliberately no fallback to the bridge: once Chat owns delivery, a second path would deliver twice.
++ * is deliberately no second path: it would deliver twice.
* Retries live in memory only (a Live restart drops the ones still waiting). Off under LIVE_DRILL.
*/
const crypto = require('crypto');
ddiff --git a/server/chat/chat-delivery.js b/server/chat/chat-delivery.js
iindex 55c4b1ad..c5976a1a 100644
---- a/server/chat/chat-delivery.js
++++ b/server/chat/chat-delivery.js
@@@ -1,31 +1,30 @@
'use strict';
/**
-- * The one seam Live's chat producers call (T3 J2, first half).
++ * The one seam Live's chat producers call (T3 J2).
*
-- * LIVE_CHAT_INGRESS=1 (with CHAT_AUTHORITY=chat): OpenVibe.Chat's typed ingress (chat-client.js) — Chat persists,
-- * broadcasts, mirrors, speaks and moderates in one call per operation, and nothing goes through the bridge.
-- * Otherwise (the default): the existing path — require('./chat-server'), which is the chat-remote.js bridge under
-- * CHAT_AUTHORITY=chat and Live's own ChatServer without it.
++ * CHAT_AUTHORITY=chat: OpenVibe.Chat's typed ingress (chat-client.js) — Chat persists, broadcasts, mirrors,
++ * speaks and moderates in one call per operation. This is the only path to Chat: the old ordered-calls bridge
++ * and its outbox are gone.
++ * Otherwise (rollback / dev): Live's own ChatServer — require('./chat-server').
*
-- * With the flag on, a Chat 4xx/5xx is logged and counted by chat-client.js and never thrown into the caller; the
-- * operation is not handed to the bridge (it would deliver twice once Chat owns delivery). Every function here
-- * returns a promise in ingress mode and the legacy call's own value otherwise.
++ * A Chat 4xx/5xx is logged and counted by chat-client.js and never thrown into the caller. Every function
++ * here returns a promise in ingress mode and the local call's own value otherwise.
*
-- * ingress() which path is active
++ * ingress() which path is active (CHAT_AUTHORITY=chat)
* message(body) ingress only: a chat line (Chat's /messages body) → Promise<real id | null>
* event(target, frame, opts) a transient frame; target { kind: stream|channel|global|all|user, id, stream }
* moderate(action, fields) ingress only: Chat's /moderation actions → Promise<Chat's answer | null>
* logModeration(entry) a moderation-log row (db.logModerationAction shape)
* disconnect({ userId, ip, streamId })
* invalidate(hint, legacy) a cache hint ({ user, user_data, approvals, bans, channel })
-- * mirror(fn, ...args) ingress only: keep Live's own copy of a bridge 'local' write (pending IP rows,
-- * hidden relay users, first chats, TTS overrides) that Live still reads
++ * mirror(fn, ...args) ingress only: keep Live's own copy of a write Chat also applies (pending IP
++ * rows, hidden relay users, first chats, TTS overrides) that Live still reads
* after(value, fn) fn(value) now for a plain value, after it resolves for a promise
*/
const chatAuthority = require('./chat-authority');
const client = require('./chat-client');
--function ingress() { return process.env.LIVE_CHAT_INGRESS === '1' && chatAuthority.isRemote(); }
++function ingress() { return chatAuthority.isRemote(); }
// Lazy: chat-server pulls in most of Live, and several producers are required by it.
function chatServer() { return require('./chat-server'); }
@@@ -84,18 +83,29 @@ function disconnect({ userId, ip, streamId } = {}) {
return moderate('disconnect', { user_id: userId || undefined, ip: ip || undefined, stream_id: streamId || undefined });
}
--/** legacy: what to run without the flag (cache hints had no single legacy call). */
++/** legacy: what to run when Live is the chat authority (cache hints had no single local call). */
function invalidate(hint, legacy) {
if (ingress()) return client.invalidate(defined(hint));
return legacy ? legacy() : undefined;
}
++// The writes Live keeps on its own read-mirror copy as well (Chat applies them too and mirrors back).
++const MIRROR_WRITES = new Set([
++ 'mergeChatMessageMetadata', 'deleteChatMessage', 'deleteUserChatMessages', 'deleteAnonChatMessages',
++ 'deleteRelayUserMessages', 'deleteChatMessagesByTimeRange', 'reviewPendingIpMessage', 'approveAllFromIp',
++ 'denyAllFromIp', 'recordRelayUser', 'unhideRelayUser', 'unhideRelayUserByIdentity', 'recordFirstChat',
++ 'setTtsVoiceOverride', 'deleteTtsVoiceOverride',
++]);
++
function mirror(fn, ...args) {
-- try { return require('./chat-remote').localWrite(fn, ...args); } catch (err) { console.warn(`[ChatIngress] local ${fn}: ${err.message}`); return undefined; }
++ try {
++ if (!MIRROR_WRITES.has(fn)) throw new Error('not a mirror write');
++ return require('../db/database')[fn](...args);
++ } catch (err) { console.warn(`[ChatIngress] local ${fn}: ${err.message}`); return undefined; }
}
function after(value, fn) {
return value && typeof value.then === 'function' ? value.then(fn) : fn(value);
}
--module.exports = { ingress, message, event, moderate, logModeration, disconnect, invalidate, mirror, after, client };
++module.exports = { ingress, message, event, moderate, logModeration, disconnect, invalidate, mirror, after, client, MIRROR_WRITES };
ddiff --git a/server/chat/chat-remote.js b/server/chat/chat-remote.js
ddeleted file mode 100644
iindex d5a07bfa..00000000
---- a/server/chat/chat-remote.js
++++ /dev/null
@@@ -1,395 +0,0 @@
--'use strict';
--/**
-- * The chat server, when OpenVibe.Chat runs it (CHAT_AUTHORITY=chat, see chat-authority.js).
-- *
-- * require('./chat/chat-server') returns this object instead of a ChatServer, so every module that
-- * talks to chat keeps working unchanged:
-- * - calls that push to browsers (broadcastToStream, forwardToGlobal, sendDm, TTS, channel sounds,
-- * disconnectUser, sendUserUpdate …) are queued, in order, and sent to Chat's
-- * POST /internal/live/calls in small batches (service token for audience openvibe.chat,
-- * capability chat.live_bridge.write);
-- * - the chat-table writes Live's other modules make (db.saveChatMessage from AI viewers, relays,
-- * donations; /api/mod deletes and queues; moderation log) are forwarded the same way, because
-- * Chat owns those tables. Updates and deletes also run on Live's mirror at once so their
-- * return values (deleted ids) are right; inserts run only in Chat and return a placeholder id
-- * (≤ -2^40) that Chat maps to the real one for every later call of this boot. Forwarded writes
-- * are kept in chat_bridge_outbox until Chat acknowledges them, so a restart or a Chat outage
-- * loses none;
-- * - synchronous reads (connection counts, viewer counts, slow modes, a connected user's address)
-- * come from a presence snapshot polled from Chat's GET /internal/live/presence every few seconds.
-- *
-- * Do NOT remove these writers (T3 J2) yet: Live still generates the messages Chat delivers (AI viewer
-- * replies, relays, donations, /api/mod deletes, deploy notices) and requests its alert sounds, and Chat
-- * does not yet consume or generate them on its own, so dropping this forwarding sends nothing to Chat and
-- * viewers lose them. Retire a path only once Chat owns it; drop chat_bridge_outbox only after its
-- * unacknowledged op = 'db' rows have been delivered. Covered by test/chat-context.test.js.
-- */
--const crypto = require('crypto');
--const db = require('../db/database');
--const principal = require('../net/network-principal');
--const { CHAT_URL } = require('./chat-authority');
--// LIVE_CHAT_INGRESS=1: Live's producers call Chat's typed ingress through chat-delivery.js; this proxy then reads
--// presence from /internal/chat/presence and sends its cache hints there too. Queued outbox rows still drain here.
--const delivery = require('./chat-delivery');
--
--const AUDIENCE = 'openvibe.chat';
--const BOOT = crypto.randomUUID();
--const REF_BASE = 2 ** 40;
--const BATCH = 200;
--const PRESENCE_MS = 3000;
--const EPHEMERAL_TTL_MS = 30000; // a broadcast older than this is dropped, not delivered late
--const MAX_QUEUE = 20000;
--
--// Chat-table writes Live still makes, and whether they also run on Live's mirror right away.
--// Inserts never run here (their ids are Chat's): the caller gets a placeholder.
--const FORWARDED_DB = {
-- saveChatMessage: 'insert',
-- logModerationAction: 'insert',
-- holdMessageForApproval: 'insert',
-- hideRelayUser: 'insert',
-- createChannelSound: 'insert',
-- mergeChatMessageMetadata: 'local',
-- deleteChatMessage: 'local',
-- deleteUserChatMessages: 'local',
-- deleteAnonChatMessages: 'local',
-- deleteRelayUserMessages: 'local',
-- deleteChatMessagesByTimeRange: 'local',
-- reviewPendingIpMessage: 'local',
-- approveAllFromIp: 'local',
-- denyAllFromIp: 'local',
-- recordRelayUser: 'local',
-- unhideRelayUser: 'local',
-- unhideRelayUserByIdentity: 'local',
-- recordFirstChat: 'local',
-- setTtsVoiceOverride: 'local',
-- deleteTtsVoiceOverride: 'local',
-- setChannelSoundEmote: 'local',
-- deleteChannelSound: 'local',
-- renameChannelSoundCommand: 'local',
-- updateChannelSoundEmoteRefs: 'local',
--};
--
--// Live-owned data Chat caches, written by Live's own routes (IP approvals, bans): after the write,
--// Chat is told to reload it instead of waiting for its cache to expire. Channel moderators, channel
--// moderation settings and alert sounds are Chat's own tables since roadmap T3, so nothing here.
--const OBSERVED_DB = {
-- approveIp: (a) => ['approvals', a[0]],
-- revokeIpApproval: (a) => ['approvals', a[0]],
-- forgiveBan: () => ['bans', null],
--};
--
--// db functions before forwarding wrapped them (module-wide: installed once per process).
--const originals = {};
--
--/** Apply a 'local'-mode write to Live's own mirror only, without the bridge (LIVE_CHAT_INGRESS sends it to Chat). */
--function localWrite(fn, ...args) {
-- if (FORWARDED_DB[fn] !== 'local') throw new Error(`${fn} is not a local mirror write`);
-- return (originals[fn] || db[fn])(...args);
--}
--
--function create(ChatServer) {
-- // A local ChatServer that never listens: its address and anon-number helpers are Live's own
-- // (call-server uses them for voice calls, and Chat asks Live for anon numbers).
-- const local = new ChatServer();
-- const queue = []; // { seq, op, args, ref, durableId, at }
-- let seq = 0;
-- let refCounter = 0;
-- let busy = false;
-- let inflight = 0; // ops at the head of the queue that are being sent right now
-- let batchLimit = BATCH; // 1 after Chat refused a batch as a whole (too large / malformed)
-- let backoffUntil = 0;
-- let flushTimer = null;
-- let presenceTimer = null;
-- let started = false;
-- let presence = { total: 0, streams: {}, slow_mode: {}, users: [], anons: [], at: null };
-- const slowModeByStream = new Map();
-- const stats = { sent: 0, failed: 0, dropped: 0, lastError: null };
--
-- function ensureOutbox() {
-- db.getDb().exec(`CREATE TABLE IF NOT EXISTS chat_bridge_outbox (
-- id INTEGER PRIMARY KEY AUTOINCREMENT,
-- boot TEXT NOT NULL,
-- ref INTEGER,
-- op TEXT NOT NULL,
-- args TEXT NOT NULL,
-- created_at DATETIME DEFAULT CURRENT_TIMESTAMP
-- )`);
-- }
--
-- function schedule(ms = 0) {
-- if (flushTimer) return;
-- flushTimer = setTimeout(() => { flushTimer = null; flush().catch(() => {}); }, ms);
-- if (flushTimer.unref) flushTimer.unref();
-- }
--
-- function enqueue(op, args, { ref = null, durable = false, boot = BOOT, durableId = null } = {}) {
-- let id = durableId;
-- if (durable && !id) {
-- try {
-- ensureOutbox();
-- id = db.getDb().prepare('INSERT INTO chat_bridge_outbox (boot, ref, op, args) VALUES (?, ?, ?, ?)').run(boot, ref, op, JSON.stringify(args)).lastInsertRowid;
-- } catch (err) { console.warn('[ChatRemote] outbox write failed:', err.message); }
-- }
-- if (queue.length >= MAX_QUEUE) {
-- // Drop the oldest broadcast, never a write.
-- const i = queue.findIndex((q, idx) => idx >= inflight && !q.durableId);
-- if (i >= 0) { queue.splice(i, 1); stats.dropped++; }
-- }
-- queue.push({ seq: ++seq, op, args, ref, durableId: id, boot, at: Date.now() });
-- schedule();
-- }
--
-- async function post(path, body, retried = false) {
-- const res = await fetch(`${CHAT_URL}${path}`, {
-- method: body ? 'POST' : 'GET',
-- headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...(await principal.serviceHeaders(AUDIENCE)) },
-- body: body ? JSON.stringify(body) : undefined,
-- signal: AbortSignal.timeout(10000),
-- });
-- if (res.status === 401 && !retried) { principal.invalidate(AUDIENCE); return post(path, body, true); }
-- if (!res.ok) { const err = new Error(`Chat ${res.status}`); err.status = res.status; throw err; }
-- return res.json();
-- }
--
-- async function flush() {
-- if (busy || !queue.length) return;
-- if (Date.now() < backoffUntil) { schedule(backoffUntil - Date.now()); return; }
-- busy = true;
-- try {
-- while (queue.length) {
-- // Stale broadcasts are dropped; forwarded writes always go.
-- const now = Date.now();
-- for (let i = queue.length - 1; i >= 0; i--) {
-- if (!queue[i].durableId && now - queue[i].at > EPHEMERAL_TTL_MS) { queue.splice(i, 1); stats.dropped++; }
-- }
-- if (!queue.length) break;
-- // One boot per request (replayed writes of an earlier boot keep their own id space).
-- const boot = queue[0].boot;
-- const batch = [];
-- for (const q of queue) { if (q.boot !== boot || batch.length >= batchLimit) break; batch.push(q); }
-- let out;
-- inflight = batch.length;
-- try {
-- // A forwarded write carries its outbox id as an idempotency key: Chat applies it once
-- // even when a lost response makes Live send it again.
-- out = await post('/internal/live/calls', { boot, ops: batch.map((q) => ({ seq: q.seq, op: q.op, args: q.args, ref: q.ref, key: q.durableId ? `live:${q.durableId}` : undefined })) });
-- } catch (err) {
-- inflight = 0;
-- stats.failed++;
-- if (err.status === 400 || err.status === 413) {
-- // Refused as a request, not per op: retry one by one; a single op still refused
-- // can never be delivered, so it is dropped (logged) instead of blocking the queue.
-- if (batch.length > 1) { batchLimit = 1; continue; }
-- const q = queue.shift();
-- stats.dropped++;
-- console.error(`[ChatRemote] Chat refused ${q.op}${q.op === 'db' ? ` ${q.args[0]}` : ''} (${err.status}); dropped`);
-- if (q.durableId) { try { db.getDb().prepare('DELETE FROM chat_bridge_outbox WHERE id = ?').run(q.durableId); } catch { /* */ } }
-- continue;
-- }
-- if (stats.lastError !== err.message) console.warn(`[ChatRemote] Chat unreachable (${queue.length} queued): ${err.message}`);
-- stats.lastError = err.message;
-- backoffUntil = Date.now() + 2000;
-- schedule(2000);
-- return;
-- }
-- stats.lastError = null;
-- inflight = 0;
-- batchLimit = BATCH;
-- queue.splice(0, batch.length);
-- stats.sent += batch.length;
-- const done = batch.filter((q) => q.durableId).map((q) => q.durableId);
-- if (done.length) {
-- try { db.getDb().prepare(`DELETE FROM chat_bridge_outbox WHERE id IN (${done.map(() => '?').join(',')})`).run(...done); } catch { /* */ }
-- }
-- for (const r of (out && out.results) || []) {
-- if (!r.ok) console.warn(`[ChatRemote] Chat refused op #${r.seq}: ${r.error}`);
-- }
-- }
-- } finally { busy = false; }
-- }
--
-- async function pollPresence() {
-- try {
-- const p = delivery.ingress() ? await delivery.client.presence() : await post('/internal/live/presence');
-- if (!p) return;
-- presence = p;
-- slowModeByStream.clear();
-- for (const [k, v] of Object.entries(p.slow_mode || {})) slowModeByStream.set(Number(k), Number(v) || 0);
-- } catch { /* keep the last snapshot */ }
-- }
--
-- // ── Forwarded database writes ───────────────────────────────
-- function installDbForwarding() {
-- for (const [fn, mode] of Object.entries(FORWARDED_DB)) {
-- if (typeof db[fn] !== 'function' || originals[fn]) continue;
-- originals[fn] = db[fn];
-- db[fn] = function forwardedChatWrite(...args) {
-- if (mode === 'insert') {
-- const ref = -(REF_BASE + (++refCounter));
-- enqueue('db', [fn, ...args], { ref, durable: true });
-- return { changes: 1, lastInsertRowid: ref };
-- }
-- let result;
-- try { result = originals[fn](...args); } catch (err) { result = undefined; console.warn(`[ChatRemote] mirror ${fn}: ${err.message}`); }
-- enqueue('db', [fn, ...args], { durable: true });
-- return result;
-- };
-- }
-- }
--
-- function installInvalidations() {
-- for (const [fn, target] of Object.entries(OBSERVED_DB)) {
-- if (typeof db[fn] !== 'function' || originals[fn]) continue;
-- originals[fn] = db[fn];
-- db[fn] = function observedLiveWrite(...args) {
-- const result = originals[fn](...args);
-- try {
-- const [kind, id] = target(args);
-- if (delivery.ingress()) delivery.invalidate(kind === 'bans' ? { bans: true } : { [kind]: Number(id) || undefined });
-- else enqueue('invalidate', [kind, id]);
-- } catch { /* */ }
-- return result;
-- };
-- }
-- }
--
-- function replayOutbox() {
-- try {
-- ensureOutbox();
-- const rows = db.getDb().prepare('SELECT id, boot, ref, op, args FROM chat_bridge_outbox ORDER BY id').all();
-- for (const r of rows) {
-- let args;
-- try { args = JSON.parse(r.args); } catch { continue; }
-- queue.push({ seq: ++seq, op: r.op, args, ref: r.ref, durableId: r.id, boot: r.boot, at: Date.now() });
-- }
-- if (rows.length) console.log(`[ChatRemote] replaying ${rows.length} chat write(s) queued before the restart`);
-- } catch (err) { console.warn('[ChatRemote] outbox replay:', err.message); }
-- }
--
-- // A stream row carries Live's stream keys: Chat only ever needs the owner.
-- const slimStream = (s) => (s ? { id: s.id, user_id: s.user_id } : s);
-- const send = (op) => (...args) => { enqueue(op, args); };
--
-- // Live code that iterated chatServer.clients and wrote to each socket (a global delete in
-- // /api/mod) gets one pseudo-socket whose send() reaches every chat client through Chat.
-- const everyone = { readyState: 1, bufferedAmount: 0, send: (raw) => enqueue('broadcastAllRaw', [String(raw)]), close() {} };
-- const clients = new Map([[everyone, { pseudo: true }]]);
--
-- const remote = {
-- remote: true,
-- boot: BOOT,
-- stats,
-- clients,
-- slowModeByStream,
--
-- init() {
-- if (started) return null;
-- started = true;
-- ensureOutbox();
-- replayOutbox();
-- schedule();
-- pollPresence(); // floating-ok: pollPresence catches and keeps the last snapshot
-- presenceTimer = setInterval(pollPresence, PRESENCE_MS);
-- if (presenceTimer.unref) presenceTimer.unref();
-- console.log(`[Chat] CHAT_AUTHORITY=chat — chat runs in OpenVibe.Chat (${CHAT_URL}); Live forwards to it`);
-- return null;
-- },
-- handleUpgrade(req, socket) {
-- // nginx sends /ws/chat to Chat; a socket that still lands here is told to retry.
-- try { socket.write('HTTP/1.1 503 Service Unavailable\r\nRetry-After: 2\r\nConnection: close\r\n\r\n'); } catch { /* */ }
-- try { socket.destroy(); } catch { /* */ }
-- return true;
-- },
-- close() {
-- if (presenceTimer) clearInterval(presenceTimer);
-- presenceTimer = null;
-- return flush().catch(() => {});
-- },
-- flush,
--
-- // ── Pushes to browsers (Chat delivers) ──
-- broadcastToStream: send('broadcastToStream'),
-- broadcastToChannelRoom: send('broadcastToChannelRoom'),
-- /** A donation / goal alert: Chat resolves the sound from its own settings row and plays it. */
-- playAlertSound: send('playAlertSound'),
-- broadcastGlobal: send('broadcastGlobal'),
-- broadcastAll: send('broadcastAll'),
-- forwardToGlobal: send('forwardToGlobal'),
-- forwardToGlobalByChannel: send('forwardToGlobalByChannel'),
-- forwardToStreamerRooms: send('forwardToStreamerRooms'),
-- broadcastToOwnerStreams: send('broadcastToOwnerStreams'),
-- sendDm: send('sendDm'),
-- sendUserUpdate: send('sendUserUpdate'),
-- disconnectUser: send('disconnectUser'),
-- sendToConn: send('sendToConn'),
-- synthesizeAndBroadcastTTS(...args) { enqueue('synthesizeAndBroadcastTTS', args); return Promise.resolve(); },
-- triggerChannelSound(ws, client, stream, command, args = [], relay = null) {
-- const c = client ? { streamId: client.streamId || null, user: client.user ? { id: client.user.id, username: client.user.username, display_name: client.user.display_name, role: client.user.role, avatar_url: client.user.avatar_url, profile_color: client.user.profile_color } : null, anonId: client.anonId || null, ip: client.ip || null } : {};
-- enqueue('triggerChannelSound', [null, c, slimStream(stream), command, args, relay]);
-- },
-- /** Commits this boot shipped (deploy-notice.js decides which; Chat stores and shows them). */
-- deployNotice(commits) { enqueue('deployNotice', [commits], { durable: true }); },
-- /** Live changed a user (role push, avatar): Chat drops what it cached about them. */
-- userChanged(userId) { if (userId) enqueue('userChanged', [userId]); },
--
-- // ── Synchronous reads (presence snapshot) ──
-- getTotalConnections() { return Number(presence.total) || 0; },
-- getStreamViewerCount(streamId) { return Number((presence.streams || {})[streamId]) || 0; },
-- getConnectedUserIp(userId) {
-- const hit = (presence.users || []).find((u) => u.user_id === userId);
-- return hit ? hit.ip : null;
-- },
-- findClientByAnonId(anonId, streamId) {
-- // Same match as ChatServer.findClientByAnonId: the socket's stream must equal streamId.
-- const hit = (presence.anons || []).find((a) => a.anon_id === anonId && (a.stream_id ?? null) === streamId);
-- return hit ? { anonId: hit.anon_id, ip: hit.ip, streamId: hit.stream_id || null, user: null } : null;
-- },
-- getUserList() { return { logged: [], anonCount: 0 }; },
--
-- // ── Live's own address / anon helpers ──
-- normalizeIp: (ip) => local.normalizeIp(ip),
-- getClientIp: (req) => local.getClientIp(req),
-- getAnonIdForIp: (ip) => local.getAnonIdForIp(ip),
-- getAnonIdForConnection(ip, streamId = null) {
-- const key = local.normalizeIp(ip);
-- const hit = (presence.anons || []).find((a) => a.ip === key && (streamId == null || (a.stream_id || null) === streamId));
-- return hit ? hit.anon_id : local.getAnonIdForIp(key);
-- },
-- /** The anon number for an address (Network's unified resolve, else Live's table) — for Chat. */
-- async resolveAnon(ip) {
-- const key = local.normalizeIp(ip);
-- const num = await local._resolveUnifiedAnonNum(key);
-- return { anon_number: num, first_seen: db.getAnonFirstSeen(key) };
-- },
-- };
--
-- installDbForwarding();
-- installInvalidations();
-- remote._restoreDb = () => { for (const [fn, orig] of Object.entries(originals)) db[fn] = orig; };
-- return remote;
--}
--
--/**
-- * Rollback: Live runs chat itself again (CHAT_AUTHORITY unset) but chat writes it forwarded were
-- * never acknowledged by Chat (Chat down). Apply them to Live's own tables so nothing is lost.
-- * A write Chat did apply whose answer was lost is applied twice — rare, and only for that window.
-- */
--function drainToLocal(log = console) {
-- try {
-- const d = db.getDb();
-- if (!d.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'chat_bridge_outbox'").get()) return 0;
-- const rows = d.prepare("SELECT id, args FROM chat_bridge_outbox WHERE op = 'db' ORDER BY id").all();
-- let applied = 0;
-- for (const r of rows) {
-- try {
-- const [fn, ...args] = JSON.parse(r.args);
-- if (FORWARDED_DB[fn] && typeof db[fn] === 'function') { db[fn](...args); applied++; }
-- } catch (err) { log.warn(`[ChatRemote] could not apply queued chat write #${r.id}: ${err.message}`); }
-- }
-- d.prepare('DELETE FROM chat_bridge_outbox').run();
-- if (rows.length) log.log(`[ChatRemote] applied ${applied} of ${rows.length} chat write(s) OpenVibe.Chat never acknowledged`);
-- return applied;
-- } catch (err) { log.warn('[ChatRemote] drain:', err.message); return 0; }
--}
--
--module.exports = { create, drainToLocal, localWrite, FORWARDED_DB, OBSERVED_DB, REF_BASE, AUDIENCE };
ddiff --git a/server/chat/chat-server.js b/server/chat/chat-server.js
iindex 512ec1cb..478df179 100644
---- a/server/chat/chat-server.js
++++ b/server/chat/chat-server.js
@@@ -2437,7 +2437,7 @@ class ChatServer {
* Donation / goal alert sounds are OpenVibe.Chat's now (roadmap T3): the mapping lives on
* channel_moderation_settings, which Chat owns, and Chat resolves it from its own row and
* broadcasts the clip. When Live runs the chat server itself (rollback / dev) there is no Chat to
-- * ask, so this is a no-op; server/monetization/alerts.js calls the chat-remote proxy's op instead.
++ * ask, so this is a no-op; with CHAT_AUTHORITY=chat server/monetization/alerts.js sends Chat an `alert` event.
*/
playAlertSound() { /* Chat plays alerts; see server/monetization/alerts.js */ }
@@@ -2509,6 +2509,140 @@ class ChatServer {
}
}
--// CHAT_AUTHORITY=chat: OpenVibe.Chat runs the chat server; Live's modules get a proxy with the same
--// methods that hands every call to it (chat-remote.js). Default: Live runs chat itself.
--module.exports = require('./chat-authority').isRemote() ? require('./chat-remote').create(ChatServer) : new ChatServer();
++// Live-owned data Chat caches, written by Live's own routes (IP approvals, bans): after the write Chat is told
++// to reload it instead of waiting for its cache to expire.
++const OBSERVED_DB = {
++ approveIp: (a) => ({ approvals: Number(a[0]) || undefined }),
++ revokeIpApproval: (a) => ({ approvals: Number(a[0]) || undefined }),
++ forgiveBan: () => ({ bans: true }),
++};
++const PRESENCE_MS = 3000;
++
++/**
++ * CHAT_AUTHORITY=chat: OpenVibe.Chat runs the chat server. Live's producers reach it through
++ * chat-delivery.js (Chat's typed ingress); this object keeps the rest of the ChatServer surface Live's
++ * modules use, without listening or serving:
++ * - synchronous reads (connection and viewer counts, slow modes, a connected user's address) come from
++ * Chat's presence snapshot (GET /internal/chat/presence), polled every few seconds;
++ * - Live's own address / anon-number helpers (call-server, Chat's /internal/chat-effects/anon);
++ * - Live's writes to data Chat caches (IP approvals, bans) send Chat a cache hint;
++ * - a push a module still makes directly goes to the same ingress (stream, channel, global, all, DM,
++ * disconnect, user change); one with no ingress target is dropped and logged, never synthesised or sent
++ * to sockets Live does not have (TTS is spoken by Chat from the message's `tts` field).
++ * A /ws/chat upgrade that still lands here is told to retry.
++ */
++class RemoteChatServer extends ChatServer {
++ constructor() {
++ super();
++ this.remote = true;
++ this._presence = { total: 0, streams: {}, slow_mode: {}, users: [], anons: [], at: null };
++ this._presenceTimer = null;
++ this._dropped = new Set();
++ this._observeDb();
++ }
++
++ _delivery() { return require('./chat-delivery'); }
++
++ _observeDb() {
++ for (const [fn, hint] of Object.entries(OBSERVED_DB)) {
++ const orig = db[fn];
++ if (typeof orig !== 'function' || orig._chatObserved) continue;
++ const observed = (...args) => {
++ const result = orig(...args);
++ try { this._delivery().invalidate(hint(args)); } catch { /* non-critical */ }
++ return result;
++ };
++ observed._chatObserved = true;
++ db[fn] = observed;
++ }
++ }
++
++ async _pollPresence() {
++ try {
++ const p = await this._delivery().client.presence();
++ if (!p) return;
++ this._presence = p;
++ this.slowModeByStream.clear();
++ for (const [k, v] of Object.entries(p.slow_mode || {})) this.slowModeByStream.set(Number(k), Number(v) || 0);
++ } catch { /* keep the last snapshot */ }
++ }
++
++ init() {
++ if (this._presenceTimer) return null;
++ this._pollPresence(); // floating-ok: _pollPresence catches and keeps the last snapshot
++ this._presenceTimer = setInterval(() => this._pollPresence(), PRESENCE_MS);
++ if (this._presenceTimer.unref) this._presenceTimer.unref();
++ console.log('[Chat] CHAT_AUTHORITY=chat — chat runs in OpenVibe.Chat; Live delivers through its ingress');
++ return null;
++ }
++
++ handleUpgrade(req, socket) {
++ // nginx sends /ws/chat to Chat; a socket that still lands here is told to retry.
++ try { socket.write('HTTP/1.1 503 Service Unavailable\r\nRetry-After: 2\r\nConnection: close\r\n\r\n'); } catch { /* */ }
++ try { socket.destroy(); } catch { /* */ }
++ return true;
++ }
++
++ close() {
++ if (this._presenceTimer) clearInterval(this._presenceTimer);
++ this._presenceTimer = null;
++ }
++
++ // ── Pushes: Chat's ingress, or dropped and logged ──
++ _drop(name) {
++ if (this._dropped.has(name)) return;
++ this._dropped.add(name);
++ console.warn(`[Chat] ${name} has no Chat ingress target and was dropped; deliver through chat-delivery.js`);
++ }
++
++ // Guarded so a flipped CHAT_AUTHORITY can never bounce a push between this object and the seam.
++ _push(name, fn) { const d = this._delivery(); return d.ingress() ? fn(d) : this._drop(name); }
++ broadcastToStream(streamId, data) { return this._push('broadcastToStream', (d) => d.event({ kind: 'stream', id: streamId }, data)); }
++ broadcastToChannelRoom(channelUserId, streamId, data) { return this._push('broadcastToChannelRoom', (d) => d.event({ kind: 'channel', id: channelUserId }, data)); }
++ broadcastGlobal(data) { return this._push('broadcastGlobal', (d) => d.event({ kind: 'global' }, data)); }
++ broadcastAll(data) { return this._push('broadcastAll', (d) => d.event({ kind: 'all' }, data)); }
++ sendDm(userId, data) { return this._push('sendDm', (d) => d.event({ kind: 'user', id: userId }, data)); }
++ disconnectUser(opts) { return this._push('disconnectUser', (d) => d.disconnect(opts)); }
++ sendUserUpdate(userId, u) {
++ const id = Number(userId) || undefined;
++ const userData = u ? { id, username: u.username, display_name: u.display_name || null, role: u.role || null, avatar_url: u.avatar_url || null, profile_color: u.profile_color || null } : undefined;
++ return this._push('sendUserUpdate', (d) => d.invalidate({ user: id, user_data: userData }));
++ }
++ forwardToGlobal() { this._drop('forwardToGlobal'); }
++ forwardToGlobalByChannel() { this._drop('forwardToGlobalByChannel'); }
++ forwardToStreamerRooms() { this._drop('forwardToStreamerRooms'); }
++ broadcastToOwnerStreams() { this._drop('broadcastToOwnerStreams'); }
++ triggerChannelSound() { this._drop('triggerChannelSound'); }
++ sendToConn() { this._drop('sendToConn'); }
++ synthesizeAndBroadcastTTS() { this._drop('synthesizeAndBroadcastTTS'); return Promise.resolve(); }
++ broadcastUserCount() { /* Chat's */ }
++ broadcastUsersList() { /* Chat's */ }
++
++ // ── Synchronous reads (presence snapshot) ──
++ getTotalConnections() { return Number(this._presence.total) || 0; }
++ getStreamViewerCount(streamId) { return Number((this._presence.streams || {})[streamId]) || 0; }
++ getConnectedUserIp(userId) {
++ const hit = (this._presence.users || []).find((u) => u.user_id === userId);
++ return hit ? hit.ip : null;
++ }
++ findClientByAnonId(anonId, streamId) {
++ // Same match as ChatServer.findClientByAnonId: the socket's stream must equal streamId.
++ const hit = (this._presence.anons || []).find((a) => a.anon_id === anonId && (a.stream_id ?? null) === streamId);
++ return hit ? { anonId: hit.anon_id, ip: hit.ip, streamId: hit.stream_id || null, user: null } : null;
++ }
++ getUserList() { return { logged: [], anonCount: 0 }; }
++ getAnonIdForConnection(ip, streamId = null) {
++ const key = this.normalizeIp(ip);
++ const hit = (this._presence.anons || []).find((a) => a.ip === key && (streamId == null || (a.stream_id || null) === streamId));
++ return hit ? hit.anon_id : this.getAnonIdForIp(key);
++ }
++
++ /** The anon number for an address (Network's unified resolve, else Live's table) — for Chat. */
++ async resolveAnon(ip) {
++ const key = this.normalizeIp(ip);
++ const num = await this._resolveUnifiedAnonNum(key);
++ return { anon_number: num, first_seen: db.getAnonFirstSeen(key) };
++ }
++}
++
++module.exports = require('./chat-authority').isRemote() ? new RemoteChatServer() : new ChatServer();
ddiff --git a/server/chat/deploy-notice.js b/server/chat/deploy-notice.js
iindex c6a898c5..386bff3c 100644
---- a/server/chat/deploy-notice.js
++++ b/server/chat/deploy-notice.js
@@@ -19,8 +19,8 @@
// (server/events/release-events.js), queued in the SAME transaction that records the commits
// as announced (and, without the Chat service, stores the chat row). OpenVibe.Chat consumes
// that event and folds the card by head commit (compatibility register C-84, proven on the
--// 2026-09-23 22:14 UTC deploy), so with Live's outbox on the bridge hop is not used; it stays
--// only as the fallback while Events publishing is off.
++// 2026-09-23 22:14 UTC deploy). With Events publishing off there is no other way to Chat (its
++// ingress has no deploy endpoint), so the commits stay unannounced and the next boot tries again.
// ═══════════════════════════════════════════════════════════════
const { execFile } = require('child_process');
const path = require('path');
@@@ -110,14 +110,10 @@ async function announce({ db, chatServer, log = console }) {
// CHAT_AUTHORITY=chat: Live still decides what shipped; OpenVibe.Chat stores the rolling
// message and shows it (its own copy of this module). It learns the commits from the
-- // live.release.deployed event; only with Events publishing off are they handed over the bridge.
-- if (chatServer && chatServer.remote) {
-- if (!outbox) {
-- // Chat's ingress has no deploy endpoint: without Events the commits stay unannounced and the
-- // next boot tries again (LIVE_CHAT_INGRESS, docs: OpenVibe.Chat docs/chat-ingress.md).
-- if (require('./chat-delivery').ingress()) { log.warn('[Deploy notice] Events outbox unavailable; left unannounced for the next boot'); return { announced: 0 }; }
-- chatServer.deployNotice(commits);
-- }
++ // live.release.deployed event only. Chat's ingress has no deploy endpoint: with Events publishing off
++ // the commits are not recorded as announced and the next boot tries again (OpenVibe.Chat docs/chat-ingress.md).
++ if (require('./chat-delivery').ingress()) {
++ if (!outbox) { log.warn('[Deploy notice] Events outbox unavailable; left unannounced for the next boot'); return { announced: 0 }; }
try { inTransaction(recordDeploy); } catch (err) { log.warn('[Deploy notice] not recorded:', err.message); return { announced: commits.length, event_id: null }; }
if (outbox) outbox.kick();
return { announced: commits.length, event_id: eventId };
ddiff --git a/server/chat/live-context-routes.js b/server/chat/live-context-routes.js
iindex 5d765d6b..c48bf70e 100644
---- a/server/chat/live-context-routes.js
++++ b/server/chat/live-context-routes.js
@@@ -453,22 +453,22 @@ effectsRouter.post('/ai/mod-command', (req, res) => {
} catch (err) { fail(res, 400, err.message); }
});
--// !hype / !beef / !arena: arena-chat answers the sender through Chat (sendToConn).
--effectsRouter.post('/arena-command', (req, res) => {
++// !hype / !beef / !arena: room lines go to Chat's typed events; the sender-only replies come back in this
++// response as `replies` (Chat has no per-connection ingress; Chat sends them to the sender's socket). The
++// answer waits for the command's own lookups, so a reply made after one (!arena's fighter card) is in it.
++effectsRouter.post('/arena-command', async (req, res) => {
const client = req.body?.client || {};
const delivery = require('./chat-delivery');
-- // LIVE_CHAT_INGRESS: room lines go to Chat's typed events; the sender-only replies come back in this
-- // response as `replies` (Chat has no per-connection ingress; it reads them from here at J4b).
const replies = [];
-- const shim = delivery.ingress() ? {
++ let work = null;
++ const shim = {
sendTo: (_ws, payload) => { replies.push(payload); },
broadcastToStream: (streamId, payload) => delivery.event({ kind: 'stream', id: streamId }, payload),
-- } : {
-- sendTo: (_ws, payload) => chatServer().sendToConn(client.conn_id, payload),
-- broadcastToStream: (streamId, payload) => chatServer().broadcastToStream(streamId, payload),
++ track: (p) => { work = p; },
};
let handled = false;
try { handled = require('../arena/arena-chat').handle(shim, null, client, String(req.body.cmd || ''), Array.isArray(req.body.parts) ? req.body.parts : []); } catch (e) { console.warn('[Arena] chat command:', e.message); }
++ if (work) { try { await work; } catch { /* the command answers its own errors */ } }
res.json(replies.length ? { handled: !!handled, replies } : { handled: !!handled });
});
ddiff --git a/server/db/migrations.js b/server/db/migrations.js
iindex 641b61d3..cb00c738 100644
---- a/server/db/migrations.js
++++ b/server/db/migrations.js
@@@ -171,6 +171,22 @@ const OPERATOR_MIGRATIONS = [
WHERE substr(password_hash, 1, 5) <> '$sso$' AND id IN (${networkLinkedUserIds(db)})`).run();
},
},
++ {
++ id: 'op_002_drop_chat_bridge_outbox',
++ // T3 J2 (scripts/chat-bridge-outbox-drop.js). Live delivers to OpenVibe.Chat through its typed ingress
++ // only; nothing under server/ reads or writes the old bridge's outbox any more (test/chat-bridge-removed.
++ // test.js). The release before still writes it, so this contract step runs once that release is out of
++ // rollback range (ADR-028), never at boot. Chat writes that release forwarded and Chat never acknowledged
++ // sit here as op = 'db' rows: the script delivers them to Chat first, and the drop refuses while any
++ // remain, so none is lost.
++ up: (db) => {
++ if (tableExists(db, 'chat_bridge_outbox')) {
++ const pending = db.prepare("SELECT COUNT(*) AS n FROM chat_bridge_outbox WHERE op = 'db'").get().n;
++ if (pending > 0) throw new Error(`chat_bridge_outbox holds ${pending} chat write(s) OpenVibe.Chat has not acknowledged; deliver them first (scripts/chat-bridge-outbox-drop.js --deliver)`);
++ }
++ db.exec('DROP TABLE IF EXISTS chat_bridge_outbox');
++ },
++ },
];
const failures = new Map(); // id -> message, for this process
ddiff --git a/server/index.js b/server/index.js
iindex 97b5c6f9..bfa05190 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -1124,8 +1124,6 @@ async function start() {
// Initialize DM tables
const dm = require('./chat/dm');
dm.ensureTables();
-- // Back from CHAT_AUTHORITY=chat (rollback): chat writes OpenVibe.Chat never acknowledged land here.
-- if (!require('./chat/chat-authority').isRemote()) require('./chat/chat-remote').drainToLocal();
// Migrate: add last_heartbeat column if missing
try { db.run("ALTER TABLE streams ADD COLUMN last_heartbeat DATETIME"); console.log('[DB] Added last_heartbeat column'); } catch { /* already exists */ }
// Migrate: add theme_id to users table if missing
ddiff --git a/server/integrations/chat-relay-service.js b/server/integrations/chat-relay-service.js
iindex fe18caa7..d56194a2 100644
---- a/server/integrations/chat-relay-service.js
++++ b/server/integrations/chat-relay-service.js
@@@ -494,7 +494,7 @@ class ChatRelayService {
} catch { /* non-critical */ }
}
-- /** LIVE_CHAT_INGRESS: Chat persists, broadcasts, mirrors, reads aloud and records the first chat in one call. */
++ /** Chat ingress: Chat persists, broadcasts, mirrors, reads aloud and records the first chat in one call. */
_deliverToChat(delivery, bridge, username, prefixedUsername, color, extras, message) {
let first = false;
try {
ddiff --git a/server/integrations/powerchat-webhook.js b/server/integrations/powerchat-webhook.js
iindex 2a3566c1..7784a916 100644
---- a/server/integrations/powerchat-webhook.js
++++ b/server/integrations/powerchat-webhook.js
@@@ -91,7 +91,7 @@ function _testFulfillmentAllowed() {
}
// ── Donation handling — mirrors POST /api/funds/donate ───────────────────────
--// A persisted chat line plus its live card: one Chat ingress call under LIVE_CHAT_INGRESS (Chat persists and
++// A persisted chat line plus its live card: one Chat ingress call with CHAT_AUTHORITY=chat (Chat persists and
// shows it; `mirror` also puts it in global chat), else the legacy card broadcast(s) and a local save.
function _chatLine(chatServer, { frame, line, mirror = false, key }) {
if (delivery.ingress()) { delivery.message({ ...line, mirror, key }); return; }
ddiff --git a/server/internal/routes.js b/server/internal/routes.js
iindex 73487242..30e3543e 100644
---- a/server/internal/routes.js
++++ b/server/internal/routes.js
@@@ -14,8 +14,7 @@ const { guard } = require('../net/service-guard');
function notifyChat(userId) {
try {
const delivery = require('../chat/chat-delivery');
-- if (delivery.ingress()) { delivery.invalidate({ user: Number(userId) }); return; }
-- const cs = require('../chat/chat-server'); if (cs.remote) cs.userChanged(userId);
++ if (delivery.ingress()) delivery.invalidate({ user: Number(userId) });
} catch { /* non-critical */ }
}
ddiff --git a/server/monetization/alerts.js b/server/monetization/alerts.js
iindex 0c8273de..80c0f6b1 100644
---- a/server/monetization/alerts.js
++++ b/server/monetization/alerts.js
@@@ -2,12 +2,11 @@
* alerts.js — streamer alert sounds for donations / donation-goal-reached events.
*
* The sound mapping lives on channel_moderation_settings, which OpenVibe.Chat owns and reads locally
-- * (roadmap T3), so Live no longer reads the file or broadcasts base64 itself. It asks Chat to play the
-- * alert through the chat bridge — op `playAlertSound [streamerId, streamId, kind]` — and Chat resolves
-- * the sound from its own settings row, reads the shared sounds file and broadcasts the
-- * `soundboard-audio` message to the channel room. In production CHAT_AUTHORITY=chat, so `chatServer`
-- * is the chat-remote proxy (server/chat/chat-remote.js) and the call is queued to Chat like every
-- * other push. A goal-reached event is `kind: 'goal'` (Chat falls back to the donation sound).
++ * (roadmap T3), so Live no longer reads the file or broadcasts base64 itself. With CHAT_AUTHORITY=chat it
++ * asks Chat to play the alert with an `alert` event on Chat's typed ingress (server/chat/chat-delivery.js),
++ * and Chat resolves the sound from its own settings row, reads the shared sounds file and broadcasts the
++ * `soundboard-audio` message to the channel room. A goal-reached event is `kind: 'goal'` (Chat falls back
++ * to the donation sound).
*/
'use strict';
@@@ -20,7 +19,7 @@ function playAlertSound(chatServer, streamerId, streamId, kind) {
try {
const delivery = require('../chat/chat-delivery');
if (delivery.ingress()) {
-- // LIVE_CHAT_INGRESS: the same request as an `alert` event on Chat's typed ingress.
++ // An `alert` event on Chat's typed ingress.
delivery.event({ kind: 'channel', id: streamerId }, { type: 'alert', streamerId: Number(streamerId), streamId: streamId ? Number(streamId) : null, kind: kind === 'goal' ? 'goal' : 'donation' });
return;
}
ddiff --git a/server/streaming/calls-authority.js b/server/streaming/calls-authority.js
iindex 029e00f4..9767a8ad 100644
---- a/server/streaming/calls-authority.js
++++ b/server/streaming/calls-authority.js
@@@ -11,8 +11,8 @@
* of the local call server:
* POST /internal/calls/stream-channel { stream_id, mode, user_id }
* DELETE /internal/calls/stream-channel/:streamId
-- * with Live's service token for audience openvibe.chat — the same principal and grant
-- * (chat.live_bridge.write) Live's chat bridge uses (../chat/chat-remote.js). Fire-and-forget for the
++ * with Live's service token for audience openvibe.chat (grant chat.live_bridge.write, which these
++ * calls still use now that Live's chat bridge is gone). Fire-and-forget for the
* caller: a network error or a 5xx is retried after 1 s, 5 s and 15 s, unless a later call for the
* same stream superseded it (a stream that ended while its go-live was being retried is not
* given a channel afterwards); a 4xx is logged and dropped. Live's /ws/call and voice-channel
ddiff --git a/test/chat-bridge-removed.test.js b/test/chat-bridge-removed.test.js
nnew file mode 100644
iindex 00000000..2fa98e33
---- /dev/null
++++ b/test/chat-bridge-removed.test.js
@@@ -0,0 +1,106 @@
++'use strict';
++
++// T3 J2: Live delivers to OpenVibe.Chat through its typed ingress only (server/chat/chat-delivery.js).
++// The old ordered-calls bridge (chat-remote.js, POST /internal/live/calls) is gone, nothing under server/
++// reads or writes its outbox (only the operator migration that drops it names the table), and the drop
++// (scripts/chat-bridge-outbox-drop.js → op_002_drop_chat_bridge_outbox) delivers queued chat writes to
++// Chat first and refuses while any remain, so a write the old release queued is never lost.
++
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const Database = require('better-sqlite3');
++
++const ROOT = path.join(__dirname, '..');
++
++// 1. Static guard: no reader or writer of the outbox, no bridge, no flag.
++const offenders = [];
++(function walk(dir) {
++ for (const e of fs.readdirSync(dir, { withFileTypes: true })) {
++ const p = path.join(dir, e.name);
++ if (e.isDirectory()) { walk(p); continue; }
++ if (!p.endsWith('.js')) continue;
++ const rel = path.relative(ROOT, p);
++ const src = fs.readFileSync(p, 'utf8');
++ if (/chat_bridge_outbox/i.test(src) && rel !== path.join('server', 'db', 'migrations.js')) offenders.push(`${rel}: chat_bridge_outbox`);
++ if (/chat-remote/.test(src)) offenders.push(`${rel}: chat-remote`);
++ if (/\/internal\/live\/(calls|presence)/.test(src)) offenders.push(`${rel}: the old bridge endpoints`);
++ if (/LIVE_CHAT_INGRESS/.test(src)) offenders.push(`${rel}: LIVE_CHAT_INGRESS (retired: CHAT_AUTHORITY=chat is the ingress)`);
++ }
++})(path.join(ROOT, 'server'));
++assert.deepStrictEqual(offenders, [], `server/ still uses the old chat bridge:\n${offenders.join('\n')}`);
++assert.ok(!fs.existsSync(path.join(ROOT, 'server', 'chat', 'chat-remote.js')), 'chat-remote.js is deleted');
++
++// 2. The drop is an operator step, never run at boot.
++const migrations = require('../server/db/migrations');
++const drop = require('../scripts/chat-bridge-outbox-drop');
++assert.ok(!migrations.MIGRATIONS.some((m) => m.id === drop.ID), 'not a boot migration');
++assert.ok(migrations.OPERATOR_MIGRATIONS.some((m) => m.id === drop.ID), 'an operator migration');
++
++(async () => {
++ const db = new Database(':memory:');
++ db.exec(`CREATE TABLE chat_bridge_outbox (
++ id INTEGER PRIMARY KEY AUTOINCREMENT, boot TEXT NOT NULL, ref INTEGER, op TEXT NOT NULL, args TEXT NOT NULL,
++ created_at DATETIME DEFAULT CURRENT_TIMESTAMP)`);
++ const put = db.prepare('INSERT INTO chat_bridge_outbox (boot, ref, op, args) VALUES (?, ?, ?, ?)');
++ put.run('boot-a', -(2 ** 40) - 1, 'db', JSON.stringify(['saveChatMessage', { stream_id: 1, message: 'queued' }]));
++ put.run('boot-a', null, 'db', JSON.stringify(['deleteChatMessage', 5, 2]));
++ put.run('boot-b', null, 'deployNotice', JSON.stringify([[{ hash: 'abc' }]]));
++ put.run('boot-b', null, 'db', JSON.stringify(['recordFirstChat', 'user:9', 1]));
++ assert.deepStrictEqual(drop.counts(db), { exists: true, total: 4, writes: 3, byOp: { db: 3, deployNotice: 1 }, boots: 2 });
++
++ // 3. Unacknowledged chat writes: the drop refuses and keeps the table.
++ let res = migrations.runOperator(db, drop.ID);
++ assert.strictEqual(res.outcome, 'failed');
++ assert.match(res.error, /3 chat write/);
++ assert.strictEqual(drop.counts(db).total, 4, 'rolled back, nothing lost');
++
++ // 4. Chat down: delivery stops, nothing is deleted.
++ const sent = [];
++ const down = async () => { const err = new Error('Chat 503'); err.status = 503; throw err; };
++ let r = await drop.deliver(db, down);
++ assert.strictEqual(r.delivered, 0);
++ assert.strictEqual(r.error, 'Chat 503');
++ assert.strictEqual(drop.counts(db).total, 4);
++
++ // 5. Chat up: rows go in order, one boot per request, each with its idempotency key; Chat refuses one.
++ const up = async (p, body) => {
++ assert.strictEqual(p, '/internal/live/calls');
++ sent.push(body);
++ return { ok: true, results: body.ops.map((o) => ({ seq: o.seq, ok: o.op !== 'deployNotice', error: o.op === 'deployNotice' ? 'nope' : undefined })) };
++ };
++ r = await drop.deliver(db, up);
++ assert.deepStrictEqual(sent.map((b) => [b.boot, b.ops.map((o) => o.op)]), [['boot-a', ['db', 'db']], ['boot-b', ['deployNotice', 'db']]]);
++ assert.deepStrictEqual(sent.flatMap((b) => b.ops.map((o) => o.key)), ['live:1', 'live:2', 'live:3', 'live:4']);
++ assert.strictEqual(sent[0].ops[0].ref, -(2 ** 40) - 1, 'the placeholder ref Chat maps');
++ assert.strictEqual(r.delivered, 3);
++ assert.deepStrictEqual(r.refused, [3], 'a refused row is left for the operator');
++ assert.deepStrictEqual(drop.counts(db), { exists: true, total: 1, writes: 0, byOp: { deployNotice: 1 }, boots: 1 });
++
++ // 5b. A whole batch refused (400) is retried row by row; an unparseable row is reported once.
++ const db2 = new Database(':memory:');
++ db2.exec(`CREATE TABLE chat_bridge_outbox (
++ id INTEGER PRIMARY KEY AUTOINCREMENT, boot TEXT NOT NULL, ref INTEGER, op TEXT NOT NULL, args TEXT NOT NULL,
++ created_at DATETIME DEFAULT CURRENT_TIMESTAMP)`);
++ const put2 = db2.prepare('INSERT INTO chat_bridge_outbox (boot, ref, op, args) VALUES (?, ?, ?, ?)');
++ put2.run('boot-c', null, 'db', '{not json');
++ put2.run('boot-c', null, 'db', JSON.stringify(['recordFirstChat', 'user:1', 1]));
++ put2.run('boot-c', null, 'db', JSON.stringify(['recordFirstChat', 'user:2', 1]));
++ const picky = async (p, body) => {
++ if (body.ops.length > 1) { const err = new Error('too big'); err.status = 413; throw err; }
++ return { ok: true, results: body.ops.map((o) => ({ seq: o.seq, ok: true })) };
++ };
++ r = await drop.deliver(db2, picky);
++ assert.strictEqual(r.delivered, 2);
++ assert.deepStrictEqual(r.refused, [1], 'each refused id appears once');
++ db2.close();
++
++ // 6. No chat write left: the drop applies once.
++ res = migrations.runOperator(db, drop.ID);
++ assert.strictEqual(res.outcome, 'applied');
++ assert.strictEqual(drop.counts(db).exists, false);
++ assert.strictEqual(migrations.runOperator(db, drop.ID).outcome, 'already');
++ db.close();
++
++ console.log('chat bridge removed: no outbox reader or writer under server/; the operator drop delivers queued writes first and refuses while any remain — all checks passed');
++})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/test/chat-context.test.js b/test/chat-context.test.js
iindex 92e46a64..a124d3c4 100644
---- a/test/chat-context.test.js
++++ b/test/chat-context.test.js
@@@ -2,8 +2,8 @@
// Chat moved to OpenVibe.Chat (roadmap Wave 6). What Live answers it on /internal/chat-context/*
// and /internal/chat-effects/* (service tokens, capabilities, re-checked moderators, the read
--// mirror), and the chat-server proxy Live's own modules get with CHAT_AUTHORITY=chat (ordered
--// bridge calls, placeholder ids for forwarded inserts, presence reads). Against stub Network and
++// mirror), and the chat server Live's own modules get with CHAT_AUTHORITY=chat (presence reads,
++// cache hints and pushes over Chat's typed ingress; no bridge, no outbox). Against stub Network and
// Chat servers.
const assert = require('assert');
@@@ -40,8 +40,8 @@ const READ = serviceToken(['live.chat_context.read']);
const WRITE = serviceToken(['live.chat_effects.write']);
const MIRROR = serviceToken(['live.chat_mirror.write']);
--// Stub Network (Live's own service token for audience openvibe.chat) and stub Chat (the bridge).
--const bridgeCalls = [];
++// Stub Network (Live's own service token for audience openvibe.chat) and stub Chat (its typed ingress).
++const ingressCalls = [];
const network = http.createServer((req, res) => {
let raw = '';
req.on('data', (c) => { raw += c; });
@@@ -49,7 +49,7 @@ const network = http.createServer((req, res) => {
res.setHeader('Content-Type', 'application/json');
if (req.url === '/oauth/token') {
const f = new URLSearchParams(raw);
-- return res.end(JSON.stringify({ access_token: serviceToken(['chat.live_bridge.write', 'chat.presence.read'], { aud: f.get('audience'), sub: 'svc:live' }), expires_in: 300 }));
++ return res.end(JSON.stringify({ access_token: serviceToken(['chat.message.send', 'chat.event.publish', 'chat.moderation.write', 'chat.cache.invalidate', 'chat.presence.read'], { aud: f.get('audience'), sub: 'svc:live' }), expires_in: 300 }));
}
res.statusCode = 404; res.end('{}');
});
@@@ -60,12 +60,13 @@ const chat = http.createServer((req, res) => {
req.on('end', () => {
res.setHeader('Content-Type', 'application/json');
assert.ok(String(req.headers.authorization || '').startsWith('Bearer '), 'Live calls Chat with a service token');
-- if (req.url === '/internal/live/calls') {
-- const body = JSON.parse(raw);
-- bridgeCalls.push(body);
-- return res.end(JSON.stringify({ ok: true, results: body.ops.map((o) => ({ seq: o.seq, ok: true })) }));
++ assert.ok(!String(req.url).startsWith('/internal/live/'), 'nothing goes over the old bridge');
++ const im = String(req.url).match(/^\/internal\/chat\/(messages|events|moderation|invalidate)$/);
++ if (im) {
++ ingressCalls.push({ family: im[1], body: JSON.parse(raw || '{}') });
++ return res.end(JSON.stringify({ ok: true }));
}
-- if (req.url === '/internal/live/presence') {
++ if (req.url === '/internal/chat/presence') {
return res.end(JSON.stringify({ total: 7, streams: { 1: 3 }, slow_mode: { 1: 5000 }, users: [{ user_id: 3, ip: '198.51.100.3', stream_id: 1 }], anons: [{ anon_id: 'anon9', ip: '203.0.113.9', stream_id: 1 }] }));
}
// Chat's internal read API (roadmap T3): Live reads the six staged tables through it now.
@@@ -269,7 +270,7 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
db.upsertSubscription({ subscriber_id: viewer, streamer_id: streamer, status: 'canceled', current_period_end: new Date(Date.now() + 86400e3).toISOString() });
assert.strictEqual((await subOf(`user_id=${viewer}&streamer_id=${streamer}`)).body.subscriber, false, 'not active');
// Alert sounds moved to Chat with channel_moderation_settings (roadmap T3): Chat resolves the
-- // sound from its own row and plays it through the bridge op, so Live's effect is gone.
++ // sound from its own row and plays it on an `alert` event (test/chat-ingress.test.js), so Live's effect is gone.
// 8b. A ring from Chat's call server (CALLS_AUTHORITY=chat): Live pushes the VC_CALL_INVITE it used to.
const notify = require('../server/utils/notify');
@@@ -303,69 +304,39 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
assert.strictEqual(db.getChatMessageById(900001), undefined);
assert.ok(db.getUserById(viewer), 'users are never touched');
-- // 10. The proxy Live's modules get: ordered bridge calls, placeholders, presence.
++ // 10. The chat server Live's modules get (T3 J2: the bridge and its outbox are gone).
const chatServer = require('../server/chat/chat-server');
assert.strictEqual(chatServer.remote, true);
chatServer.init();
-- const before = d.prepare('SELECT COUNT(*) AS n FROM chat_messages').get().n;
-- const saved = db.saveChatMessage({ stream_id: streamId, user_id: null, username: 'Bot', message: 'beep', message_type: 'chat', source_platform: 'ai' });
-- assert.ok(saved.lastInsertRowid <= -(2 ** 40), 'a placeholder id');
-- assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM chat_messages').get().n, before, 'inserts happen in Chat, not here');
-- assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM chat_bridge_outbox').get().n, 1, 'kept until Chat acknowledges');
-- chatServer.broadcastToStream(streamId, { type: 'chat', id: saved.lastInsertRowid, message: 'beep' });
-- chatServer.synthesizeAndBroadcastTTS(streamId, 'Bot', 'beep', null, null, 'aibot:bot', null, `m${saved.lastInsertRowid}`);
-- await chatServer.flush();
-- await sleep(100);
-- const ops = bridgeCalls.flatMap((c) => c.ops);
-- assert.deepStrictEqual(ops.map((o) => o.op), ['db', 'broadcastToStream', 'synthesizeAndBroadcastTTS']);
-- assert.strictEqual(ops[0].args[0], 'saveChatMessage');
-- assert.strictEqual(ops[0].ref, saved.lastInsertRowid);
-- assert.ok(/^live:\d+$/.test(ops[0].key), 'forwarded writes carry an idempotency key');
-- assert.strictEqual(ops[1].args[1].id, saved.lastInsertRowid);
-- assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM chat_bridge_outbox').get().n, 0, 'acknowledged writes leave the outbox');
-- // Deletes run on the mirror at once (their ids are returned) and are forwarded too.
-- d.prepare("INSERT INTO chat_messages (id, user_id, username, message) VALUES (900002, ?, 'MODDY', 'x')").run(mod);
-- assert.deepStrictEqual(db.deleteUserChatMessages(mod, {}), [900002]);
-- await chatServer.flush();
-- assert.deepStrictEqual(bridgeCalls.at(-1).ops[0].args.slice(0, 2), ['deleteUserChatMessages', mod]);
-- // Live's own writes to data Chat caches (IP approvals, bans) tell Chat to reload it; the
-- // channel-moderation writes went to Chat with the six staged tables (roadmap T3).
-- // The /api/mod global delete loops over chatServer.clients: one pseudo-socket reaches everyone.
-- for (const [ws] of chatServer.clients) { if (ws.readyState === 1) ws.send(JSON.stringify({ type: 'delete-messages', ids: [1] })); }
-- await chatServer.flush();
-- assert.strictEqual(bridgeCalls.at(-1).ops[0].op, 'broadcastAllRaw');
-- // Every viewer-facing path Chat does not yet own itself must keep reaching it over the bridge
-- // (review 2026-10-02, PR #12): alert sounds, the /api/mod per-socket reply, deploy notices and
-- // user-change invalidations. A durable write waits in the outbox until Chat acknowledges it.
-- chatServer.playAlertSound(streamer, streamId, 'donation');
-- chatServer.sendToConn({ remote: true }, { type: 'mod-action' });
-- chatServer.deployNotice([{ hash: 'deadbeef' }]);
-- chatServer.userChanged(viewer);
-- assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM chat_bridge_outbox').get().n, 1, 'a durable write waits in the outbox');
-- await chatServer.flush();
-- assert.deepStrictEqual(bridgeCalls.at(-1).ops.map((o) => o.op), ['playAlertSound', 'sendToConn', 'deployNotice', 'userChanged']);
-- assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM chat_bridge_outbox').get().n, 0, 'Chat acknowledged it');
++ assert.ok(!d.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'chat_bridge_outbox'").get(), 'no outbox');
++ // Live's own writes to data Chat caches (IP approvals, bans) send Chat a cache hint.
++ db.approveIp(channel.id, '203.0.113.7', streamer, 'manual');
++ db.forgiveBan(viewer);
++ for (let i = 0; i < 100 && ingressCalls.filter((c) => c.family === 'invalidate').length < 2; i++) await sleep(20);
++ assert.deepStrictEqual(ingressCalls.filter((c) => c.family === 'invalidate').map((c) => (c.body.bans ? 'bans' : c.body.approvals)), [channel.id, 'bans']);
// Synchronous reads come from Chat's presence snapshot.
-- await sleep(200);
++ for (let i = 0; i < 100 && chatServer.getTotalConnections() !== 7; i++) await sleep(20);
assert.strictEqual(chatServer.getTotalConnections(), 7);
assert.strictEqual(chatServer.getStreamViewerCount(1), 3);
assert.strictEqual(chatServer.slowModeByStream.get(1), 5000);
assert.strictEqual(chatServer.getConnectedUserIp(3), '198.51.100.3');
assert.strictEqual(chatServer.findClientByAnonId('anon9', 1).ip, '203.0.113.9');
assert.strictEqual(chatServer.findClientByAnonId('anon9', 2), null);
++ assert.strictEqual(chatServer.getAnonIdForConnection('203.0.113.9', 1), 'anon9');
++ // Chat asks Live for anon numbers; Live answers from its own helpers.
++ assert.strictEqual(typeof (await chatServer.resolveAnon('203.0.113.9')).anon_number, 'number');
++ // !arena answers the sender in the response, after its own lookup (review 2026-10-02, PR #12).
++ const arenaCmd = await call('POST', '/internal/chat-effects/arena-command', { token: WRITE, body: { cmd: '!arena', parts: ['!arena', 'nobody_here_at_all'], client: { conn_id: 'c1', streamId, ip: '203.0.113.30' } } });
++ assert.strictEqual(arenaCmd.status, 200);
++ assert.strictEqual(arenaCmd.body.handled, true);
++ assert.strictEqual(arenaCmd.body.replies.length, 1, 'the reply made after the lookup is in the answer');
++ assert.strictEqual(arenaCmd.body.replies[0].type, 'system');
// A /ws/chat upgrade that still lands on Live is refused, not served from the mirror.
let written = '';
chatServer.handleUpgrade({}, { write: (x) => { written += x; }, destroy() {} });
assert.match(written, /^HTTP\/1\.1 503/);
chatServer.close();
-- // Rollback: writes Chat never acknowledged are applied to Live's own tables.
-- d.prepare("INSERT INTO chat_bridge_outbox (boot, ref, op, args) VALUES ('old', -1, 'db', ?)").run(JSON.stringify(['recordFirstChat', 'user:77', streamer]));
-- chatServer._restoreDb();
-- assert.strictEqual(require('../server/chat/chat-remote').drainToLocal({ log() {}, warn() {} }), 1);
-- assert.ok(d.prepare("SELECT 1 FROM stream_first_chats WHERE chatter_key = 'user:77'").get());
-- assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM chat_bridge_outbox').get().n, 0);
--
// 11. Robot commands from chat pass the control panel's gate: control mode, anonymous switch, whitelist, cooldown.
{
const sent = [];
ddiff --git a/test/chat-ingress.test.js b/test/chat-ingress.test.js
iindex d2492616..076b7b00 100644
---- a/test/chat-ingress.test.js
++++ b/test/chat-ingress.test.js
@@@ -1,11 +1,10 @@
'use strict';
--// LIVE_CHAT_INGRESS=1 (T3 J2, first half): Live's chat producers call OpenVibe.Chat's typed
--// service-token ingress (/internal/chat/messages|events|moderation|invalidate, presence) through
--// server/chat/chat-delivery.js instead of the chat-remote.js bridge. Each producer family reaches
--// its endpoint with a bearer token and a stable idempotency key, nothing lands in
--// chat_bridge_outbox, and a Chat 5xx is logged and counted, never thrown and never bridged.
--// Against stub Network and Chat servers. The flag-off bridge path is test/chat-context.test.js.
++// CHAT_AUTHORITY=chat (T3 J2): Live's chat producers call OpenVibe.Chat's typed service-token
++// ingress (/internal/chat/messages|events|moderation|invalidate, presence) through
++// server/chat/chat-delivery.js — the only path; the old bridge (/internal/live/calls) and its outbox
++// are gone. Each producer family reaches its endpoint with a bearer token and a stable idempotency
++// key, and a Chat 5xx is logged and counted, never thrown. Against stub Network and Chat servers.
const assert = require('assert');
const fs = require('fs');
@@@ -26,7 +25,6 @@ process.env.OV_NETWORK_URL = 'https://openvibe.network';
process.env.OV_OAUTH_CLIENT_ID = 'live';
process.env.OV_OAUTH_CLIENT_SECRET = 'live-secret';
process.env.CHAT_AUTHORITY = 'chat';
--process.env.LIVE_CHAT_INGRESS = '1';
const quiet = console.log;
console.log = () => {};
console.warn = () => {};
@@@ -109,12 +107,11 @@ function assertSigned(c, what) {
db.createChannel({ user_id: streamer, title: 'Streamer TV' });
const channel = db.getChannelByUserId(streamer);
const streamId = Number(db.createStream({ user_id: streamer, channel_id: channel.id, title: 'Live now' }).lastInsertRowid);
-- const outbox = () => d.prepare('SELECT COUNT(*) AS n FROM chat_bridge_outbox').get().n;
const delivery = require('../server/chat/chat-delivery');
const chatServer = require('../server/chat/chat-server');
assert.strictEqual(chatServer.remote, true);
-- assert.strictEqual(delivery.ingress(), true, 'LIVE_CHAT_INGRESS=1 with CHAT_AUTHORITY=chat');
++ assert.strictEqual(delivery.ingress(), true, 'CHAT_AUTHORITY=chat is ingress, with no flag');
chatServer.init();
delivery.client._setRetryMs([20, 20]);
@@@ -124,7 +121,7 @@ function assertSigned(c, what) {
// 1. AI viewer line: one /messages call (persist + mirror + TTS); the caller gets no placeholder id.
const poster = require('../server/ai/viewers/poster');
const ret = poster.post({ streamId, userId: streamer, settings: { powerchat_forward: false } }, { id: 1, username: 'Botty', persona_json: '{}' }, 'hello from a bot');
-- assert.strictEqual(ret, null, 'no placeholder id with the flag on');
++ assert.strictEqual(ret, null, 'no placeholder id');
const ai = await waitFor(() => find('messages', (b) => b.source_platform === 'ai'), 'AI line');
assertSigned(ai, 'AI line');
assert.deepStrictEqual([ai.body.stream_id, ai.body.channel_user_id, ai.body.username, ai.body.mirror, ai.body.is_global, ai.body.frame.is_ai, ai.body.tts.identity_key],
@@@ -235,21 +232,38 @@ function assertSigned(c, what) {
assert.strictEqual(await delivery.moderate('log', { action_type: 'x' }), null, 'a 4xx resolves null');
assert.strictEqual(calls.filter((c) => c.family === 'moderation' && c.body.action_type === 'x').length, 1, 'a 4xx is not retried');
-- // 10. Nothing went over the bridge.
-- await chatServer.flush();
-- assert.strictEqual(outbox(), 0, 'nothing written to chat_bridge_outbox');
-- assert.deepStrictEqual(bridgeCalls.flatMap((c) => c.ops || []).map((o) => o.op), [], 'no bridge calls');
++ // 10. Nothing went over the bridge, and Live keeps no outbox.
++ assert.deepStrictEqual(bridgeCalls, [], 'no bridge calls');
++ assert.ok(!d.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'chat_bridge_outbox'").get(), 'no outbox table');
-- // 11. Flag off: the same seam is the legacy path again.
-- process.env.LIVE_CHAT_INGRESS = '';
++ // 11. A push a module still makes on the chat server goes to the ingress too (review 2026-10-02, PR #12);
++ // one with no ingress target is dropped, and TTS is never synthesised here (Chat speaks `tts` lines).
++ const tts = require('../server/chat/tts-engine');
++ const synth = Object.keys(tts).filter((k) => typeof tts[k] === 'function');
++ const touched = [];
++ const saved = {};
++ for (const k of synth) { saved[k] = tts[k]; tts[k] = (...a) => { touched.push(k); return saved[k](...a); }; }
++ const n = calls.length;
++ chatServer.broadcastToStream(streamId, { type: 'system', message: 'direct' });
++ chatServer.sendDm(viewer, { type: 'dm-notice' });
++ chatServer.disconnectUser({ userId: viewer, streamId });
++ chatServer.sendUserUpdate(viewer, { username: 'viewer', display_name: 'VIEWER', password_hash: 'never' });
++ chatServer.forwardToGlobal(streamId, { type: 'chat', message: 'no target' });
++ await chatServer.synthesizeAndBroadcastTTS(streamId, 'Bot', 'beep', null, null, 'aibot:bot', null, 'm1');
++ for (const k of synth) tts[k] = saved[k];
++ assert.deepStrictEqual(touched, [], 'no TTS synthesis in Live');
++ await waitFor(() => calls.length >= n + 4, 'direct pushes');
++ const direct = calls.slice(n).map((c) => [c.family, (c.body.target && c.body.target.kind) || c.body.action || (c.body.user != null ? 'user' : null)]);
++ assert.deepStrictEqual(direct.sort(), [['events', 'stream'], ['events', 'user'], ['invalidate', 'user'], ['moderation', 'disconnect']].sort());
++ assert.ok(!JSON.stringify(calls.slice(n)).includes('never'), 'only the public account fields');
++ // CHAT_AUTHORITY unset under a RemoteChatServer (a test or a bad env edit): dropped, never bounced back.
++ process.env.CHAT_AUTHORITY = '';
assert.strictEqual(delivery.ingress(), false);
-- chatServer.broadcastToStream(streamId, { type: 'system', message: 'x' });
-- delivery.event({ kind: 'stream', id: streamId }, { type: 'system', message: 'legacy' });
-- await chatServer.flush();
-- assert.deepStrictEqual(bridgeCalls.flatMap((c) => c.ops || []).map((o) => o.op), ['broadcastToStream', 'broadcastToStream']);
++ assert.strictEqual(chatServer.broadcastToStream(streamId, { type: 'system', message: 'x' }), undefined);
++ process.env.CHAT_AUTHORITY = 'chat';
chatServer.close();
-- quiet('chat ingress (LIVE_CHAT_INGRESS=1): all checks passed');
++ quiet('chat ingress (CHAT_AUTHORITY=chat): all checks passed');
} catch (err) {
console.error(err);
exit = 1;
ddiff --git a/test/fixtures/destructive-migrations.json b/test/fixtures/destructive-migrations.json
iindex d8bdb52c..5eb73d65 100644
---- a/test/fixtures/destructive-migrations.json
++++ b/test/fixtures/destructive-migrations.json
@@@ -16,6 +16,7 @@
{ "file": "server/db/migrations.js", "statement": "DROP TABLE IF EXISTS CHAT_TIMELINE_EVENTS", "count": 1, "contract": "roadmap T3 N+2 (005_drop_chat_staged_tables): OpenVibe.Chat owns the table and N-1 (8654ef6) no longer reads it (test/n-1.test.js); backups kept operator-side" },
{ "file": "server/db/migrations.js", "statement": "DROP TABLE IF EXISTS CHAT_STAGED_OUTBOX", "count": 1, "contract": "roadmap T3 N+2 (005_drop_chat_staged_tables): the staged-table relay was deleted in N+1 and N-1 (8654ef6) no longer uses it; backups kept operator-side" },
{ "file": "server/db/migrations.js", "statement": "DROP TABLE IF EXISTS CHAT_DUAL_READ_STATS", "count": 1, "contract": "roadmap T3 N+2 (005_drop_chat_staged_tables): the dual-read machinery was deleted in N+1 and N-1 (8654ef6) no longer uses it; backups kept operator-side" },
-- { "file": "server/db/migrations.js", "statement": "DROP TABLE IF EXISTS EMOTES", "count": 1, "contract": "roadmap T3 N+3 (006_drop_emotes): Chat owns emotes; N-1 fb3957f does not read its Live copy; backups kept operator-side" }
++ { "file": "server/db/migrations.js", "statement": "DROP TABLE IF EXISTS EMOTES", "count": 1, "contract": "roadmap T3 N+3 (006_drop_emotes): Chat owns emotes; N-1 fb3957f does not read its Live copy; backups kept operator-side" },
++ { "file": "server/db/migrations.js", "statement": "DROP TABLE IF EXISTS CHAT_BRIDGE_OUTBOX", "count": 1, "contract": "roadmap T3 J2 (op_002_drop_chat_bridge_outbox, operator-run via scripts/chat-bridge-outbox-drop.js): Live's chat bridge was deleted in this release; N-1 still writes the table, so the drop waits until N-1 is out of rollback range, delivers unacknowledged writes to Chat first and refuses while any remain; backups kept operator-side" }
]
}
ddiff --git a/test/release-events.test.js b/test/release-events.test.js
iindex 7fca92ea..5e316cb5 100644
---- a/test/release-events.test.js
++++ b/test/release-events.test.js
@@@ -102,30 +102,29 @@ const stub = http.createServer((req, res) => {
assert.strictEqual(deployRows().map(x => x.metadata).join('|'), chatBefore, 'the chat row rolled back too');
assert.strictEqual(outboxRows().length, 1);
-- // CHAT_AUTHORITY=chat with Events on: Chat learns the deploy from the event alone (C-84); no bridge hop.
-- const bridged = [];
-- r = await dn.announce({ db, chatServer: { remote: true, deployNotice: (commits) => bridged.push(commits) }, log: quiet });
++ // CHAT_AUTHORITY=chat with Events on: Chat learns the deploy from the event alone (C-84).
++ process.env.CHAT_AUTHORITY = 'chat';
++ r = await dn.announce({ db, chatServer: { remote: true }, log: quiet });
assert.ok(r.announced >= 1);
-- assert.strictEqual(bridged.length, 0, 'the event carries the deploy to Chat');
assert.strictEqual(db.getSetting(dn.SETTING), head);
rows = outboxRows();
assert.strictEqual(rows.length, 2);
assert.strictEqual(rows[1].event_id, r.event_id);
-- // Events publishing off: the bridge is the fallback, so Chat still hears about it.
++ // Events publishing off: Chat has no other way to hear about it (its ingress has no deploy endpoint), so
++ // the commits stay unannounced and the next boot tries again (review 2026-10-02, PR #12).
streamEvents._reset();
process.env.EVENTS_PUBLISH = 'off';
db.setSetting(dn.SETTING, '');
-- r = await dn.announce({ db, chatServer: { remote: true, deployNotice: (commits) => bridged.push(commits) }, log: quiet });
++ r = await dn.announce({ db, chatServer: { remote: true }, log: quiet });
delete process.env.EVENTS_PUBLISH;
-- assert.ok(r.announced >= 1);
-- assert.strictEqual(bridged.length, 1, 'without Events the bridge carries it');
-- assert.strictEqual(bridged[0][0].hash, head);
-- assert.strictEqual(r.event_id, null);
++ delete process.env.CHAT_AUTHORITY;
++ assert.strictEqual(r.announced, 0);
++ assert.strictEqual(db.getSetting(dn.SETTING), '', 'not recorded as announced');
assert.strictEqual(outboxRows().length, 2, 'nothing queued while Events is off');
streamEvents._reset();
stub.close();
-- console.log('release events: live.release.deployed queued with the announcement, published; Chat gets it from the event, the bridge only without Events — all checks passed');
++ console.log('release events: live.release.deployed queued with the announcement, published; Chat gets it from the event; without Events it stays unannounced — all checks passed');
process.exit(0);
})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex 95915b5c..e143ffbc 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -346,7 +346,6 @@ async function check(name, fn) {
'server/lineage/media-source.js': 'OpenVibe.Media (configured)',
'server/pastes-client.js': 'OpenVibe.Community (configured)',
'server/comments-client.js': 'OpenVibe.Community (configured)',
-- 'server/chat/chat-remote.js': 'OpenVibe.Chat (configured)',
'server/chat/moderation-client.js': 'OpenVibe.Chat internal read API (configured)',
'server/chat/insight-client.js': 'OpenVibe.Chat public chat-AI reads (configured OV_CHAT_INTERNAL_URL, fixed paths, ids validated)',
'server/chat/chat-client.js': 'OpenVibe.Chat typed ingress (configured OV_CHAT_INTERNAL_URL, fixed /internal/chat/* paths)',