openvibers/openvibe.live Commits

Staged chat tables move to OpenVibe.Chat one table at a time (WS-I task 2, register C-04): server/chat/chat-tables.js reads who writes channel_moderators, channel_moderation_settings, emotes, user_tags, chat_ai_summaries and chat_timeline_events from app_state chat_table_authority (every table 'live' by default, always 'live' without CHAT_AUTHORITY=chat), and every Live writer goes through its write(): the dashboard's moderators and settings, /slow and alert sounds, emote upload/edit/delete and the Media asset-sync, chat-AI summaries and timeline (test/chat-tables-writers.test.js keeps it so; user_tags has no writer). At 'live' Live writes its own table and TEMP-trigger capture relays every change to Chat's copy (chat_staged_outbox, bridge op stagedApply; POST /internal/chat-tables/relay pauses it for an import); a per-table dual read compares Live's reads with Chat's copy in the background and counts matched/mismatched/inconclusive in chat_dual_read_stats, never changing an answer. POST /internal/chat-tables/:table (X-Internal-Key) hands a table over, its writers waiting: Live's queue drained, Chat takes it, Live stops writing; back: Chat's mirror drained first, then Live writes again (force for Chat down); GET /internal/chat-tables shows both sides. At 'chat' Live's writers call Chat over the bridge (awaited, idempotency key) and apply Chat's rows to Live's copy, which is then the read mirror (applyMirror takes staged rows only for a table Chat writes). Nothing starts under LIVE_DRILL or without CHAT_AUTHORITY=chat; no schema change beyond two new tables. The emote upload's 409 path no longer throws inside its catch. Runbook: OpenVibe.Chat docs/staged-tables-cutover.md

OP OpenVibers committed 26 Sept 2026 · d6480da64c · 15 files, +1129 −35


Changes

ddiff --git a/AGENTS.md b/AGENTS.md
iindex 568fe5ae..0be2a098 100644
---- a/AGENTS.md
++++ b/AGENTS.md
@@@ -4,7 +4,7 @@
Self-hosted live streaming platform — Node.js/Express monolith, vanilla JS SPA frontend, SQLite (better-sqlite3). Part of the **OpenVibe network** with SSO via openvibe.network. See [README.md](README.md) for features and [docs/architecture.md](docs/architecture.md) for system design.
--**Network context:** SSO/OAuth2 + the OpenCoins wallet come from **OpenVibe.Network** (`server/monetization/wallet-client.js`). The media subsystem (VODs/clips/pastes/thumbnails/files) lives in **OpenVibe.Media** — Live talks to it via [server/media-client.js](server/media-client.js), the thin proxy routers in [server/media-proxy/](server/media-proxy/), and the Media-backed recorder [server/streaming/recorder.js](server/streaming/recorder.js). Media completion events arrive over OpenVibe.Events at `POST /internal/media-events` (`MEDIA_EVENTS_AUTHORITY`; the older `POST /internal/media-webhook` stays during the transition). The legacy local `vods`/`clips`/`pastes`/`paste_likes`/`paste_comments` tables are FROZEN: never read or write them (`test/frozen-tables.test.js` enforces it; ask Media or Community through `server/media-proxy/lookups.js`, `media-client.js`, `pastes-client.js`); VOD/clip comments are OpenVibe.Community threads ([server/comments-client.js](server/comments-client.js)) and the local `comments` table is READ-ONLY too; Live-owned AI/transcript state for Media-hosted content lives in `vod_ai_state`/`clip_ai_state`. See ../CONTRACTS.md for the binding inter-service contracts. **Money:** `BILLING_AUTHORITY` (read only in [server/monetization/money-authority.js](server/monetization/money-authority.js)) is `live` by default; `billing` makes OpenVibe.Billing the ledger ([billing-client.js](server/monetization/billing-client.js), [billing-actions.js](server/monetization/billing-actions.js)) and Live's `openvibe_bucks_*`/`transactions`/`payment_orders`/`subscriptions` read-only (a database-level tripwire throws). The owner-only `money_writes_frozen` flag (`/api/admin/money`) refuses money actions in both modes.
++**Network context:** SSO/OAuth2 + the OpenCoins wallet come from **OpenVibe.Network** (`server/monetization/wallet-client.js`). The media subsystem (VODs/clips/pastes/thumbnails/files) lives in **OpenVibe.Media** — Live talks to it via [server/media-client.js](server/media-client.js), the thin proxy routers in [server/media-proxy/](server/media-proxy/), and the Media-backed recorder [server/streaming/recorder.js](server/streaming/recorder.js). Media completion events arrive over OpenVibe.Events at `POST /internal/media-events` (`MEDIA_EVENTS_AUTHORITY`; the older `POST /internal/media-webhook` stays during the transition). The legacy local `vods`/`clips`/`pastes`/`paste_likes`/`paste_comments` tables are FROZEN: never read or write them (`test/frozen-tables.test.js` enforces it; ask Media or Community through `server/media-proxy/lookups.js`, `media-client.js`, `pastes-client.js`); VOD/clip comments are OpenVibe.Community threads ([server/comments-client.js](server/comments-client.js)) and the local `comments` table is READ-ONLY too; the staged chat tables (`channel_moderators`, `channel_moderation_settings`, `emotes`, `user_tags`, `chat_ai_summaries`, `chat_timeline_events`) move to OpenVibe.Chat one at a time (register C-04, runbook in Chat `docs/staged-tables-cutover.md`), so write them only through `await require('./chat/chat-tables').write(...)` (`test/chat-tables-writers.test.js` enforces it); Live-owned AI/transcript state for Media-hosted content lives in `vod_ai_state`/`clip_ai_state`. See ../CONTRACTS.md for the binding inter-service contracts. **Money:** `BILLING_AUTHORITY` (read only in [server/monetization/money-authority.js](server/monetization/money-authority.js)) is `live` by default; `billing` makes OpenVibe.Billing the ledger ([billing-client.js](server/monetization/billing-client.js), [billing-actions.js](server/monetization/billing-actions.js)) and Live's `openvibe_bucks_*`/`transactions`/`payment_orders`/`subscriptions` read-only (a database-level tripwire throws). The owner-only `money_writes_frozen` flag (`/api/admin/money`) refuses money actions in both modes.
## Commands
ddiff --git a/server/admin/channel-mod-routes.js b/server/admin/channel-mod-routes.js
iindex f1e4e3f1..c8d49382 100644
---- a/server/admin/channel-mod-routes.js
++++ b/server/admin/channel-mod-routes.js
@@@ -18,6 +18,8 @@ const db = require('../db/database');
const { requireAuth } = require('../auth/auth');
const permissions = require('../auth/permissions');
const chatServer = require('../chat/chat-server');
++// Moderators and settings are written wherever their table's authority is (Live or OpenVibe.Chat).
++const chatTables = require('../chat/chat-tables');
const router = express.Router({ mergeParams: true });
@@@ -86,7 +88,7 @@ router.get('/:channelId/mods', requireAuth, (req, res) => {
});
// ── Add channel mod ──────────────────────────────────────────
--router.post('/:channelId/mods', requireAuth, (req, res) => {
++router.post('/:channelId/mods', requireAuth, async (req, res) => {
try {
const channelId = parseInt(req.params.channelId);
const channel = db.getChannelById(channelId);
@@@ -106,7 +108,7 @@ router.post('/:channelId/mods', requireAuth, (req, res) => {
return res.status(400).json({ error: 'Channel owner is already a moderator' });
}
-- db.addChannelModerator(channelId, targetUser.id, req.user.id);
++ await chatTables.write('addChannelModerator', channelId, targetUser.id, req.user.id);
db.logModerationAction({
scope_type: 'channel',
@@@ -124,7 +126,7 @@ router.post('/:channelId/mods', requireAuth, (req, res) => {
});
// ── Remove channel mod ───────────────────────────────────────
--router.delete('/:channelId/mods/:userId', requireAuth, (req, res) => {
++router.delete('/:channelId/mods/:userId', requireAuth, async (req, res) => {
try {
const channelId = parseInt(req.params.channelId);
const userId = parseInt(req.params.userId);
@@@ -136,7 +138,7 @@ router.delete('/:channelId/mods/:userId', requireAuth, (req, res) => {
}
const target = db.getUserById(userId);
-- db.removeChannelModerator(channelId, userId);
++ await chatTables.write('removeChannelModerator', channelId, userId);
db.logModerationAction({
scope_type: 'channel',
@@@ -167,12 +169,12 @@ router.get('/:channelId/moderation', requireAuth, requireChannelAccess, (req, re
// Settings that decide what moderators themselves may do, or who can chat at all, belong to the
// channel owner. A mod could otherwise grant themselves About-page editing or switch off IP approval.
const OWNER_POLICY_KEYS = ['mods_can_edit_about', 'ip_approval_mode', 'uploads_mods_only', 'allow_anonymous', 'sounds_mods_only'];
--router.put('/:channelId/moderation', requireAuth, requireChannelAccess, (req, res) => {
++router.put('/:channelId/moderation', requireAuth, requireChannelAccess, async (req, res) => {
try {
const isOwnerOrStaff = req.channel.user_id === req.user.id || permissions.isGlobalModOrAbove(req.user);
// The dashboard form sends every field, so for a moderator these are ignored rather than refused.
if (!isOwnerOrStaff) for (const k of OWNER_POLICY_KEYS) delete req.body[k];
-- const settings = db.upsertChannelModerationSettings(req.channel.id, {
++ const settings = await chatTables.write('upsertChannelModerationSettings', req.channel.id, {
slow_mode_seconds: req.body.slow_mode_seconds !== undefined
? Math.max(0, parseInt(req.body.slow_mode_seconds) || 0) : undefined,
slowmode_seconds: req.body.slowmode_seconds !== undefined
ddiff --git a/server/ai/chat-ai.js b/server/ai/chat-ai.js
iindex 08ffe311..7b02e0df 100644
---- a/server/ai/chat-ai.js
++++ b/server/ai/chat-ai.js
@@@ -20,6 +20,9 @@
'use strict';
const db = require('../db/database');
++// chat_ai_summaries and chat_timeline_events are staged chat tables: every write goes where the
++// table's authority is — Live's own table, or OpenVibe.Chat (chat/chat-tables.js).
++const chatTables = require('../chat/chat-tables');
// ── Tunables ─────────────────────────────────────────────────────────────────
const TICK_MS = 45 * 1000;
@@@ -186,8 +189,8 @@ ${_fmtMessages(rows, { includeChannel: true, now })}`;
const nowIso = _sqlTime(now);
// Persist the new moments to the growing timeline log (for the browsable/searchable view).
-- try { db.addChatTimelineEvents('global', 0, _stampAdditions(parsed.timeline, nowIso, now)); } catch { /* */ }
-- db.upsertChatAiSummary({
++ try { await chatTables.write('addChatTimelineEvents', 'global', 0, _stampAdditions(parsed.timeline, nowIso, now)); } catch { /* */ }
++ await chatTables.write('upsertChatAiSummary', {
scope: 'global', subject_id: 0, window: 'global',
overview: _clip(parsed.recent_overview || '', 2000),
memory_json: _clip(parsed.memory || priorMemory, MEMORY_MAX_CHARS),
@@@ -250,7 +253,7 @@ ${_fmtMessages(dayRows, { now })}`;
const _lastRow = dayRows[dayRows.length - 1] || {};
const activityTs = _lastRow.timestamp || _lastRow.created_at || nowIso;
// Canonical rolling row: memory + timeline + high-water + both overviews (as JSON).
-- db.upsertChatAiSummary({
++ await chatTables.write('upsertChatAiSummary', {
scope: 'user', subject_id: uid, window: 'rolling',
overview: JSON.stringify({
today: _clip(parsed.overview_24h || '', 1200),
@@@ -325,7 +328,7 @@ ${_fmtMessages(dayRows, { now })}`;
if (!parsed) { console.warn(`[ChatAI] relay ${ru.id}: unparseable model output`); return false; }
const nowIso = _sqlTime(now);
-- db.upsertChatAiSummary({
++ await chatTables.write('upsertChatAiSummary', {
scope: 'relay', subject_id: ru.id, window: 'rolling',
overview: JSON.stringify({
today: _clip(parsed.overview_24h || '', 1200),
@@@ -396,7 +399,7 @@ ${_fmtMessages(dayRows, { now })}`;
if (!parsed) { console.warn(`[ChatAI] anon ${anonId}: unparseable model output`); return false; }
const nowIso = _sqlTime(now);
-- db.upsertChatAiSummary({
++ await chatTables.write('upsertChatAiSummary', {
scope: 'anon', subject_id: subjectId, window: 'rolling',
overview: JSON.stringify({
today: _clip(parsed.overview_24h || '', 1200),
@@@ -455,21 +458,21 @@ async function _tick() {
// One-time: backfill the growing timeline log from the existing summary JSON so the browsable
// view has history immediately (before new moments accumulate).
--function _seedTimelineEvents() {
++async function _seedTimelineEvents() {
try {
if ((db.getChatTimelineEvents({ scope: 'global', limit: 1 }) || []).length) return;
const row = db.getChatAiSummary('global', 0, 'global');
if (!row) return;
let tl = []; try { tl = JSON.parse(row.timeline_json || '[]'); } catch { tl = []; }
const cleaned = _cleanTimeline(tl);
-- if (cleaned.length) { db.addChatTimelineEvents('global', 0, cleaned); console.log(`[ChatAI] Seeded ${cleaned.length} timeline event(s)`); }
++ if (cleaned.length) { await chatTables.write('addChatTimelineEvents', 'global', 0, cleaned); console.log(`[ChatAI] Seeded ${cleaned.length} timeline event(s)`); }
} catch { /* */ }
}
function start() {
if (_running) return;
_running = true;
-- try { _seedTimelineEvents(); } catch { /* */ }
++ _seedTimelineEvents().catch(() => {});
_timer = setInterval(() => { _tick().catch(() => {}); }, TICK_MS);
if (_timer.unref) _timer.unref();
console.log('[AI] Chat-AI job started (global overview/timeline + per-user insights)');
ddiff --git a/server/chat/chat-server.js b/server/chat/chat-server.js
iindex d88484ab..2919e9b1 100644
---- a/server/chat/chat-server.js
++++ b/server/chat/chat-server.js
@@@ -1576,7 +1576,7 @@ class ChatServer {
try {
const stream = db.getStreamById(client.streamId);
if (stream?.channel_id) {
-- db.upsertChannelModerationSettings(stream.channel_id, { slow_mode_seconds: seconds });
++ require('./chat-tables').write('upsertChannelModerationSettings', stream.channel_id, { slow_mode_seconds: seconds }).catch(() => { /* non-critical */ });
}
} catch { /* non-critical */ }
}
ddiff --git a/server/chat/chat-tables-sync.js b/server/chat/chat-tables-sync.js
nnew file mode 100644
iindex 00000000..24adbe84
---- /dev/null
++++ b/server/chat/chat-tables-sync.js
@@@ -0,0 +1,325 @@
++'use strict';
++/**
++ * Keeping OpenVibe.Chat's copy of the staged tables current while Live writes them, and checking it
++ * (roadmap C-04; chat-tables.js says who writes what). Started by chat-tables.init() only.
++ *
++ * Capture. TEMP triggers on Live's connection record every insert, update and delete on a staged
++ * table in chat_staged_outbox, while the table is listed in temp.chat_staged_capture — the tables Live
++ * writes. Rows Chat's mirror applies to a table Chat writes are never captured.
++ *
++ * Relay. Every 2 s the outbox goes to Chat's bridge as `stagedApply` — each row as it is now (so
++ * repeated changes to a row collapse into one), in the order of their last change — and leaves the
++ * outbox once Chat has acknowledged it. drain(table) empties one table's part before a handoff.
++ * Paused (app_state `chat_table_relay_paused`, POST /internal/chat-tables/relay) the capture goes on
++ * and nothing is sent: pause, snapshot live.db, import the snapshot into Chat, resume — every change
++ * made after the pause then reaches Chat after the import, so the import never undoes one.
++ *
++ * Dual read. With a table's flag on, Live's reads of it (the database.js readers below, and
++ * tags.getUserTags) also ask Chat for the same slice — the rows whose columns equal the read's keys —
++ * and compare row count and content hash (Chat's database.js sliceHash, computed the same way here).
++ * After the read, in the background, at most once a minute per slice and 30 times a minute in all;
++ * inconclusive (not counted as a mismatch) while the table still has changes on their way to Chat
++ * or when Live's own rows changed during the comparison. Counts are kept in chat_dual_read_stats.
++ * Live's answer is never changed.
++ */
++const crypto = require('crypto');
++const db = require('../db/database');
++const chatTables = require('./chat-tables');
++
++const BATCH = 200;
++const RELAY_MS = 2000;
++const SLICE_EVERY_MS = 60_000;
++const COMPARES_PER_MIN = 30;
++
++const relay = { sent: 0, failed: 0, lastError: null, lastSentAt: null };
++let started = false;
++const PAUSE_KEY = 'chat_table_relay_paused';
++
++function paused() { try { return db.getState(PAUSE_KEY) === '1'; } catch { return false; } }
++function setPaused(on) {
++ if (on) db.setState(PAUSE_KEY, '1'); else db.deleteState(PAUSE_KEY);
++ return paused();
++}
++
++function ensureTables() {
++ db.getDb().exec(`
++ CREATE TABLE IF NOT EXISTS chat_staged_outbox (
++ seq INTEGER PRIMARY KEY AUTOINCREMENT,
++ tbl TEXT NOT NULL,
++ op TEXT NOT NULL,
++ pk TEXT NOT NULL,
++ created_at DATETIME DEFAULT CURRENT_TIMESTAMP
++ );
++ CREATE TABLE IF NOT EXISTS chat_dual_read_stats (
++ tbl TEXT PRIMARY KEY,
++ compared INTEGER NOT NULL DEFAULT 0,
++ matched INTEGER NOT NULL DEFAULT 0,
++ mismatched INTEGER NOT NULL DEFAULT 0,
++ inconclusive INTEGER NOT NULL DEFAULT 0,
++ errors INTEGER NOT NULL DEFAULT 0,
++ last_mismatch_at DATETIME,
++ last_mismatch TEXT,
++ since DATETIME DEFAULT CURRENT_TIMESTAMP
++ );
++ `);
++}
++
++// ── Capture ─────────────────────────────────────────────────────
++
++function installCapture() {
++ const d = db.getDb();
++ d.exec('CREATE TEMP TABLE IF NOT EXISTS chat_staged_capture (tbl TEXT PRIMARY KEY)');
++ for (const [table, pk] of Object.entries(chatTables.TABLES)) {
++ const obj = (alias) => `json_object(${pk.map((c) => `'${c}', ${alias}.${c}`).join(', ')})`;
++ const when = `WHEN EXISTS (SELECT 1 FROM temp.chat_staged_capture WHERE tbl = '${table}')`;
++ try {
++ d.exec(`
++ CREATE TEMP TRIGGER IF NOT EXISTS chat_staged_${table}_ins AFTER INSERT ON main.${table} ${when}
++ BEGIN INSERT INTO chat_staged_outbox (tbl, op, pk) VALUES ('${table}', 'upsert', ${obj('NEW')}); END;
++ CREATE TEMP TRIGGER IF NOT EXISTS chat_staged_${table}_upd AFTER UPDATE ON main.${table} ${when}
++ BEGIN INSERT INTO chat_staged_outbox (tbl, op, pk) VALUES ('${table}', 'upsert', ${obj('NEW')}); END;
++ CREATE TEMP TRIGGER IF NOT EXISTS chat_staged_${table}_del AFTER DELETE ON main.${table} ${when}
++ BEGIN INSERT INTO chat_staged_outbox (tbl, op, pk) VALUES ('${table}', 'delete', ${obj('OLD')}); END;
++ `);
++ } catch (err) { console.warn(`[ChatTables] capture for ${table}: ${err.message}`); }
++ setCapture(table, chatTables.authority(table) === 'live');
++ }
++}
++
++function setCapture(table, on) {
++ const d = db.getDb();
++ if (on) d.prepare('INSERT OR IGNORE INTO temp.chat_staged_capture (tbl) VALUES (?)').run(table);
++ else d.prepare('DELETE FROM temp.chat_staged_capture WHERE tbl = ?').run(table);
++}
++
++function pending(table = null) {
++ try {
++ return table
++ ? db.get('SELECT COUNT(*) AS n FROM chat_staged_outbox WHERE tbl = ?', [table]).n
++ : db.get('SELECT COUNT(*) AS n FROM chat_staged_outbox').n;
++ } catch { return 0; } // no outbox: nothing was ever captured
++}
++
++// ── Relay ───────────────────────────────────────────────────────
++
++/** The outbox rows as changes: each row as it is now, ordered by its last change. */
++function buildChanges(rows) {
++ const last = new Map();
++ for (const r of rows) last.set(`${r.tbl}|${r.pk}`, r);
++ const changes = [];
++ for (const r of [...last.values()].sort((a, b) => a.seq - b.seq)) {
++ const cols = chatTables.TABLES[r.tbl];
++ if (!cols) continue;
++ let pk;
++ try { pk = JSON.parse(r.pk); } catch { continue; }
++ const row = r.op === 'delete' ? null : db.get(`SELECT * FROM ${r.tbl} WHERE ${cols.map((c) => `${c} = ?`).join(' AND ')}`, cols.map((c) => pk[c]));
++ changes.push(row ? { table: r.tbl, op: 'upsert', row } : { table: r.tbl, op: 'delete', pk });
++ }
++ return changes;
++}
++
++let busy = null;
++/** Send the outbox (or only one table's rows) to Chat. → { sent, pending, error } */
++function flush(table = null) {
++ if (busy) return busy.then(() => flush(table));
++ if (paused()) return Promise.resolve({ sent: 0, pending: pending(table), error: 'relay paused' });
++ busy = (async () => {
++ let sent = 0;
++ for (;;) {
++ const rows = table
++ ? db.all('SELECT seq, tbl, op, pk FROM chat_staged_outbox WHERE tbl = ? ORDER BY seq LIMIT ?', [table, BATCH])
++ : db.all('SELECT seq, tbl, op, pk FROM chat_staged_outbox ORDER BY seq LIMIT ?', [BATCH]);
++ if (!rows.length) break;
++ const changes = buildChanges(rows);
++ let out;
++ try {
++ out = await chatTables.callChat('stagedApply', [changes]);
++ } catch (err) {
++ relay.failed++;
++ if (relay.lastError !== err.message) console.warn(`[ChatTables] relay to Chat waiting (${pending()} queued): ${err.message}`);
++ relay.lastError = err.message;
++ break;
++ }
++ // Chat refusing a change means it thinks it writes that table: keep those rows (never lose
++ // Live's write), report it, and stop until an operator makes both sides agree.
++ const refused = new Set((out.skipped || []).filter((s) => /table_authority chat/.test(s.reason || '')).map((s) => s.table));
++ const done = rows.filter((r) => !refused.has(r.tbl)).map((r) => r.seq);
++ for (let i = 0; i < done.length; i += 500) {
++ const part = done.slice(i, i + 500);
++ db.run(`DELETE FROM chat_staged_outbox WHERE seq IN (${part.map(() => '?').join(',')})`, part);
++ }
++ sent += changes.length;
++ relay.sent += changes.length;
++ relay.lastSentAt = new Date().toISOString();
++ if (refused.size) {
++ relay.lastError = `OpenVibe.Chat writes ${[...refused].join(', ')} but Live does too: run the handoff again (POST /internal/chat-tables/:table)`;
++ console.warn(`[ChatTables] ${relay.lastError}`);
++ break;
++ }
++ relay.lastError = null;
++ const other = (out.skipped || []).filter((s) => !refused.has(s.table));
++ if (other.length) console.warn(`[ChatTables] Chat skipped ${other.length} change(s):`, JSON.stringify(other.slice(0, 3)));
++ }
++ return { sent, pending: pending(table), error: relay.lastError };
++ })().finally(() => { busy = null; });
++ return busy;
++}
++
++/** Before a handoff to Chat: every captured change of the table acknowledged, or throw. */
++async function drain(table, { timeoutMs = 20000 } = {}) {
++ const until = Date.now() + timeoutMs;
++ while (pending(table) > 0) {
++ const r = await flush(table);
++ if (!pending(table)) break;
++ if (r.error || Date.now() > until) {
++ throw Object.assign(new Error(`${pending(table)} change(s) to ${table} not in OpenVibe.Chat yet${r.error ? `: ${r.error}` : ''}`), { status: 503 });
++ }
++ }
++}
++
++function relayStats() { return { ...relay, paused: paused(), pending: pending() }; }
++
++// ── Dual read ───────────────────────────────────────────────────
++
++const num = (v) => (typeof v === 'string' && /^-?\d+$/.test(v) ? Number(v) : v);
++// Live's readers of the staged tables and the slices each read touches.
++const READERS = [
++ ['getChannelModerationSettings', 'channel_moderation_settings', (a) => [{ channel_id: num(a[0]) }]],
++ ['getChannelModerators', 'channel_moderators', (a) => [{ channel_id: num(a[0]) }]],
++ ['isChannelModerator', 'channel_moderators', (a) => [{ channel_id: num(a[1]) }]],
++ ['getChannelsByModerator', 'channel_moderators', (a) => [{ user_id: num(a[0]) }]],
++ ['getEmoteById', 'emotes', (a) => [{ id: num(a[0]) }]],
++ ['getEmotesByUser', 'emotes', (a) => [{ user_id: num(a[0]) }]],
++ ['getChannelEmotes', 'emotes', (a) => [{ channel_owner_id: num(a[0]) }, { user_id: num(a[0]), channel_owner_id: null }]],
++ ['getGlobalEmotes', 'emotes', () => [{ is_global: 1 }]],
++ ['getChatAiSummary', 'chat_ai_summaries', (a) => [{ scope: a[0], subject_id: num(a[1]) || 0, window: a[2] }]],
++ ['getChatAiSummaries', 'chat_ai_summaries', (a) => [{ scope: a[0], subject_id: num(a[1]) || 0 }]],
++ ['getChatTimelineEvents', 'chat_timeline_events', (a) => [{ scope: (a[0] && a[0].scope) || 'global', subject_id: num(a[0] && a[0].subjectId) || 0 }]],
++];
++
++/** Chat's sliceHash (OpenVibe.Chat server/db/database.js): sha256 of the rows as arrays in column order. */
++function sliceHash(columns, rows) {
++ const body = JSON.stringify(rows.map((r) => columns.map((c) => (r[c] === undefined ? null : r[c]))));
++ return crypto.createHash('sha256').update(body).digest('hex');
++}
++const _cols = new Map();
++function liveColumns(table) {
++ if (!_cols.has(table)) _cols.set(table, db.getDb().prepare(`PRAGMA table_info(${table})`).all().map((c) => c.name).sort());
++ return _cols.get(table);
++}
++function localSlice(table, where, columns) {
++ const keys = Object.keys(where);
++ const rows = db.all(`SELECT ${columns.join(', ')} FROM ${table} WHERE ${keys.map((k) => `${k} IS ?`).join(' AND ') || '1'} ORDER BY ${chatTables.TABLES[table].join(', ')}`, keys.map((k) => where[k]));
++ return { count: rows.length, hash: sliceHash(columns, rows), rows };
++}
++
++function bump(table, field, mismatch = null) {
++ try {
++ db.run(`INSERT INTO chat_dual_read_stats (tbl, compared, ${field}) VALUES (?, 1, 1)
++ ON CONFLICT(tbl) DO UPDATE SET compared = compared + 1, ${field} = ${field} + 1`, [table]);
++ if (mismatch) db.run('UPDATE chat_dual_read_stats SET last_mismatch_at = CURRENT_TIMESTAMP, last_mismatch = ? WHERE tbl = ?', [JSON.stringify(mismatch).slice(0, 2000), table]);
++ } catch { /* counting must never break a read */ }
++}
++
++function dualReadStats(table) {
++ try {
++ const r = db.get('SELECT compared, matched, mismatched, inconclusive, errors, last_mismatch_at, last_mismatch, since FROM chat_dual_read_stats WHERE tbl = ?', [table]);
++ if (!r) return { compared: 0, matched: 0, mismatched: 0, inconclusive: 0, errors: 0, last_mismatch_at: null, last_mismatch: null, since: null };
++ let last = null; try { last = r.last_mismatch ? JSON.parse(r.last_mismatch) : null; } catch { last = r.last_mismatch; }
++ return { ...r, last_mismatch: last };
++ } catch { return null; }
++}
++function resetDualReadStats(table) { try { db.run('DELETE FROM chat_dual_read_stats WHERE tbl = ?', [table]); } catch { /* */ } }
++
++const lastSlice = new Map(); // `${table}|${where}` → when last compared
++let budget = { minute: 0, used: 0 };
++
++/** Compare one slice with Chat's copy (in the background). → 'matched' | 'mismatched' | 'inconclusive' | 'error' */
++async function compare(table, where) {
++ const columns = liveColumns(table);
++ if (pending(table) > 0) { bump(table, 'inconclusive'); return 'inconclusive'; }
++ const before = localSlice(table, where, columns);
++ let theirs;
++ try {
++ theirs = await chatTables.callChat('stagedSlice', [table, where, columns], { timeoutMs: 5000 });
++ } catch (err) {
++ bump(table, 'errors');
++ return 'error';
++ }
++ const cols = Array.isArray(theirs.columns) ? theirs.columns : columns;
++ const mine = localSlice(table, where, cols);
++ // Live's rows changed meanwhile, or a change is on its way: no verdict.
++ if (mine.count !== before.count || localSlice(table, where, columns).hash !== before.hash || pending(table) > 0) { bump(table, 'inconclusive'); return 'inconclusive'; }
++ if (mine.count === theirs.count && mine.hash === theirs.hash) { bump(table, 'matched'); return 'matched'; }
++ const key = (r) => JSON.stringify(chatTables.TABLES[table].map((k) => r[k]));
++ const sample = { where, live: { count: mine.count, hash: mine.hash.slice(0, 16) }, chat: { count: theirs.count, hash: String(theirs.hash).slice(0, 16) } };
++ if (Array.isArray(theirs.rows) && mine.rows.length <= 50) {
++ const a = new Map(mine.rows.map((r) => [key(r), JSON.stringify(cols.map((c) => r[c] ?? null))]));
++ const b = new Map(theirs.rows.map((r) => [key(r), JSON.stringify(cols.map((c) => r[c] ?? null))]));
++ sample.only_live = [...a.keys()].filter((k) => !b.has(k)).slice(0, 5);
++ sample.only_chat = [...b.keys()].filter((k) => !a.has(k)).slice(0, 5);
++ sample.differ = [...a.keys()].filter((k) => b.has(k) && a.get(k) !== b.get(k)).slice(0, 5);
++ }
++ bump(table, 'mismatched', sample);
++ console.warn(`[ChatTables] dual read: ${table} ${JSON.stringify(where)} differs — Live ${mine.count} row(s), Chat ${theirs.count}`);
++ return 'mismatched';
++}
++
++/** A read of `table` touched these slices: compare them later, within the rate limits. */
++function observe(table, slices) {
++ if (chatTables.authority(table) !== 'live' || !chatTables.dualRead(table)) return;
++ const now = Date.now();
++ const minute = Math.floor(now / 60_000);
++ if (budget.minute !== minute) budget = { minute, used: 0 };
++ for (const where of slices) {
++ const k = `${table}|${JSON.stringify(where)}`;
++ if (now - (lastSlice.get(k) || 0) < SLICE_EVERY_MS || budget.used >= COMPARES_PER_MIN) continue;
++ lastSlice.set(k, now);
++ if (lastSlice.size > 5000) lastSlice.delete(lastSlice.keys().next().value);
++ budget.used++;
++ setImmediate(() => { compare(table, where).catch(() => {}); });
++ }
++}
++
++const originals = {};
++function installDualRead() {
++ for (const [fn, table, slices] of READERS) {
++ if (typeof db[fn] !== 'function' || originals[fn]) continue;
++ const orig = originals[fn] = db[fn];
++ db[fn] = function dualReadObserved(...args) {
++ const out = orig.apply(this, args);
++ try { observe(table, slices(args)); } catch { /* never breaks the read */ }
++ return out;
++ };
++ }
++ try {
++ const tags = require('./tags');
++ if (!originals.getUserTags) {
++ const orig = originals.getUserTags = tags.getUserTags;
++ tags.getUserTags = function dualReadObserved(userId) {
++ const out = orig(userId);
++ try { observe('user_tags', [{ user_id: num(userId) }]); } catch { /* */ }
++ return out;
++ };
++ }
++ } catch { /* no tags module */ }
++}
++
++// ── Start ───────────────────────────────────────────────────────
++
++function start() {
++ if (started) return;
++ started = true;
++ ensureTables();
++ installCapture();
++ installDualRead();
++ require('../utils/jobs').every('chat-staged-relay', RELAY_MS, () => flush(), { initialDelayMs: RELAY_MS });
++ const queued = pending();
++ console.log(`[ChatTables] staged tables: ${Object.keys(chatTables.TABLES).map((t) => `${t}=${chatTables.authority(t)}${chatTables.dualRead(t) ? '+dual-read' : ''}`).join(', ')}${queued ? ` (${queued} change(s) queued for Chat)` : ''}`);
++}
++
++module.exports = {
++ start, ensureTables, installCapture, setCapture, pending, flush, drain, buildChanges, relayStats, paused, setPaused,
++ compare, observe, dualReadStats, resetDualReadStats, sliceHash, READERS,
++ _restore() { for (const [fn, orig] of Object.entries(originals)) { if (fn === 'getUserTags') require('./tags').getUserTags = orig; else db[fn] = orig; delete originals[fn]; } started = false; },
++};
ddiff --git a/server/chat/chat-tables.js b/server/chat/chat-tables.js
nnew file mode 100644
iindex 00000000..f73c8fea
---- /dev/null
++++ b/server/chat/chat-tables.js
@@@ -0,0 +1,241 @@
++'use strict';
++/**
++ * The staged chat tables (roadmap C-04; OpenVibe.Chat docs/staged-tables-cutover.md): who writes
++ * channel_moderators, channel_moderation_settings, emotes, user_tags, chat_ai_summaries and
++ * chat_timeline_events, one table at a time.
++ *
++ * authority(table) is 'live' or 'chat', from app_state `chat_table_authority` (JSON; a table not in
++ * it is 'live', so deploying this changes nothing). It is always 'live' unless CHAT_AUTHORITY=chat:
++ * without it Live runs chat itself and OpenVibe.Chat is out of the picture.
++ *
++ * 'live' Live writes its table, as before (write() runs Live's own function). Every change is
++ * captured and relayed to Chat's copy (chat-tables-sync.js). With the table's dual-read
++ * flag (app_state `chat_table_dual_read`), Live's reads of it are also compared with Chat's
++ * copy and mismatches are counted; Live's answers never change.
++ * 'chat' Chat writes it. write() hands the writer's call to Chat over the bridge (POST
++ * /internal/live/calls, chat.live_bridge.write, awaited, with an idempotency key) and
++ * applies the rows Chat answers to Live's table, which is from then on the read mirror
++ * Chat keeps current (POST /internal/chat-effects/mirror). Live's readers do not change.
++ *
++ * Moving a table is setAuthority() (POST /internal/chat-tables/:table with X-Internal-Key), never
++ * automatic. The table's writers wait while it runs. To 'chat': Live's queued changes reach Chat,
++ * Chat takes the table, Live stops writing it. Back to 'live': Chat sends Live every change it made
++ * to the table and gives it back, then Live writes it again. Running it again is safe and brings Live
++ * and Chat back into agreement after a failed attempt.
++ *
++ * Nothing here starts at require time; init() (server start, CHAT_AUTHORITY=chat only) installs the
++ * capture, the dual read and the relay. A restore drill never calls it.
++ */
++const crypto = require('crypto');
++const db = require('../db/database');
++const { isRemote, CHAT_URL } = require('./chat-authority');
++
++// The staged tables and their keys (Chat's STAGED_KEYS).
++const TABLES = {
++ channel_moderators: ['id'],
++ channel_moderation_settings: ['channel_id'],
++ emotes: ['id'],
++ user_tags: ['id'],
++ chat_ai_summaries: ['id'],
++ chat_timeline_events: ['id'],
++};
++// Live's writes to them: database.js function → table. Chat's database.js has the same functions.
++// user_tags has no writer in Live (chat tags are read-only since the game moved to OpenVibe.Games).
++const OPS = {
++ addChannelModerator: 'channel_moderators',
++ removeChannelModerator: 'channel_moderators',
++ upsertChannelModerationSettings: 'channel_moderation_settings',
++ setChannelAlertSound: 'channel_moderation_settings',
++ createEmote: 'emotes',
++ updateEmote: 'emotes',
++ deleteEmote: 'emotes',
++ setEmoteMedia: 'emotes',
++ upsertChatAiSummary: 'chat_ai_summaries',
++ addChatTimelineEvents: 'chat_timeline_events',
++};
++
++const AUTHORITY_KEY = 'chat_table_authority';
++const DUAL_READ_KEY = 'chat_table_dual_read';
++const AUDIENCE = 'openvibe.chat';
++const BOOT = `tables-${crypto.randomUUID()}`;
++
++function httpError(status, message) { return Object.assign(new Error(message), { status }); }
++
++// ── Authority and dual-read flags ───────────────────────────────
++
++function readJson(key) {
++ try {
++ const v = db.getState(key);
++ const o = v ? JSON.parse(v) : {};
++ return o && typeof o === 'object' && !Array.isArray(o) ? o : {};
++ } catch { return {}; }
++}
++let cache = { at: 0, authority: {}, dual: {} };
++function flags() {
++ if (Date.now() - cache.at > 1000) cache = { at: Date.now(), authority: readJson(AUTHORITY_KEY), dual: readJson(DUAL_READ_KEY) };
++ return cache;
++}
++/** What app_state says, whether or not CHAT_AUTHORITY=chat. */
++function stored(table) { return flags().authority[table] === 'chat' ? 'chat' : 'live'; }
++/** Who writes the table now. */
++function authority(table) { return isRemote() && stored(table) === 'chat' ? 'chat' : 'live'; }
++function dualRead(table) { return isRemote() && flags().dual[table] === true; }
++
++function saveFlag(key, table, value) {
++ const cur = readJson(key);
++ if (value === undefined || value === null || value === false || value === 'live') delete cur[table];
++ else cur[table] = value;
++ db.setState(key, JSON.stringify(cur));
++ cache.at = 0;
++}
++function setDualRead(table, on) {
++ if (!TABLES[table]) throw httpError(404, `${table} is not a staged table`);
++ saveFlag(DUAL_READ_KEY, table, on ? true : undefined);
++ return dualRead(table);
++}
++
++// ── Chat's bridge ───────────────────────────────────────────────
++
++/** One op on Chat's bridge → its result; throws with Chat's refusal. A lost answer is retried once with the same key. */
++async function callChat(op, args, { key = null, timeoutMs = 10000 } = {}) {
++ const principal = require('../net/network-principal');
++ const body = JSON.stringify({ boot: BOOT, ops: [{ seq: 1, op, args, ...(key ? { key } : {}) }] });
++ let lastErr = null;
++ for (let attempt = 0; attempt < 2; attempt++) {
++ let res;
++ try {
++ res = await fetch(`${CHAT_URL}/internal/live/calls`, {
++ method: 'POST',
++ headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...(await principal.serviceHeaders(AUDIENCE)) },
++ body,
++ signal: AbortSignal.timeout(timeoutMs),
++ });
++ } catch (err) {
++ lastErr = httpError(503, `OpenVibe.Chat unreachable: ${err.message}`);
++ if (!key) break; // only an idempotent op is sent twice
++ continue;
++ }
++ if (res.status === 401 && attempt === 0) { principal.invalidate(AUDIENCE); continue; }
++ if (!res.ok) throw httpError(502, `OpenVibe.Chat answered ${res.status}`);
++ const out = await res.json();
++ const r = out && Array.isArray(out.results) ? out.results[0] : null;
++ if (!r) throw httpError(502, 'OpenVibe.Chat gave no result');
++ if (!r.ok) throw httpError(409, `OpenVibe.Chat refused ${op === 'db' ? args[0] : op}: ${r.error}`);
++ return r.result;
++ }
++ throw lastErr;
++}
++
++// ── Writes ──────────────────────────────────────────────────────
++
++const handoffs = new Map(); // table → promise of the running handoff
++
++/**
++ * Live's write to a staged table, wherever the table is written: `op` is the database.js function
++ * (OPS), `args` its arguments, the answer what that function returns. Await it.
++ */
++async function write(op, ...args) {
++ const table = OPS[op];
++ if (!table) throw new Error(`${op} is not a staged-table write`);
++ const running = handoffs.get(table);
++ if (running) await running.catch(() => {});
++ if (authority(table) === 'live') return db[op](...args);
++ const r = await callChat('db', [op, ...args], { key: `staged:${crypto.randomUUID()}` });
++ // Live's copy follows at once (Chat's mirror sends the same rows again later).
++ if (r && Array.isArray(r.mirror) && r.mirror.length) {
++ const out = require('./live-context-routes').applyMirror(r.mirror);
++ if (out.skipped.length) console.warn(`[ChatTables] ${op}: ${out.skipped.length} row(s) of Chat's answer not applied here:`, JSON.stringify(out.skipped.slice(0, 2)));
++ }
++ return r ? r.value : null;
++}
++
++/** Whether Live takes Chat's mirror rows for this table (Chat writes it, or it is being handed back). */
++function acceptsMirror(table) {
++ return !!TABLES[table] && isRemote() && (stored(table) === 'chat' || handoffs.has(table));
++}
++
++// ── The handoff ─────────────────────────────────────────────────
++
++/**
++ * Hand `table` to `target`. `force` (back to 'live' only): Live takes the table without asking Chat —
++ * for when Chat is down; first send Chat's queued changes (OpenVibe.Chat scripts/mirror-flush.js) and
++ * set Chat's side (scripts/table-authority.js set <table> live --force), docs/staged-tables-cutover.md.
++ */
++async function setAuthority(table, target, { by = 'operator', force = false } = {}) {
++ if (!TABLES[table]) throw httpError(404, `${table} is not a staged table`);
++ if (target !== 'live' && target !== 'chat') throw httpError(400, 'authority is "live" or "chat"');
++ if (force && target !== 'live') throw httpError(400, 'force is only for taking a table back');
++ if (!isRemote()) throw httpError(409, 'CHAT_AUTHORITY is not "chat": Live runs chat and writes every staged table itself');
++ if (handoffs.has(table)) throw httpError(409, `a handoff of ${table} is already running`);
++ const sync = require('./chat-tables-sync');
++ let finish;
++ handoffs.set(table, new Promise((resolve) => { finish = resolve; }));
++ const before = stored(table);
++ const started = Date.now();
++ try {
++ if (target === 'chat') {
++ await sync.drain(table); // 1. Live's changes are in Chat's copy
++ const r = await callChat('setTableAuthority', [table, 'chat'], { timeoutMs: 20000 }); // 2. Chat writes it
++ if (!r || r.authority !== 'chat') throw httpError(502, `OpenVibe.Chat did not take ${table}`);
++ saveFlag(AUTHORITY_KEY, table, 'chat'); // 3. Live stops writing it
++ sync.setCapture(table, false);
++ } else {
++ if (force) {
++ console.warn(`[ChatTables] ${table}: taken back without asking OpenVibe.Chat (force, ${by})`);
++ } else {
++ const r = await callChat('setTableAuthority', [table, 'live'], { timeoutMs: 30000 }); // 1–2. Chat's changes are here; Chat stops
++ if (!r || r.authority !== 'live') throw httpError(502, `OpenVibe.Chat did not give ${table} back`);
++ }
++ sync.setCapture(table, true); // 3. Live writes it again
++ saveFlag(AUTHORITY_KEY, table, undefined);
++ }
++ console.log(`[ChatTables] ${table}: ${before} → ${target} (${by}${force ? ', forced' : ''}, ${Date.now() - started} ms)`);
++ return { table, before, authority: target };
++ } catch (err) {
++ console.warn(`[ChatTables] ${table}: handoff to ${target} failed, still ${stored(table)}: ${err.message}`);
++ throw err;
++ } finally {
++ handoffs.delete(table);
++ finish();
++ }
++}
++
++// ── Status ──────────────────────────────────────────────────────
++
++/** Every staged table: who writes it (here and, when asked and reachable, in Chat), flags, the relay, the dual-read counts. */
++async function status({ askChat = true } = {}) {
++ const sync = require('./chat-tables-sync');
++ let chat = null;
++ // (a restore drill asks no other service anything)
++ if (askChat && isRemote() && !require('../drill').enabled) {
++ try { chat = await callChat('tableAuthority', [], { timeoutMs: 3000 }); } catch (err) { chat = { error: err.message }; }
++ }
++ const tables = {};
++ for (const t of Object.keys(TABLES)) {
++ tables[t] = {
++ authority: authority(t),
++ stored: stored(t),
++ chat_authority: chat && !chat.error ? chat[t] || null : null,
++ dual_read: dualRead(t),
++ handoff_running: handoffs.has(t),
++ outbox_pending: sync.pending(t),
++ dual_read_stats: sync.dualReadStats(t),
++ };
++ if (tables[t].chat_authority && tables[t].chat_authority !== tables[t].authority) tables[t].disagree = true;
++ }
++ return { chat_authority: isRemote() ? 'chat' : 'live', chat_error: chat && chat.error ? chat.error : null, relay: sync.relayStats(), tables };
++}
++
++/** Server start (CHAT_AUTHORITY=chat only): capture, dual read, relay. */
++function init() {
++ if (!isRemote()) return false;
++ require('./chat-tables-sync').start();
++ return true;
++}
++
++module.exports = {
++ TABLES, OPS, AUTHORITY_KEY, DUAL_READ_KEY,
++ authority, stored, dualRead, setDualRead, acceptsMirror,
++ write, setAuthority, status, init, callChat,
++ _resetCache() { cache.at = 0; },
++};
ddiff --git a/server/chat/live-context-routes.js b/server/chat/live-context-routes.js
iindex a9b402fd..bf831f21 100644
---- a/server/chat/live-context-routes.js
++++ b/server/chat/live-context-routes.js
@@@ -38,6 +38,7 @@ const config = require('../config');
const permissions = require('../auth/permissions');
const { guard } = require('../net/service-guard');
const { isRemote } = require('./chat-authority');
++const chatTables = require('./chat-tables');
const contextRouter = express.Router();
const effectsRouter = express.Router();
@@@ -348,8 +349,8 @@ effectsRouter.post('/channel-settings', (req, res) => {
const fields = {};
if (req.body.fields && req.body.fields.slow_mode_seconds !== undefined) fields.slow_mode_seconds = Math.max(0, int(req.body.fields.slow_mode_seconds));
if (!Object.keys(fields).length) return fail(res, 400, 'no chat-settable fields');
-- db.upsertChannelModerationSettings(channelId, fields);
-- res.json({ ok: true });
++ chatTables.write('upsertChannelModerationSettings', channelId, fields)
++ .then(() => res.json({ ok: true }), (err) => fail(res, err.status || 500, err.message));
});
// Donation / goal alert sounds: the streamer's own channel only, files in the shared sounds dir.
@@@ -364,8 +365,8 @@ effectsRouter.post('/alert-sound', (req, res) => {
try { inside = path.dirname(fs.realpathSync(url)) === fs.realpathSync(path.resolve(config.sounds.path)); } catch { inside = false; }
if (!inside) return fail(res, 400, 'alert sounds live in the sounds directory');
}
-- db.setChannelAlertSound(channelId, req.body.kind === 'goal' ? 'goal' : 'donation', url, url ? String(req.body.mime || 'audio/mpeg') : null);
-- res.json({ ok: true });
++ chatTables.write('setChannelAlertSound', channelId, req.body.kind === 'goal' ? 'goal' : 'donation', url, url ? String(req.body.mime || 'audio/mpeg') : null)
++ .then(() => res.json({ ok: true }), (err) => fail(res, err.status || 500, err.message));
});
effectsRouter.post('/ensure-channel', (req, res) => {
@@@ -607,6 +608,10 @@ effectsRouter.post('/asset-sync', (req, res) => {
// ── The read mirror of Chat's tables ──────────────────────────────
// Same tables, same ids. Upserts set only the columns Live's table has (Chat's *subject_id
// columns are skipped; Live-only columns such as channel_sounds.media_asset_id are kept).
++// A staged table (roadmap C-04, chat-tables.js) is mirrored only while Chat writes it or hands it
++// back; while Live writes it, Chat's rows for it are refused. Its rows REPLACE: the authority's row
++// wins over whatever holds its key or one of its unique columns here (both copies have the same
++// columns, so nothing of Live's is lost).
const MIRROR_TABLES = {
chat_messages: ['id'],
dm_conversations: ['id'],
@@@ -637,12 +642,16 @@ function applyMirror(changes) {
try {
d.transaction(() => {
for (const c of changes) {
-- const pk = MIRROR_TABLES[c && c.table];
-- if (!pk) { skipped.push({ table: c && c.table, reason: 'not a mirrored table' }); continue; }
++ const staged = !!(c && chatTables.TABLES[c.table]);
++ const pk = MIRROR_TABLES[c && c.table] || (staged && chatTables.acceptsMirror(c.table) ? chatTables.TABLES[c.table] : null);
++ if (!pk) { skipped.push({ table: c && c.table, reason: staged ? 'Live writes this table (chat_table_authority live)' : 'not a mirrored table' }); continue; }
const have = liveColumns(c.table);
try {
if (c.op === 'delete') {
d.prepare(`DELETE FROM ${c.table} WHERE ${pk.map((k) => `${k} = ?`).join(' AND ')}`).run(...pk.map((k) => c.pk[k]));
++ } else if (c.op === 'upsert' && c.row && staged) {
++ const cols = Object.keys(c.row).filter((k) => have.has(k) && /^[a-z_]+$/.test(k));
++ d.prepare(`INSERT OR REPLACE INTO ${c.table} (${cols.join(', ')}) VALUES (${cols.map(() => '?').join(', ')})`).run(...cols.map((k) => c.row[k]));
} else if (c.op === 'upsert' && c.row) {
const cols = Object.keys(c.row).filter((k) => have.has(k) && /^[a-z_]+$/.test(k));
const upd = cols.filter((k) => !pk.includes(k));
ddiff --git a/server/chat/sounds-routes.js b/server/chat/sounds-routes.js
iindex 825127e2..1c4ab3be 100644
---- a/server/chat/sounds-routes.js
++++ b/server/chat/sounds-routes.js
@@@ -428,7 +428,7 @@ router.post('/alert/:kind', requireAuth, soundUpload.single('sound'), async (req
const ext = path.extname(finalPath).toLowerCase();
const mime = EXT_TO_MIME[ext] || 'audio/mpeg';
-- db.setChannelAlertSound(channel.id, kind, finalPath, mime);
++ await require('./chat-tables').write('setChannelAlertSound', channel.id, kind, finalPath, mime);
res.json({ set: true, kind, url: `/api/sounds/file/${path.basename(finalPath)}` });
} catch (err) {
if (req.file) fs.unlink(req.file.path, () => {});
@@@ -437,7 +437,7 @@ router.post('/alert/:kind', requireAuth, soundUpload.single('sound'), async (req
});
// Clear an alert sound.
--router.delete('/alert/:kind', requireAuth, (req, res) => {
++router.delete('/alert/:kind', requireAuth, async (req, res) => {
const kind = _alertKind(req.params.kind);
try {
const channel = db.getChannelByUserId(req.user.id);
@@@ -445,7 +445,7 @@ router.delete('/alert/:kind', requireAuth, (req, res) => {
const prev = db.getChannelAlertSoundsByUser(req.user.id) || {};
const prevPath = kind === 'goal' ? prev.goal_sound_url : prev.donation_sound_url;
if (prevPath && fs.existsSync(prevPath)) { try { fs.unlinkSync(prevPath); } catch { /* */ } }
-- db.setChannelAlertSound(channel.id, kind, null, null);
++ await require('./chat-tables').write('setChannelAlertSound', channel.id, kind, null, null);
}
res.json({ set: false, kind });
} catch (err) {
ddiff --git a/server/db/database.js b/server/db/database.js
iindex afbcc0ad..ce4e8261 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -5796,6 +5796,11 @@ function updateEmote(id, { code, size }) {
return run(`UPDATE emotes SET ${sets.join(', ')} WHERE id = ?`, params);
}
++// The emote's copy on OpenVibe.Media (media-proxy/asset-sync.js), through chat-tables.write().
++function setEmoteMedia(id, mediaUrl, mediaAssetId) {
++ return run('UPDATE emotes SET media_url = ?, media_asset_id = ? WHERE id = ?', [mediaUrl || null, mediaAssetId || null, id]);
++}
++
function getEmoteByCode(code, userId) {
// Check channel emotes first, then global
return get(
@@@ -7739,7 +7744,7 @@ module.exports = {
// Bans
isUserBanned, isIpBanned, getIpBan, invalidateIpBanCache, forgiveBan,
// Emotes
-- createEmote, getEmoteById, getEmotesByUser, getGlobalEmotes, getChannelEmotes, updateEmote,
++ createEmote, getEmoteById, getEmotesByUser, getGlobalEmotes, getChannelEmotes, updateEmote, setEmoteMedia,
deleteEmote, getEmoteByCode, countUserEmotes, countChannelEmotes, getChannelEmoteByCode,
createChannelSound, setChannelSoundEmote, getChannelSounds, getChannelSoundByCommand, getChannelSoundById, renameChannelSoundCommand, updateChannelSoundEmoteRefs,
countChannelSounds, countChannelSoundsByUploader, deleteChannelSound,
ddiff --git a/server/emotes/routes.js b/server/emotes/routes.js
iindex 7439fbe0..b7e31c80 100644
---- a/server/emotes/routes.js
++++ b/server/emotes/routes.js
@@@ -20,6 +20,8 @@ const db = require('../db/database');
const { requireAuth, optionalAuth } = require('../auth/auth');
const permissions = require('../auth/permissions');
const config = require('../config');
++// Emote rows are written wherever the table's authority is (Live or OpenVibe.Chat, chat-tables.js).
++const chatTables = require('../chat/chat-tables');
const router = express.Router();
@@@ -276,7 +278,7 @@ router.get('/mine', requireAuth, (req, res) => {
});
// ── Upload a custom emote ────────────────────────────────────
--router.post('/', requireAuth, emoteUpload.single('image'), (req, res) => {
++router.post('/', requireAuth, emoteUpload.single('image'), async (req, res) => {
try {
if (!req.file) return res.status(400).json({ error: 'No image file uploaded' });
@@@ -361,7 +363,7 @@ router.post('/', requireAuth, emoteUpload.single('image'), (req, res) => {
}
const size = Math.min(sizeMax, Math.max(sizeMin, parseInt(req.body.size) || 100));
-- const result = db.createEmote({
++ const result = await chatTables.write('createEmote', {
user_id: req.user.id,
code,
url: req.file.path,
@@@ -392,6 +394,8 @@ router.post('/', requireAuth, emoteUpload.single('image'), (req, res) => {
console.error('[Emotes] Upload error:', err);
if (req.file && fs.existsSync(req.file.path)) { try { fs.unlinkSync(req.file.path); } catch {} }
if (err && /UNIQUE constraint/i.test(err.message || '')) {
++ // (`code` is the try block's; an async handler must not throw from here)
++ const code = String((req.body && req.body.code) || '').trim();
return res.status(409).json({ error: `This channel already has an emote named "${code}" — pick a different code for this channel.` });
}
res.status(500).json({ error: 'Failed to upload emote' });
@@@ -413,7 +417,7 @@ router.use((err, req, res, next) => {
// ── Delete an emote ──────────────────────────────────────────
// ── Edit an emote in place (rename code / change display size) ──
// Same permission model as delete: uploader, admin, or channel mod/owner.
--router.patch('/:id', requireAuth, (req, res) => {
++router.patch('/:id', requireAuth, async (req, res) => {
try {
const emote = db.getEmoteById(req.params.id);
if (!emote) return res.status(404).json({ error: 'Emote not found' });
@@@ -456,7 +460,7 @@ router.patch('/:id', requireAuth, (req, res) => {
return res.status(400).json({ error: 'Nothing to change' });
}
-- db.updateEmote(emote.id, patch);
++ await chatTables.write('updateEmote', emote.id, patch);
// Sound commands attach this emote by code — carry the rename over.
if (patch.code && patch.code !== emote.code) {
try { db.updateChannelSoundEmoteRefs(emote.channel_owner_id || emote.user_id, emote.code, patch.code); } catch { /* */ }
@@@ -470,7 +474,7 @@ router.patch('/:id', requireAuth, (req, res) => {
}
});
--router.delete('/:id', requireAuth, (req, res) => {
++router.delete('/:id', requireAuth, async (req, res) => {
try {
const emote = db.getEmoteById(req.params.id);
if (!emote) return res.status(404).json({ error: 'Emote not found' });
@@@ -485,12 +489,11 @@ router.delete('/:id', requireAuth, (req, res) => {
return res.status(403).json({ error: 'Not your emote' });
}
-- // Delete file
++ // The row first: if that write fails, the file is still there.
++ await chatTables.write('deleteEmote', emote.id);
if (emote.url && fs.existsSync(emote.url)) {
fs.unlinkSync(emote.url);
}
--
-- db.deleteEmote(req.params.id);
try { require('../media-proxy/asset-sync').removeAsset(emote.media_asset_id); } catch { /* */ }
if (emote.channel_owner_id) {
try { require('../chat/chat-server').broadcastToOwnerStreams(emote.channel_owner_id, { type: 'emotes-updated' }); } catch { /* */ }
ddiff --git a/server/index.js b/server/index.js
iindex 71e9a92f..741963b8 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -1128,6 +1128,9 @@ async function start() {
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();
++ // The staged chat tables (C-04): with CHAT_AUTHORITY=chat, Live's changes to the ones it writes
++ // reach OpenVibe.Chat, and the dual read compares (server/chat/chat-tables.js). Nothing otherwise.
++ require('./chat/chat-tables').init();
// 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/internal/routes.js b/server/internal/routes.js
iindex a774c113..3ef11d33 100644
---- a/server/internal/routes.js
++++ b/server/internal/routes.js
@@@ -23,6 +23,41 @@ function notifyChat(userId) {
try { const cs = require('../chat/chat-server'); if (cs.remote) cs.userChanged(userId); } catch { /* non-critical */ }
}
++// ── The staged chat tables (roadmap C-04, server/chat/chat-tables.js) ─────────────
++// GET /internal/chat-tables who writes each table (here and in OpenVibe.Chat), the relay to
++// Chat, the dual-read counts
++// POST /internal/chat-tables/:table { dual_read?: bool, reset_counters?: bool, authority?: 'live'|'chat',
++// by?: 'who', force?: true (back to 'live' with Chat down) } — the
++// flag, the counters, then the handoff
++router.get('/chat-tables', async (req, res) => {
++ try { res.json(await require('../chat/chat-tables').status()); } catch (err) { res.status(500).json({ error: err.message }); }
++});
++// POST /internal/chat-tables/relay { paused: bool } — hold the relay to Chat (the capture goes on) for an import
++router.post('/chat-tables/relay', (req, res) => {
++ try {
++ const sync = require('../chat/chat-tables-sync');
++ if (req.body && req.body.paused !== undefined) sync.setPaused(req.body.paused === true || req.body.paused === 'true');
++ res.json({ ok: true, relay: sync.relayStats() });
++ } catch (err) { res.status(500).json({ error: err.message }); }
++});
++router.post('/chat-tables/:table', async (req, res) => {
++ const chatTables = require('../chat/chat-tables');
++ const table = String(req.params.table);
++ const b = req.body || {};
++ try {
++ if (!chatTables.TABLES[table]) return res.status(404).json({ error: `${table} is not a staged table` });
++ if (b.dual_read !== undefined) chatTables.setDualRead(table, b.dual_read === true || b.dual_read === 'true' || b.dual_read === 1);
++ if (b.reset_counters) require('../chat/chat-tables-sync').resetDualReadStats(table);
++ let handoff = null;
++ if (b.authority !== undefined) handoff = await chatTables.setAuthority(table, String(b.authority), { by: String(b.by || 'operator').slice(0, 80), force: b.force === true });
++ const st = await chatTables.status();
++ res.json({ ok: true, handoff, table: st.tables[table], chat_error: st.chat_error });
++ } catch (err) {
++ const st = await chatTables.status({ askChat: false }).catch(() => null);
++ res.status(err.status || 500).json({ ok: false, error: err.message, table: st ? st.tables[table] : null });
++ }
++});
++
// Footer site copy is written by OpenVibe.AI for the Network directly (network.site_copy); the
// /internal/ai/site-copy fallback that used to live here was retired on 2026-09-23.
ddiff --git a/server/media-proxy/asset-sync.js b/server/media-proxy/asset-sync.js
iindex 39e29527..087c89ca 100644
---- a/server/media-proxy/asset-sync.js
++++ b/server/media-proxy/asset-sync.js
@@@ -76,7 +76,8 @@ async function syncAll() {
channel_username: uname(e.channel_owner_id || e.user_id),
meta: { animated: !!e.animated, is_global: !!e.is_global },
});
-- if (asset) { db.run('UPDATE emotes SET media_url = ?, media_asset_id = ? WHERE id = ?', [asset.url, asset.id, e.id]); synced++; }
++ // emotes is a staged chat table: written where its authority is (chat/chat-tables.js).
++ if (asset) { await require('../chat/chat-tables').write('setEmoteMedia', e.id, asset.url, asset.id); synced++; }
} catch (err) { failed++; if (failed <= 3) console.warn('[AssetSync] emote', e.code, err.message); }
}
ddiff --git a/test/chat-tables-writers.test.js b/test/chat-tables-writers.test.js
nnew file mode 100644
iindex 00000000..0339a21c
---- /dev/null
++++ b/test/chat-tables-writers.test.js
@@@ -0,0 +1,63 @@
++'use strict';
++
++// Every write to a staged chat table (channel_moderators, channel_moderation_settings, emotes,
++// user_tags, chat_ai_summaries, chat_timeline_events — roadmap C-04) goes through
++// server/chat/chat-tables.js write(), which sends it to OpenVibe.Chat once Chat writes the table.
++// A writer that called database.js directly, or wrote SQL of its own, would keep writing Live's copy
++// after the handoff — a second writer next to Chat. So:
++// - no server/ or scripts/ code calls one of database.js's staged-table writers directly (chat-tables
++// calls them as db[op] while Live writes the table);
++// - SQL that writes a staged table appears only in database.js, inside one of those writers;
++// - chat-tables.OPS lists exactly database.js's writers.
++// Schema upkeep (CREATE/ALTER, the one-time emotes_new rebuild) is not a write. Comment lines never count.
++
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++
++const ROOT = path.join(__dirname, '..');
++const { OPS, TABLES } = require('../server/chat/chat-tables');
++const T = Object.keys(TABLES).join('|');
++const WRITE_SQL = new RegExp(`\\b(INSERT(?:\\s+OR\\s+\\w+)?\\s+INTO|UPDATE|DELETE\\s+FROM|REPLACE\\s+INTO)\\s+(${T})\\b`, 'i');
++const DIRECT_CALL = new RegExp(`\\bdb\\.(${Object.keys(OPS).join('|')})\\s*\\(`);
++const isCommentLine = (line) => /^\s*(\/\/|\*|\/\*)/.test(line);
++
++function walk(dir, out = []) {
++ for (const e of fs.readdirSync(dir, { withFileTypes: true })) {
++ const p = path.join(dir, e.name);
++ if (e.isDirectory()) { if (e.name !== 'node_modules') walk(p, out); }
++ else if (e.name.endsWith('.js')) out.push(p);
++ }
++ return out;
++}
++
++const problems = [];
++const writersInDb = new Set();
++for (const file of [...walk(path.join(ROOT, 'server')), ...walk(path.join(ROOT, 'scripts'))]) {
++ const rel = path.relative(ROOT, file);
++ const lines = fs.readFileSync(file, 'utf8').split('\n');
++ let fn = null;
++ lines.forEach((line, i) => {
++ const def = /^(?:async\s+)?function\s+(\w+)/.exec(line);
++ if (def) fn = def[1];
++ if (isCommentLine(line)) return;
++ if (DIRECT_CALL.test(line)) problems.push(`${rel}:${i + 1} calls database.js directly — use require('../chat/chat-tables').write(): ${line.trim().slice(0, 120)}`);
++ if (WRITE_SQL.test(line)) {
++ if (rel === path.join('server', 'db', 'database.js') && fn && OPS[fn]) { writersInDb.add(fn); return; }
++ problems.push(`${rel}:${i + 1} writes a staged chat table outside database.js's writers (${fn || 'top level'}): ${line.trim().slice(0, 120)}`);
++ }
++ });
++}
++
++try {
++ assert.deepStrictEqual(problems, [], `staged-table writes that bypass chat-tables.write():\n ${problems.join('\n ')}`);
++ // Every writer chat-tables knows exists in database.js and writes SQL there (setChannelAlertSound and
++ // upsertChannelModerationSettings write the same table; each has its own SQL).
++ const db = require('../server/db/database');
++ for (const op of Object.keys(OPS)) assert.strictEqual(typeof db[op], 'function', `database.js exports ${op}`);
++ assert.deepStrictEqual([...writersInDb].sort(), Object.keys(OPS).sort(), 'chat-tables.OPS lists exactly the functions that write the staged tables');
++ console.log(`chat-tables writers: ${Object.keys(OPS).length} writers, all through chat-tables.write()`);
++} catch (err) {
++ console.error(err.message);
++ process.exit(1);
++}
ddiff --git a/test/chat-tables.test.js b/test/chat-tables.test.js
nnew file mode 100644
iindex 00000000..cb493d70
---- /dev/null
++++ b/test/chat-tables.test.js
@@@ -0,0 +1,404 @@
++'use strict';
++
++// The staged chat tables (roadmap C-04; OpenVibe.Chat docs/staged-tables-cutover.md): the authority
++// per table (app_state, 'live' by default, and 'live' whatever it says without CHAT_AUTHORITY=chat);
++// at 'live' Live writes locally and its captured changes reach Chat's copy (ordered, newest state,
++// kept while Chat is down); the dual-read counter (matched, mismatched, inconclusive; Live's answer
++// never changes); the internal status and flip routes; the handoff to 'chat' (Live's queued
++// changes first, then Chat, then Live stops writing); Live's writers going to Chat when flipped —
++// through a real route — with Live's copy following Chat's answer, Chat's mirror accepted only for
++// a table Chat writes; a refused or unreachable Chat changes nothing; the flip back. Against stub
++// Network and Chat servers (the stub keeps its copy in SQLite and hashes slices like Chat does).
++
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const http = require('http');
++const crypto = require('crypto');
++const jwt = require('jsonwebtoken');
++const Database = require('better-sqlite3');
++const { serviceAuth } = require('openvibe-contracts');
++
++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-chattables-'));
++const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });
++fs.writeFileSync(path.join(tmp, 'network.pem'), keys.publicKey);
++process.env.DB_PATH = path.join(tmp, 'live.db');
++process.env.OV_NETWORK_PUBLIC_KEY = path.join(tmp, 'network.pem');
++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.INTERNAL_API_KEY = 'internal-test-key';
++process.env.CHAT_AUTHORITY = 'chat';
++const quiet = console.log;
++console.log = () => {};
++console.warn = () => {};
++
++const ISS = 'https://openvibe.network';
++const now = () => Math.floor(Date.now() / 1000);
++let jti = 0;
++const serviceToken = (cap, aud) => serviceAuth.signServiceToken({ iss: ISS, sub: 'svc:live', actor_type: 'service', aud: [aud], cap, iat: now(), exp: now() + 300, jti: `tok_test_${++jti}` }, keys.privateKey);
++
++const network = http.createServer((req, res) => {
++ let raw = '';
++ req.on('data', (c) => { raw += c; });
++ req.on('end', () => {
++ res.setHeader('Content-Type', 'application/json');
++ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: serviceToken(['chat.live_bridge.write'], new URLSearchParams(raw).get('audience')), expires_in: 300 }));
++ res.statusCode = 404; res.end('{}');
++ });
++});
++
++// ── Stub OpenVibe.Chat: the bridge ops of the staged tables, on a SQLite copy ──
++const TABLES = ['channel_moderators', 'channel_moderation_settings', 'emotes', 'user_tags', 'chat_ai_summaries', 'chat_timeline_events'];
++const PK = { channel_moderation_settings: 'channel_id' };
++const pkOf = (t) => PK[t] || 'id';
++const stub = { calls: [], down: false, refuseHandback: false, authority: Object.fromEntries(TABLES.map((t) => [t, 'live'])), copy: null };
++const hash = (cols, rows) => crypto.createHash('sha256').update(JSON.stringify(rows.map((r) => cols.map((c) => (r[c] === undefined ? null : r[c]))))).digest('hex');
++function stubOp(o) {
++ const c = stub.copy;
++ const [fn, ...a] = o.op === 'db' ? o.args : [];
++ const rowOf = (t, id) => c.prepare(`SELECT * FROM ${t} WHERE ${pkOf(t)} = ?`).get(id);
++ const answer = (value, t, ids = [], deleted = []) => ({ value, mirror: [...deleted.map((id) => ({ table: t, op: 'delete', pk: { [pkOf(t)]: id } })), ...ids.map((id) => ({ table: t, op: 'upsert', row: rowOf(t, id) }))] });
++ const plain = (r) => ({ changes: r.changes, lastInsertRowid: Number(r.lastInsertRowid) });
++ switch (o.op) {
++ case 'tableAuthority': return { ...stub.authority };
++ case 'setTableAuthority':
++ if (o.args[1] === 'live' && stub.refuseHandback) throw new Error(`1 change(s) to ${o.args[0]} not in Live yet`);
++ stub.authority[o.args[0]] = o.args[1];
++ return { table: o.args[0], authority: o.args[1], mirror_pending: 0 };
++ case 'stagedApply': {
++ const out = { applied: 0, skipped: [] };
++ for (const ch of o.args[0]) {
++ if (stub.authority[ch.table] !== 'live') { out.skipped.push({ table: ch.table, reason: 'Chat writes this table (table_authority chat)' }); continue; }
++ if (ch.op === 'delete') c.prepare(`DELETE FROM ${ch.table} WHERE ${pkOf(ch.table)} = ?`).run(ch.pk[pkOf(ch.table)]);
++ else { const cols = Object.keys(ch.row); c.prepare(`INSERT OR REPLACE INTO ${ch.table} (${cols.join(', ')}) VALUES (${cols.map(() => '?').join(', ')})`).run(...cols.map((k) => ch.row[k])); }
++ out.applied++;
++ }
++ return out;
++ }
++ case 'stagedSlice': {
++ const [t, where, cols] = o.args;
++ const ks = Object.keys(where);
++ const use = [...cols].sort();
++ const rows = c.prepare(`SELECT ${use.join(', ')} FROM ${t} WHERE ${ks.map((k) => `${k} IS ?`).join(' AND ') || '1'} ORDER BY ${pkOf(t)}`).all(...ks.map((k) => where[k]));
++ return { count: rows.length, hash: hash(use, rows), columns: use, rows: rows.length <= 50 ? rows : undefined };
++ }
++ case 'db': {
++ const table = { addChannelModerator: 'channel_moderators', removeChannelModerator: 'channel_moderators', upsertChannelModerationSettings: 'channel_moderation_settings', setChannelAlertSound: 'channel_moderation_settings', createEmote: 'emotes', updateEmote: 'emotes', deleteEmote: 'emotes', setEmoteMedia: 'emotes', upsertChatAiSummary: 'chat_ai_summaries', addChatTimelineEvents: 'chat_timeline_events' }[fn];
++ if (stub.authority[table] !== 'chat') throw new Error(`${table} is written by Live (table_authority live)`);
++ if (fn === 'addChannelModerator') {
++ const r = c.prepare('INSERT OR IGNORE INTO channel_moderators (channel_id, user_id, added_by) VALUES (?, ?, ?)').run(...a);
++ return answer(plain(r), table, [c.prepare('SELECT id FROM channel_moderators WHERE channel_id = ? AND user_id = ?').get(a[0], a[1]).id]);
++ }
++ if (fn === 'removeChannelModerator') {
++ const gone = c.prepare('SELECT id FROM channel_moderators WHERE channel_id = ? AND user_id = ?').all(a[0], a[1]).map((r) => r.id);
++ return answer(plain(c.prepare('DELETE FROM channel_moderators WHERE channel_id = ? AND user_id = ?').run(a[0], a[1])), table, [], gone);
++ }
++ if (fn === 'upsertChannelModerationSettings') {
++ c.prepare('INSERT OR IGNORE INTO channel_moderation_settings (channel_id) VALUES (?)').run(a[0]);
++ for (const [k, v] of Object.entries(a[1])) if (v !== undefined && /^[a-z_]+$/.test(k) && k !== 'slowmode_seconds') c.prepare(`UPDATE channel_moderation_settings SET ${k} = ? WHERE channel_id = ?`).run(typeof v === 'boolean' ? Number(v) : v, a[0]);
++ return answer(rowOf(table, a[0]), table, [a[0]]);
++ }
++ if (fn === 'createEmote') {
++ const e = a[0];
++ const r = c.prepare('INSERT INTO emotes (user_id, code, url, animated, channel_owner_id, size) VALUES (?, ?, ?, ?, ?, ?)').run(e.user_id, e.code, e.url, e.animated ? 1 : 0, e.channel_owner_id || null, e.size || 100);
++ return answer(plain(r), table, [Number(r.lastInsertRowid)]);
++ }
++ if (fn === 'deleteEmote') return answer(plain(c.prepare('DELETE FROM emotes WHERE id = ?').run(a[0])), table, [], [a[0]]);
++ if (fn === 'upsertChatAiSummary') {
++ const s = a[0];
++ c.prepare(`INSERT INTO chat_ai_summaries (scope, subject_id, window, overview) VALUES (?, ?, ?, ?)
++ ON CONFLICT(scope, subject_id, window) DO UPDATE SET overview = excluded.overview`).run(s.scope, s.subject_id || 0, s.window, s.overview || '');
++ return answer({ changes: 1, lastInsertRowid: 0 }, table, [c.prepare('SELECT id FROM chat_ai_summaries WHERE scope = ? AND subject_id = ? AND window = ?').get(s.scope, s.subject_id || 0, s.window).id]);
++ }
++ throw new Error(`stub has no ${fn}`);
++ }
++ default: throw new Error(`unknown op ${o.op}`);
++ }
++}
++const chat = http.createServer((req, res) => {
++ let raw = '';
++ req.on('data', (c) => { raw += c; });
++ req.on('end', () => {
++ if (stub.down) { req.socket.destroy(); return; }
++ res.setHeader('Content-Type', 'application/json');
++ const auth = String(req.headers.authorization || '');
++ const v = serviceAuth.verifyServiceToken(auth.slice(7), { publicKey: keys.publicKey, issuer: ISS, audience: 'openvibe.chat' });
++ if (!v.ok || !v.claims.cap.includes('chat.live_bridge.write')) { res.statusCode = 401; return res.end('{}'); }
++ if (req.url !== '/internal/live/calls') { res.statusCode = 404; return res.end('{}'); }
++ const body = JSON.parse(raw);
++ const results = body.ops.map((o) => {
++ stub.calls.push({ op: o.op, args: o.args, key: o.key });
++ try { return { seq: o.seq, ok: true, result: stubOp(o) }; } catch (err) { return { seq: o.seq, ok: false, error: err.message }; }
++ });
++ res.end(JSON.stringify({ ok: true, results }));
++ });
++});
++
++const listen = (s) => new Promise((r) => s.listen(0, '127.0.0.1', () => r(s.address().port)));
++const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
++
++(async () => {
++ process.env.OV_NETWORK_INTERNAL_URL = `http://127.0.0.1:${await listen(network)}`;
++ process.env.OV_CHAT_INTERNAL_URL = `http://127.0.0.1:${await listen(chat)}`;
++
++ const db = require('../server/db/database');
++ db.initDb();
++ require('../server/chat/tags').ensureTagTables();
++ for (const col of ['media_url TEXT', 'media_asset_id INTEGER']) { try { db.getDb().exec(`ALTER TABLE emotes ADD COLUMN ${col}`); } catch { /* present */ } }
++ const d = db.getDb();
++ // The stub's copy: the same tables, created from Live's own definitions (no foreign keys to follow).
++ stub.copy = new Database(':memory:');
++ stub.copy.pragma('foreign_keys = OFF');
++ for (const t of TABLES) stub.copy.exec(d.prepare("SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?").get(t).sql);
++ for (const r of d.prepare(`SELECT sql FROM sqlite_master WHERE type = 'index' AND sql IS NOT NULL AND tbl_name IN (${TABLES.map(() => '?').join(',')})`).all(...TABLES)) stub.copy.exec(r.sql);
++
++ const mkUser = (username, role = 'user') => {
++ const id = Number(db.createUser({ username, email: `${username}@example.test`, password_hash: '!x', display_name: username.toUpperCase(), stream_key: `key-${username}` }).lastInsertRowid);
++ d.prepare('UPDATE users SET role = ? WHERE id = ?').run(role, id);
++ return id;
++ };
++ const streamer = mkUser('streamer', 'streamer');
++ const mod = mkUser('moddy');
++ const viewer = mkUser('viewer');
++ d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (?, 'network', '601', 'usr_01J9SSSSSSSSSSSSSSSSSSSSSS')").run(streamer);
++ db.createChannel({ user_id: streamer, title: 'Streamer TV' });
++ const channel = db.getChannelByUserId(streamer);
++ db.createChannel({ user_id: viewer, title: 'Viewer TV' });
++ const channel2 = db.getChannelByUserId(viewer);
++ const streamerJwt = jwt.sign({ sub: '601', username: 'streamer', subject_id: 'usr_01J9SSSSSSSSSSSSSSSSSSSSSS' }, keys.privateKey, { algorithm: 'RS256', issuer: ISS, expiresIn: 600 });
++
++ const express = require('express');
++ const app = express();
++ app.use(express.json());
++ const routes = require('../server/chat/live-context-routes');
++ app.use('/internal/chat-effects', routes.effectsRouter);
++ app.use('/internal', require('../server/internal/routes'));
++ app.use('/api/channels', require('../server/admin/channel-mod-routes'));
++ const port = await listen(http.createServer(app));
++ const call = async (method, p, { body, headers = {} } = {}) => {
++ const res = await fetch(`http://127.0.0.1:${port}${p}`, { method, headers: { 'Content-Type': 'application/json', ...headers }, body: body ? JSON.stringify(body) : undefined });
++ const text = await res.text();
++ let json = null; try { json = JSON.parse(text); } catch { /* */ }
++ return { status: res.status, body: json, text };
++ };
++ const KEY = { 'X-Internal-Key': 'internal-test-key' };
++ const internal = (method, p, body) => call(method, `/internal${p}`, { body, headers: KEY });
++ const MIRROR = serviceToken(['live.chat_mirror.write'], 'openvibe.live');
++ const mirror = (changes) => call('POST', '/internal/chat-effects/mirror', { body: { changes }, headers: { Authorization: `Bearer ${MIRROR}` } });
++
++ const chatTables = require('../server/chat/chat-tables');
++ const sync = require('../server/chat/chat-tables-sync');
++ const outbox = (t) => d.prepare('SELECT tbl, op, pk FROM chat_staged_outbox WHERE tbl = ? ORDER BY seq').all(t);
++ const stats = (t) => sync.dualReadStats(t);
++ const stubRows = (t) => stub.copy.prepare(`SELECT * FROM ${t} ORDER BY ${pkOf(t)}`).all();
++
++ let exit = 0;
++ try {
++ // 1. Deploying changes nothing: every table 'live', writes local, and without CHAT_AUTHORITY=chat
++ // a stored 'chat' does not count.
++ for (const t of TABLES) assert.strictEqual(chatTables.authority(t), 'live', t);
++ await chatTables.write('addChannelModerator', channel.id, mod, streamer);
++ assert.ok(db.isChannelModerator(mod, channel.id));
++ db.setState(chatTables.AUTHORITY_KEY, JSON.stringify({ emotes: 'chat' }));
++ chatTables._resetCache();
++ assert.strictEqual(chatTables.authority('emotes'), 'chat');
++ process.env.CHAT_AUTHORITY = '';
++ assert.strictEqual(chatTables.authority('emotes'), 'live', 'Live runs chat itself: it writes everything');
++ assert.strictEqual(chatTables.init(), false, 'and starts nothing');
++ process.env.CHAT_AUTHORITY = 'chat';
++ db.setState(chatTables.AUTHORITY_KEY, '{}');
++ chatTables._resetCache();
++ await assert.rejects(chatTables.write('grantUserTag', viewer, 'x'), /not a staged-table write/, 'user_tags has no writer in Live');
++
++ // 2. init: capture, dual read, relay (the relay's timer is stopped; the test flushes by hand).
++ assert.strictEqual(chatTables.init(), true);
++ require('../server/utils/jobs').stopAll();
++ assert.strictEqual(sync.pending(), 0, 'what was written before init is the import’s business');
++
++ // 3. At 'live': every change to a staged table is captured, whoever writes it.
++ await chatTables.write('upsertChannelModerationSettings', channel.id, { slow_mode_seconds: 5 });
++ await chatTables.write('upsertChannelModerationSettings', channel.id, { emote_scale: 150 });
++ const e1 = Number((await chatTables.write('createEmote', { user_id: streamer, code: 'pog', url: '/data/emotes/pog.png', channel_owner_id: streamer })).lastInsertRowid);
++ await chatTables.write('removeChannelModerator', channel.id, mod);
++ await chatTables.write('upsertChatAiSummary', { scope: 'global', subject_id: 0, window: 'global', overview: 'busy' });
++ await chatTables.write('addChatTimelineEvents', 'global', 0, [{ ts: '2026-09-25 10:00:00', label: 'raid' }]);
++ d.prepare("INSERT INTO user_tags (user_id, tag_id, source) VALUES (?, 'legacy', 'migration')").run(viewer);
++ assert.deepStrictEqual(outbox('channel_moderation_settings').map((r) => r.op), ['upsert', 'upsert'], 'the insert and the update');
++ assert.deepStrictEqual(outbox('channel_moderators').map((r) => [r.op, JSON.parse(r.pk).id > 0]), [['delete', true]]);
++ assert.strictEqual(outbox('user_tags').length, 1, 'a raw write is captured too');
++
++ // 4. The relay: newest state per row, in order; kept while Chat is down; acknowledged rows go.
++ stub.down = true;
++ let r = await sync.flush();
++ assert.ok(r.pending >= 6 && /unreachable/.test(r.error), JSON.stringify(r));
++ stub.down = false;
++ stub.calls.length = 0;
++ r = await sync.flush();
++ assert.deepStrictEqual([r.pending, r.error], [0, null]);
++ const applied = stub.calls.filter((c) => c.op === 'stagedApply').flatMap((c) => c.args[0]);
++ const cms = applied.filter((c) => c.table === 'channel_moderation_settings');
++ assert.strictEqual(cms.length, 1, 'two changes to one row: one upsert');
++ assert.deepStrictEqual([cms[0].row.slow_mode_seconds, cms[0].row.emote_scale], [5, 150]);
++ assert.deepStrictEqual(applied.find((c) => c.table === 'channel_moderators'), { table: 'channel_moderators', op: 'delete', pk: { id: 1 } });
++ assert.strictEqual(stubRows('emotes')[0].code, 'pog');
++ assert.strictEqual(stubRows('user_tags').length, 1);
++ // Paused for an import: the capture goes on, nothing is sent; resumed, everything since goes.
++ assert.strictEqual((await internal('POST', '/chat-tables/relay', { paused: true })).body.relay.paused, true);
++ await chatTables.write('addChatTimelineEvents', 'global', 0, [{ ts: '2026-09-25 10:05:00', label: 'after the pause' }]);
++ r = await sync.flush();
++ assert.deepStrictEqual([r.error, r.pending], ['relay paused', 1]);
++ assert.strictEqual(stubRows('chat_timeline_events').length, 1);
++ assert.strictEqual((await internal('POST', '/chat-tables/relay', { paused: false })).body.relay.paused, false);
++ assert.strictEqual((await sync.flush()).pending, 0);
++ assert.strictEqual(stubRows('chat_timeline_events').length, 2);
++
++ // 5. The dual read: off by default; on, a read is compared in the background; Live's answer never changes.
++ const liveAnswer = db.getChannelModerationSettings(channel.id);
++ await sleep(50);
++ assert.strictEqual(stats('channel_moderation_settings').compared, 0, 'off: nothing compared');
++ assert.strictEqual((await internal('POST', '/chat-tables/channel_moderation_settings', { dual_read: true })).body.table.dual_read, true);
++ assert.deepStrictEqual(db.getChannelModerationSettings(channel.id), liveAnswer);
++ await sleep(150);
++ assert.deepStrictEqual([stats('channel_moderation_settings').compared, stats('channel_moderation_settings').matched], [1, 1]);
++ // Chat's copy differs for another channel: counted, sampled, logged; Live answers from its own row.
++ await chatTables.write('upsertChannelModerationSettings', channel2.id, { slow_mode_seconds: 30 });
++ await sync.flush();
++ stub.copy.prepare('UPDATE channel_moderation_settings SET slow_mode_seconds = 31 WHERE channel_id = ?').run(channel2.id);
++ assert.strictEqual(db.getChannelModerationSettings(channel2.id).slow_mode_seconds, 30);
++ await sleep(150);
++ let st = stats('channel_moderation_settings');
++ assert.deepStrictEqual([st.compared, st.mismatched], [2, 1]);
++ assert.deepStrictEqual(st.last_mismatch.where, { channel_id: channel2.id });
++ assert.deepStrictEqual(st.last_mismatch.differ, [JSON.stringify([channel2.id])]);
++ // A change still on its way to Chat: no verdict.
++ await internal('POST', '/chat-tables/emotes', { dual_read: true });
++ await chatTables.write('updateEmote', e1, { size: 120 });
++ db.getEmoteById(e1);
++ await sleep(150);
++ assert.deepStrictEqual([stats('emotes').inconclusive, stats('emotes').mismatched], [1, 0]);
++ await sync.flush();
++ // The same slice is compared at most once a minute.
++ db.getChannelModerationSettings(channel2.id);
++ await sleep(100);
++ assert.strictEqual(stats('channel_moderation_settings').compared, 2);
++ assert.strictEqual((await internal('POST', '/chat-tables/channel_moderation_settings', { reset_counters: true })).body.table.dual_read_stats.compared, 0);
++
++ // 6. The status route: loopback + internal key; both sides' view of every table.
++ assert.strictEqual((await call('GET', '/internal/chat-tables')).status, 403);
++ assert.strictEqual((await call('GET', '/internal/chat-tables', { headers: { ...KEY, 'X-Forwarded-For': '203.0.113.1' } })).status, 403);
++ let s = (await internal('GET', '/chat-tables')).body;
++ assert.strictEqual(s.chat_authority, 'chat');
++ assert.deepStrictEqual([s.tables.emotes.authority, s.tables.emotes.chat_authority, s.tables.emotes.dual_read, s.tables.emotes.outbox_pending], ['live', 'live', true, 0]);
++ assert.strictEqual((await internal('POST', '/chat-tables/chat_messages', { authority: 'chat' })).status, 404);
++
++ // 7. While Live writes a table, Chat's mirror rows for it are refused.
++ let m = await mirror([{ table: 'channel_moderators', op: 'upsert', row: { id: 900, channel_id: channel.id, user_id: viewer, added_by: streamer } }]);
++ assert.deepStrictEqual([m.body.applied, m.body.skipped[0].reason], [0, 'Live writes this table (chat_table_authority live)']);
++
++ // 8. Handoff to 'chat': Live's queued change reaches Chat first, then Chat takes it, then Live stops.
++ await chatTables.write('addChannelModerator', channel.id, viewer, streamer);
++ assert.strictEqual(outbox('channel_moderators').length, 1);
++ stub.calls.length = 0;
++ let f = await internal('POST', '/chat-tables/channel_moderators', { authority: 'chat', by: 'test' });
++ assert.strictEqual(f.status, 200, f.text);
++ // (chat-remote's own traffic — invalidations, forwarded moderation-log rows — is not this.)
++ const tableOps = () => stub.calls.filter((c) => ['stagedApply', 'setTableAuthority', 'tableAuthority'].includes(c.op) || (c.op === 'db' && c.args[0] in chatTables.OPS));
++ assert.deepStrictEqual(tableOps().map((c) => c.op), ['stagedApply', 'setTableAuthority', 'tableAuthority']);
++ assert.deepStrictEqual([f.body.handoff.before, f.body.handoff.authority, f.body.table.authority, f.body.table.chat_authority], ['live', 'chat', 'chat', 'chat']);
++ assert.strictEqual(stubRows('channel_moderators').length, 1, 'Chat has the row Live wrote last');
++ d.prepare('DELETE FROM channel_moderators WHERE user_id = ?').run(viewer);
++ assert.strictEqual(outbox('channel_moderators').length, 0, 'Live’s copy is the mirror now: nothing captured');
++ m = await mirror([{ table: 'channel_moderators', op: 'upsert', row: stubRows('channel_moderators')[0] }]);
++ assert.strictEqual(m.body.applied, 1, 'and Chat’s mirror is taken');
++ assert.ok(db.isChannelModerator(viewer, channel.id));
++
++ // 9. Live's writers go to Chat: the real route, Chat's id, Live's copy right away, nothing written locally first.
++ stub.calls.length = 0;
++ let res = await call('POST', `/api/channels/${channel.id}/mods`, { body: { username: 'moddy' }, headers: { Authorization: `Bearer ${streamerJwt}` } });
++ assert.strictEqual(res.status, 200, res.text);
++ const dbCall = tableOps().find((c) => c.op === 'db');
++ assert.deepStrictEqual(dbCall.args, ['addChannelModerator', channel.id, mod, streamer]);
++ assert.match(dbCall.key, /^staged:/, 'an idempotency key');
++ const chatRow = stubRows('channel_moderators').find((x) => x.user_id === mod);
++ assert.deepStrictEqual(res.body.moderators.map((x) => x.user_id).sort(), [mod, viewer].sort(), 'the answer reads Live’s copy, already current');
++ assert.strictEqual(d.prepare('SELECT id FROM channel_moderators WHERE user_id = ?').get(mod).id, chatRow.id, 'Chat’s id');
++ assert.strictEqual(outbox('channel_moderators').length, 0);
++ res = await call('DELETE', `/api/channels/${channel.id}/mods/${mod}`, { headers: { Authorization: `Bearer ${streamerJwt}` } });
++ assert.strictEqual(res.status, 200, res.text);
++ assert.ok(!db.isChannelModerator(mod, channel.id) && !stubRows('channel_moderators').some((x) => x.user_id === mod));
++ // Chat unreachable: the write fails and Live's copy is untouched.
++ stub.down = true;
++ res = await call('POST', `/api/channels/${channel.id}/mods`, { body: { username: 'moddy' }, headers: { Authorization: `Bearer ${streamerJwt}` } });
++ assert.strictEqual(res.status, 500);
++ assert.ok(!db.isChannelModerator(mod, channel.id), 'no local write when Chat writes the table');
++ stub.down = false;
++ // Other tables keep their own authority: emotes is still Live's.
++ const e2 = Number((await chatTables.write('createEmote', { user_id: viewer, code: 'kek', url: '/e/kek.png' })).lastInsertRowid);
++ assert.ok(db.getEmoteById(e2) && outbox('emotes').length === 1 && !stubRows('emotes').some((x) => x.code === 'kek'));
++ await sync.flush();
++
++ // 10. A handoff Chat refuses changes nothing; running it again is safe.
++ stub.authority.emotes = 'chat';
++ stub.refuseHandback = false;
++ f = await internal('POST', '/chat-tables/emotes', { authority: 'chat' });
++ assert.strictEqual(f.status, 200, 'Chat already agreed: the same call brings Live along');
++ stub.down = true;
++ f = await internal('POST', '/chat-tables/emotes', { authority: 'live' });
++ assert.strictEqual(f.status, 503, f.text);
++ assert.strictEqual(chatTables.authority('emotes'), 'chat', 'Chat unreachable: still Chat’s');
++ stub.down = false;
++ // Emote writes at 'chat' through the helper: Live's copy follows Chat's answer.
++ const e3 = Number((await chatTables.write('createEmote', { user_id: viewer, code: 'lul', url: '/e/lul.png' })).lastInsertRowid);
++ assert.strictEqual(db.getEmoteById(e3).code, 'lul');
++ await chatTables.write('deleteEmote', e3);
++ assert.strictEqual(db.getEmoteById(e3), undefined);
++
++ stub.refuseHandback = true;
++ f = await internal('POST', '/chat-tables/channel_moderators', { authority: 'live' });
++ assert.deepStrictEqual([f.status, chatTables.authority('channel_moderators')], [409, 'chat'], 'Chat still has changes for Live: nothing moves');
++ assert.match(f.body.error, /not in Live yet/);
++ stub.refuseHandback = false;
++
++ // 11. Back to 'live': Chat's changes first (the stub gave them), then Live writes and captures again.
++ f = await internal('POST', '/chat-tables/channel_moderators', { authority: 'live' });
++ assert.strictEqual(f.status, 200, f.text);
++ assert.deepStrictEqual([f.body.table.authority, f.body.table.chat_authority], ['live', 'live']);
++ await chatTables.write('addChannelModerator', channel.id, mod, streamer);
++ assert.ok(db.isChannelModerator(mod, channel.id));
++ assert.strictEqual(outbox('channel_moderators').length, 1, 'captured again');
++ m = await mirror([{ table: 'channel_moderators', op: 'delete', pk: { id: 1 } }]);
++ assert.strictEqual(m.body.applied, 0, 'Chat’s mirror is refused again');
++ s = (await internal('GET', '/chat-tables')).body;
++ assert.deepStrictEqual(Object.fromEntries(Object.entries(s.tables).map(([t, v]) => [t, v.authority])), { channel_moderators: 'live', channel_moderation_settings: 'live', emotes: 'chat', user_tags: 'live', chat_ai_summaries: 'live', chat_timeline_events: 'live' });
++ assert.ok(Object.values(s.tables).every((v) => !v.disagree), 'Live and Chat agree on every table');
++
++ // 12. Chat down with a table at 'chat': Live takes it back by force (after the operator drained
++ // Chat's mirror); until Chat's side is set too, Chat refuses Live's changes and they wait.
++ stub.down = true;
++ assert.strictEqual((await internal('POST', '/chat-tables/emotes', { authority: 'chat', force: true })).status, 400, 'force only takes back');
++ f = await internal('POST', '/chat-tables/emotes', { authority: 'live', force: true, by: 'test' });
++ assert.strictEqual(f.status, 200, f.text);
++ assert.strictEqual(chatTables.authority('emotes'), 'live');
++ stub.down = false;
++ await chatTables.write('updateEmote', e1, { code: 'pog2' });
++ r = await sync.flush();
++ assert.match(r.error, /OpenVibe.Chat writes emotes but Live does too/);
++ assert.strictEqual(outbox('emotes').length, 1, 'Live’s change is kept');
++ assert.strictEqual((await internal('GET', '/chat-tables')).body.tables.emotes.disagree, true);
++ stub.authority.emotes = 'live'; // scripts/table-authority.js set emotes live --force, in Chat
++ r = await sync.flush();
++ assert.deepStrictEqual([r.pending, r.error], [0, null]);
++ assert.strictEqual(stubRows('emotes').find((x) => x.id === e1).code, 'pog2');
++
++ quiet('chat tables (C-04, Live side): all checks passed');
++ } catch (err) {
++ console.error(err);
++ exit = 1;
++ } finally {
++ try { db.close(); } catch { /* */ }
++ fs.rmSync(tmp, { recursive: true, force: true });
++ process.exit(exit);
++ }
++})();