openvibers/openvibe.chat Commits

Staged tables move to Chat one table at a time (WS-I task 2, register C-04; Live d6480da): table_authority now says who writes channel_moderators, channel_moderation_settings, emotes, user_tags, chat_ai_summaries and chat_timeline_events, and a restart keeps what a table was handed. At 'live' Chat's copy follows Live's captured changes (bridge op stagedApply, rows REPLACEd, never mirrored back) and answers Live's dual read (stagedSlice: count and sha256 of a slice). setTableAuthority is Chat's half of Live's handoff: to 'chat' needs LIVE_MIRROR, back to 'live' sends the table's queued mirror rows first and refuses while any is left. At 'chat' the bridge takes Live's writers under Live's function names (addChannelModerator, removeChannelModerator, upsertChannelModerationSettings with Live's clamps, setChannelAlertSound, createEmote, updateEmote, deleteEmote, setEmoteMedia, grantUserTag, revokeUserTag, upsertChatAiSummary, addChatTimelineEvents), each refusing while its table is Live's and answering Live's return value plus the rows; the mirror triggers copy the table to Live; Chat's chat server reads moderators and settings in place and writes /slow and alert sounds itself. scripts/import-from-live.js is a dry run unless --apply (which backs chat.db up with VACUUM INTO first), makes a staged table at 'live' equal to Live's (refreshed, and pruned of rows Live removed, before inserting) and only compares one at 'chat'; --tables no longer crashes in the id-headroom step. scripts/parity-check.js --tables compares row counts and a content hash per staged table; scripts/table-authority.js shows both sides and sets Chat's side when Live cannot. /ready reports table_authority. Runbook with dates and rollback per step: docs/staged-tables-cutover.md

OP OpenVibers committed 26 Sept 2026 · e17dbbbf73 · 16 files, +1330 −67


Changes

ddiff --git a/README.md b/README.md
iindex c25615f..4a62e4c 100644
---- a/README.md
++++ b/README.md
@@@ -51,8 +51,12 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
Live's own chat calls here (`POST /internal/live/calls`).
- **Own database** (`CHAT_DB_PATH`, systemd `StateDirectory=openvibe-chat`) with Live's tables and
ids, the Network subject on new rows, a transactional events outbox and a read mirror back into Live.
--- **Import:** `scripts/import-from-live.js --live-db <snapshot> [--dry-run]` — idempotent, never drops
-- a row (`import_hold`), reports counts per table.
++- **Import:** `scripts/import-from-live.js --live-db <snapshot> [--apply]` — a dry run unless
++ `--apply` (which backs Chat's database up first), idempotent, never drops a row of Chat's tables
++ (`import_hold`), reports counts per table.
++- **Staged tables** (`channel_moderators`, `channel_moderation_settings`, `emotes`, `user_tags`,
++ `chat_ai_summaries`, `chat_timeline_events`): Live writes them until each is handed to Chat, one
++ table at a time (`table_authority`, register C-04). Runbook: `docs/staged-tables-cutover.md`.
## How it fits the network
@@@ -251,8 +255,10 @@ npm install
cp .env.example .env # OV_LIVE_INTERNAL_URL, Network URLs, OV_OAUTH_CLIENT_SECRET, SOUNDS_PATH
npm start # 127.0.0.1:4400 — /ws/chat, /api/{chat,dm,tts,sounds}, /health, /ready (CHAT_CALLS=1: /ws/call, /api/streams/…)
npm test # Node 22; stub Live and Network in-process
--node scripts/import-from-live.js --live-db /tmp/live-snapshot.db --dry-run
++node scripts/import-from-live.js --live-db /tmp/live-snapshot.db # dry run; --apply writes
node scripts/parity-check.js --live https://openvibe.live --chat http://127.0.0.1:4401 --before "…"
++node scripts/parity-check.js --tables --live-db /tmp/live-snapshot.db --chat-db /tmp/chat-snapshot.db
++node scripts/table-authority.js # who writes each staged table (the handoff is Live's)
node scripts/mirror-flush.js # rollback helper: push queued mirror rows to Live
node scripts/subscribe-events.js --dry-run # Chat's Events subscriptions (boot creates missing ones)
```
@@@ -282,8 +288,8 @@ server/net/service-auth.js service tokens: client (Chat → others) and guard (o
server/prefs/ chat preferences in the Network user module chat.preferences (routes, cache, migration from Live)
server/calls/ moved from Live: the call server (/ws/call), its REST routes, Live's stream hooks (/internal/calls), the calls lifecycle
server/db/ schema.sql, database.js (Live's chat functions, same names and arguments)
--scripts/ import-from-live, parity-check, mirror-flush, migrate-chat-preferences, subscribe-events
--docs/ cutover.md, calls-cutover.md, live-patch.diff, capabilities-proposal/
++scripts/ import-from-live, parity-check, mirror-flush, table-authority, migrate-chat-preferences, subscribe-events
++docs/ cutover.md, calls-cutover.md, staged-tables-cutover.md, live-patch.diff, capabilities-proposal/
```
## Owns
ddiff --git a/docs/cutover.md b/docs/cutover.md
iindex d7cec29..5b2fb3e 100644
---- a/docs/cutover.md
++++ b/docs/cutover.md
@@@ -41,7 +41,7 @@ nginx).
| Table (Live baseline, target OpenVibe.Chat) | W6 authority | Notes |
| --- | --- | --- |
| `chat_messages`, `dm_conversations`, `dm_participants`, `dm_messages`, `dm_blocks`, `tts_voice_overrides`, `channel_sounds`, `relay_users`, `hidden_relay_users`, `pending_ip_messages`, `stream_first_chats`, `moderation_actions` | **Chat** | Imported with their ids. Live keeps a read mirror (Chat → `POST /internal/chat-effects/mirror`); Live's remaining writers (AI viewers, relays, donations, `/api/mod`, `/api/channels` deletes, emote renames) forward to Chat. |
--| `channel_moderators`, `channel_moderation_settings`, `emotes`, `user_tags`, `chat_ai_summaries`, `chat_timeline_events` | **Live** (staged in Chat) | Their writers (`/api/channels`, dashboard, `/api/emotes` + Media asset-sync, game tags, chat AI) have not moved. Imported into Chat (a later import run refreshes them); Chat reads the live values through `live-context` and Live pushes invalidations. `/slow` and alert sounds are written by Live on Chat's behalf. |
++| `channel_moderators`, `channel_moderation_settings`, `emotes`, `user_tags`, `chat_ai_summaries`, `chat_timeline_events` | **Live** (staged in Chat) until each is handed over | Moved one table at a time by `docs/staged-tables-cutover.md` (register C-04): at `live` Live writes and its changes reach Chat's copy over the bridge; at `chat` Live's writers call Chat and the mirror copies the table back. Chat reads the moderation tables through `live-context` while Live writes them and in place once Chat does. |
| `media_requests`, `media_request_settings` | **Live** (not moved) | Decision with evidence: every writer is `server/media/media-queue.js` / `server/media/routes.js` (`/api/media`: gold payment, yt-dlp download, playback state, the streamer's overlay and dashboard). Chat only has the `!sr/!queue/!np/!skip` entry points, which call Live (`POST /internal/chat-effects/media-queue`). Not chat-owned in practice; it moves with the queue lifecycle the charter describes, not before. The importer reports them as `not_moved`. |
| `chat_messages_new`, `emotes_new`, `channel_sounds_new` | — | Transient tables of Live's table rebuilds; empty in a consistent snapshot. Rows found there go to `import_hold`. |
@@@ -101,8 +101,8 @@ SNAP_AT="$(date -u '+%Y-%m-%d %H:%M:%S')"
# 2. import into a scratch database
cd /opt/openvibe.chat
--node scripts/import-from-live.js --live-db /tmp/live-rehearsal.db --chat-db /tmp/chat-rehearsal.db --dry-run
node scripts/import-from-live.js --live-db /tmp/live-rehearsal.db --chat-db /tmp/chat-rehearsal.db
++node scripts/import-from-live.js --live-db /tmp/live-rehearsal.db --chat-db /tmp/chat-rehearsal.db --apply --no-backup
```
Check the report: per table `live` = `inserted` (first run), `held_total` 0 (anything held: look at
@@@ -136,7 +136,7 @@ streamers; `deploy.sh --wait-idle` waits for none).
```bash
sudo systemctl start openvibe-chat && curl -s http://127.0.0.1:4400/ready | jq
sqlite3 /opt/openvibe.live/data/live.db ".backup /tmp/live-cutover-1.db"
-- sudo -u ubuntu node scripts/import-from-live.js --live-db /tmp/live-cutover-1.db --chat-db /var/lib/openvibe-chat/chat.db
++ sudo -u ubuntu node scripts/import-from-live.js --live-db /tmp/live-cutover-1.db --chat-db /var/lib/openvibe-chat/chat.db --apply
```
Chat now holds everything up to snapshot 1 and its id sequences start 10000 above Live's
(`--headroom`), so rows Live writes until the flip still fit below.
@@@ -153,7 +153,7 @@ streamers; `deploy.sh --wait-idle` waits for none).
3. **Second import pass** — rows Live wrote between snapshot 1 and its restart:
```bash
sqlite3 /opt/openvibe.live/data/live.db ".backup /tmp/live-cutover-2.db"
-- sudo -u ubuntu node scripts/import-from-live.js --live-db /tmp/live-cutover-2.db --chat-db /var/lib/openvibe-chat/chat.db
++ sudo -u ubuntu node scripts/import-from-live.js --live-db /tmp/live-cutover-2.db --chat-db /var/lib/openvibe-chat/chat.db --apply
```
Chat's own new rows are already in Live (mirror) and count as `identical`; `chat_kept` counts rows
Chat edited since pass 1 (Chat wins); `held` must be 0.
ddiff --git a/docs/staged-tables-cutover.md b/docs/staged-tables-cutover.md
nnew file mode 100644
iindex 0000000..aa4cbec
---- /dev/null
++++ b/docs/staged-tables-cutover.md
@@@ -0,0 +1,288 @@
++# Staged tables cutover — Live → OpenVibe.Chat (WS-I task 2, register C-04)
++
++Six chat tables are still written by Live since the Wave 6 cutover (`docs/cutover.md`). This moves
++their authority to Chat **one table at a time**, each move reversible:
++
++| Table | Live's writers today | After the flip |
++| --- | --- | --- |
++| `chat_timeline_events` | `server/ai/chat-ai.js` (global timeline, seeding) | Live's chat AI calls Chat's `addChatTimelineEvents` |
++| `chat_ai_summaries` | `server/ai/chat-ai.js` (global, user, relay-user, anon insights) | Live's chat AI calls Chat's `upsertChatAiSummary` |
++| `user_tags` | none (chat tags are read-only in Live since the game moved to OpenVibe.Games) | data only; Chat has `grantUserTag` / `revokeUserTag` |
++| `emotes` | `POST/PATCH/DELETE /api/emotes`, Media asset-sync (`media_url`, `media_asset_id`) | Live's routes and asset-sync call Chat's `createEmote`, `updateEmote`, `deleteEmote`, `setEmoteMedia` |
++| `channel_moderation_settings` | `PUT /api/channels/:id/moderation` (dashboard), Chat's `/slow` and alert sounds (Live's `/internal/chat-effects/channel-settings` and `alert-sound`) | the dashboard calls Chat's `upsertChannelModerationSettings`; `/slow` and alert sounds are written by Chat itself |
++| `channel_moderators` | `POST/DELETE /api/channels/:id/mods` | Live's routes call Chat's `addChannelModerator` / `removeChannelModerator` |
++
++`media_requests` is **not** part of this: it is not a chat table. Its writers are Live's media
++queue (`server/media/media-queue.js`, `server/media/routes.js`, `server/tips/delivery-routes.js`:
++gold payment, yt-dlp download, playback state); Chat only has the `!sr/!queue/!np/!skip` entry
++points, which call Live; the plan's own media namespaces list it as `live.media_requests`. The
++decision of `docs/cutover.md` ("Data authority") stands.
++
++- [How it works](#how-it-works)
++- [Calendar](#calendar)
++- [Shell helpers](#shell-helpers)
++- [0. Deploy](#0-deploy)
++- [1. Import](#1-import)
++- [2. Parity](#2-parity)
++- [3. Dual read for one release](#3-dual-read-for-one-release)
++- [4. Flip each table](#4-flip-each-table)
++- [5. Live stops writing](#5-live-stops-writing)
++- [6. Rollback](#6-rollback)
++
++## How it works
++
++Two records say who writes each table, and they move together:
++
++- **Chat:** `table_authority` in `chat.db` (`scripts/table-authority.js`, `GET /ready` →
++ `table_authority`). A restart never changes it.
++- **Live:** app_state `chat_table_authority` (JSON; a missing table is `live`), read by
++ `server/chat/chat-tables.js`; `GET /internal/chat-tables` shows both sides. Without
++ `CHAT_AUTHORITY=chat` Live treats every table as `live`.
++
++**At `live`** (every table after the deploy): Live writes its own table as before. TEMP triggers
++capture every change into `chat_staged_outbox`, and a relay sends the rows as they are now to Chat's
++bridge (`POST /internal/live/calls`, op `stagedApply`, capability `chat.live_bridge.write`) every
++2 s, so Chat's copy stays current. Chat's staged writes refuse (`table.not_chat`). Chat's chat
++server reads moderators and settings through Live (`live-context`), as it has since Wave 6. With a
++table's **dual-read** flag on, Live's reads of it are compared in the background with Chat's copy
++(op `stagedSlice`: row count and a sha256 over the rows) and counted in `chat_dual_read_stats`;
++Live's answers never change.
++
++**The flip** (`POST /internal/chat-tables/:table {"authority":"chat"}` on Live, never automatic):
++the table's writers wait; Live's queued changes reach Chat; Chat takes the table (op
++`setTableAuthority`; refused if `LIVE_MIRROR` is off); Live records `chat` and stops capturing.
++
++**At `chat`:** Live's writers call Chat's write of the same name over the bridge (awaited, with an
++idempotency key); Chat answers Live's own return value plus the rows, which Live applies to its copy
++at once. Live's table is from then on a read mirror: Chat's triggers queue every change of the table
++in `live_mirror_outbox` and the mirror sends it to Live (`/internal/chat-effects/mirror`, as for
++Chat's own tables), which takes staged rows only for a table Chat writes. Live's readers do not
++change. Chat's chat server reads `channel_moderators` and `channel_moderation_settings` in place,
++and writes `/slow` and alert sounds itself.
++
++**The flip back** (`{"authority":"live"}`): Chat sends Live every queued change of the table and
++refuses to give it back while any is left; then Live records `live` and captures again. Running a
++flip again is safe: it brings both sides back into agreement after a failed attempt.
++
++## Calendar
++
++The plan (WS-I task 2) asks for dual read during one release before the flip. Live's rollback
++window is a week (`test/rollback-newer-writes.test.js`: production may roll back to any release of
++the last seven days), and a Live release from before this change would write a flipped table itself.
++So a table is flipped only when **both** hold: the dual-read counters stayed at 0 mismatches
++through at least one Live release shipped after D0, and every Live release of the last seven days
++has this change (at the earliest D0 + 7 days).
++
++| When (D0 = 2026-09-28) | What |
++| --- | --- |
++| D0, Mon 2026-09-28 | Deploy Chat, then Live. Import (dry run, apply). Parity. Dual read on for all six tables. |
++| D0 … D0 + 7 | Watch the counters daily. At least one more Live release ships in this week (any deploy counts), with dual read still on. |
++| Mon 2026-10-05 | Flip `chat_timeline_events`. |
++| Tue 2026-10-06 | Flip `chat_ai_summaries`. |
++| Wed 2026-10-07 | Flip `user_tags`. |
++| Thu 2026-10-08 | Flip `emotes`. |
++| Fri 2026-10-09 | Flip `channel_moderation_settings`. |
++| Mon 2026-10-12 | Flip `channel_moderators` (last: it decides who may moderate). |
++| until 2026-11-06 | Keep the flip back available. C-01 to C-03 (Live's chat mirror and bridge) have that target date and carry these tables too. |
++| not before 2026-12-18 | Contract step (register C-04 target): remove Live's local write path, the capture and the relay for the staged tables, after 30 days on `chat` without a flip back. |
++
++If the deploy slips, shift every date by the same number of days. If the counters show mismatches,
++the flips wait until a full release passes without any.
++
++## Shell helpers
++
++Live's internal routes answer only on loopback with `X-Internal-Key`. The helper takes the key from
++the running service and passes it to curl on stdin, so it is never printed or put on a command line.
++
++```bash
++live_tables() {
++ local key
++ key="$(sudo cat /proc/"$(systemctl show -p MainPID --value openvibe-live)"/environ | tr '\0' '\n' | sed -n 's/^INTERNAL_API_KEY=//p')"
++ printf 'X-Internal-Key: %s\n' "$key" | curl -sS -H @- -H 'Content-Type: application/json' "$@"
++}
++LT=http://127.0.0.1:3000/internal/chat-tables
++# Chat's side (same database as the service)
++chat_tables() { (cd /opt/openvibe.chat && sudo -u ubuntu env CHAT_DB_PATH=/var/lib/openvibe-chat/chat.db node scripts/table-authority.js "$@"); }
++STAGED=channel_moderators,channel_moderation_settings,emotes,user_tags,chat_ai_summaries,chat_timeline_events
++# The overview used below
++status() { live_tables "$LT" | jq '{relay, chat_error, tables: (.tables | map_values({authority, chat_authority, disagree, dual_read, outbox_pending, dr: (.dual_read_stats | {compared, matched, mismatched, inconclusive, errors, last_mismatch_at})}))}'; }
++```
++
++## 0. Deploy
++
++Chat first (it must know the new bridge ops before Live starts relaying), then Live. Nothing changes
++for users: every table stays at `live`.
++
++```bash
++cd /opt/openvibe.chat && sudo -u ubuntu git pull --ff-only
++sudo systemctl restart openvibe-chat && until curl -sf 127.0.0.1:4400/ready >/dev/null; do sleep 1; done
++curl -s 127.0.0.1:4400/ready | jq '{status, table_authority, mirror}'
++
++cd /opt/openvibe.live/current && sudo deploy/scripts/deploy.sh --wait-idle
++until curl -sf https://openvibe.live/api/ready >/dev/null; do sleep 1; done
++journalctl -u openvibe-live -n 300 | grep ChatTables
++status
++```
++
++Look at:
++
++- Chat `/ready`: `status` `ready`, every `table_authority` value `live`, `mirror.last_error` null.
++- Live's journal: `[ChatTables] staged tables: channel_moderators=live, …` (the capture and relay
++ started). No `relay to Chat waiting` lines after the first minute.
++- `status`: every table `authority` and `chat_authority` `live`, no `disagree`, `relay.pending`
++ near 0, `relay.last_error` null, `relay.paused` false.
++
++## 1. Import
++
++Chat's copy was last filled at the Wave 6 cutover. The import makes it equal to Live's: refreshed
++rows, missing rows added, rows Live no longer has removed (`pruned`). The relay is held meanwhile
++(Live keeps capturing), so every change made after the pause reaches Chat after the import and the
++import never undoes one.
++
++```bash
++# 1a. hold the relay
++live_tables -X POST -d '{"paused":true}' "$LT/relay" | jq .relay # paused: true
++# 1b. snapshot Live (online, consistent)
++sqlite3 /opt/openvibe.live/data/live.db ".backup /tmp/live-staged-import.db"
++# 1c. dry run (the default: nothing is written)
++cd /opt/openvibe.chat
++sudo -u ubuntu node scripts/import-from-live.js --live-db /tmp/live-staged-import.db \
++ --chat-db /var/lib/openvibe-chat/chat.db --tables "$STAGED" --no-projections | tee /tmp/staged-import-dry.json | jq '.tables, .held_total'
++# 1d. apply (copies chat.db to /var/lib/openvibe-chat/chat.db.pre-import-<time> first)
++sudo -u ubuntu node scripts/import-from-live.js --live-db /tmp/live-staged-import.db \
++ --chat-db /var/lib/openvibe-chat/chat.db --tables "$STAGED" --no-projections --apply | tee /tmp/staged-import.json | jq '.backup, .tables, .held_total'
++# 1e. release the relay: the changes since 1a go now
++live_tables -X POST -d '{"paused":false}' "$LT/relay" | jq .relay
++sleep 5; status
++```
++
++Look at:
++
++- Dry run and apply report the same numbers. Per table: `authority` `live`; `inserted + identical +
++ refreshed` = `live` (the snapshot's rows); `pruned` = rows only Chat had (removed moderators,
++ deleted emotes since 2026-09-23); `held` 0. `held_total` unchanged from before the run (anything
++ new: `sqlite3 /var/lib/openvibe-chat/chat.db "select source_table, reason from import_hold order by id desc limit 5"`).
++- `backup` names the copy of `chat.db` taken before the write.
++- After 1e: `relay.paused` false, `relay.pending` back near 0, `relay.last_error` null.
++- A second dry run (1c again) reports `inserted` 0 and `pruned` 0 for every table (idempotent).
++
++## 2. Parity
++
++Row counts and a content hash per table, on two snapshots taken back to back:
++
++```bash
++status | jq .relay.pending # 0 (nothing on its way)
++sqlite3 /opt/openvibe.live/data/live.db ".backup /tmp/live-parity.db"
++sudo -u ubuntu sqlite3 /var/lib/openvibe-chat/chat.db ".backup /tmp/chat-parity.db"
++cd /opt/openvibe.chat && sudo -u ubuntu node scripts/parity-check.js --tables --live-db /tmp/live-parity.db --chat-db /tmp/chat-parity.db
++```
++
++Look at: `all staged tables are the same`, one `same` line per table with its row count and
++authority `(live)`. A `DIFFERENT` line lists the keys only in Live, only in Chat, or different. A
++write between the two snapshots shows up this way once: run the three commands again a minute
++later. A key that differs twice is real: look at both rows
++(`sqlite3 … "select * from <table> where id = <key>"`), then repeat step 1 for that table
++(`--tables <table>`) and step 2. Afterwards delete the snapshots (they hold user data):
++`rm -f /tmp/live-staged-import.db /tmp/live-parity.db /tmp/chat-parity.db /tmp/staged-import*.json`.
++
++## 3. Dual read for one release
++
++```bash
++for t in ${STAGED//,/ }; do live_tables -X POST -d '{"dual_read":true}' "$LT/$t" | jq -c '{t: "'$t'", dual_read: .table.dual_read}'; done
++status
++```
++
++Every day until the flips (D0 … D0 + 7, and through at least one more Live release — each release
++restarts Live; the counters are kept in `chat_dual_read_stats` and survive it):
++
++```bash
++status
++journalctl -u openvibe-live --since -24h | grep -E 'ChatTables\] (dual read|relay)'
++ls /opt/openvibe.live/releases # a release after D0 has shipped
++```
++
++Look at:
++
++- `dr.mismatched` stays **0** for every table; `dr.compared` grows for the tables Live reads
++ (`channel_moderation_settings`, `channel_moderators`, `emotes`, `chat_ai_summaries`,
++ `chat_timeline_events`). `user_tags` has no reader in Live, so it compares nothing: parity (step
++ 2) is its check — run step 2 again the day before its flip.
++- `dr.inconclusive` counts comparisons made while a change was on its way to Chat; it is not a
++ mismatch. `dr.errors` counts Chat not answering; a few during Chat restarts are expected.
++- `relay.pending` near 0 and `relay.last_error` null at every look.
++- A mismatch: `live_tables "$LT" | jq '.tables.<table>.dual_read_stats.last_mismatch'` gives the
++ slice (`where`), both counts and hashes, and the keys (`only_live`, `only_chat`, `differ`). Find
++ the cause, repeat steps 1 and 2 for that table, reset its counters
++ (`live_tables -X POST -d '{"reset_counters":true}' "$LT/<table>"`) and start the release over.
++
++## 4. Flip each table
++
++One table per day, in the order of the calendar. Before each flip:
++
++```bash
++status # this table: mismatched 0, outbox_pending 0, no disagree
++curl -s 127.0.0.1:4400/ready | jq '{status, mirror}' # mirror.enabled true, pending near 0, last_error null
++C04=d6480da # the Live commit of this change (git -C /opt/openvibe.live/repo log --oneline --grep 'C-04')
++for r in /opt/openvibe.live/releases/*; do git -C /opt/openvibe.live/repo merge-base --is-ancestor "$C04" "${r##*-}" && echo "has it $r" || echo "OLDER $r"; done
++```
++
++Every release of the last seven days must say `has it` (`--rollback` must never land on one that
++writes the table itself). Then flip:
++
++```bash
++T=chat_timeline_events # the table of the day
++live_tables -X POST -d '{"authority":"chat","by":"'"$USER"'"}' "$LT/$T" | jq
++```
++
++Look at:
++
++- The answer: `ok` true, `handoff.authority` `chat`, `table.authority` and `table.chat_authority`
++ both `chat`, `table.outbox_pending` 0. `ok` false: nothing moved (`error` says why; Chat not
++ reachable, `LIVE_MIRROR is off`, Live's queue not drained). Fix it and run the same command again.
++- `journalctl -u openvibe-live -n 50 | grep "ChatTables\] $T"` → `live → chat`;
++ `journalctl -u openvibe-chat -n 50 | grep "$T: Chat writes it now"`.
++- `chat_tables` → `"$T": { "authority": "chat", … }`; `curl -s 127.0.0.1:4400/ready | jq .table_authority`.
++
++## 5. Live stops writing
++
++The flip is the moment Live stops writing the table: its writers call Chat and Live's copy only
++takes Chat's rows. Check it the same day, with a real write through the feature:
++
++| Table | A write to make | Then |
++| --- | --- | --- |
++| `chat_timeline_events`, `chat_ai_summaries` | none: the chat AI writes within 30 minutes while chat is active (`journalctl -u openvibe-live \| grep ChatAI`) | the newest row by `id` / `updated_at` is the same in both databases |
++| `user_tags` | none | step 2 for `user_tags` |
++| `emotes` | upload a test emote on a test channel, rename it, delete it (dashboard → Emotes) | each step shows in Chat's `emotes` with the same `id` as in Live's, then in Live's |
++| `channel_moderation_settings` | change a setting on a test channel in the dashboard; `/slow 5` then `/slow off` as its moderator | Chat's row changes first, Live's copy follows; slow mode works in the stream chat |
++| `channel_moderators` | add and remove a test moderator in the dashboard | the moderator can `/timeout` a test account in that channel's chat, and cannot after removal |
++
++```bash
++sudo -u ubuntu sqlite3 /var/lib/openvibe-chat/chat.db "select * from $T order by rowid desc limit 3"
++sqlite3 /opt/openvibe.live/data/live.db "select * from $T order by rowid desc limit 3"
++status | jq ".tables.$T" # outbox_pending stays 0: Live captures nothing now
++curl -s 127.0.0.1:4400/ready | jq .mirror # pending near 0, last_error null
++journalctl -u openvibe-live --since -1h | grep -E "ChatTables|refused" | grep -v 'staged tables:'
++```
++
++Look at: the same rows (same ids) on both sides; `outbox_pending` 0; no `OpenVibe.Chat refused`
++and no `not applied here` lines; Chat's mirror queue drains. For the emote, settings and moderator
++checks, also run step 2 for the table (the `same` line says `(chat)`). A write that fails while Chat
++is unreachable answers an error to the user and changes nothing in Live: that is the table being
++Chat's. After the last table (2026-10-12), run step 2 for all six.
++
++## 6. Rollback
++
++| Step | Rollback | Look at |
++| --- | --- | --- |
++| 0. Deploy | Nothing to undo while every table is `live`. Before rolling Live (`deploy.sh --rollback`) or Chat back past these changes, every table must be `live` again (row 4). | `status`: no table at `chat` |
++| 1. Import | The staged copy in Chat is read by nothing while its table is `live`, so a bad import has no effect on users. Repeat step 1 (pause, fresh snapshot, `--apply`, resume). **Never copy the `.pre-import-*` backup over `chat.db`**: it would drop the chat messages written since; it is for looking at. If step 1 stopped between 1a and 1e, release the relay: `live_tables -X POST -d '{"paused":false}' "$LT/relay"`. | `status`: `relay.paused` false, pending drains |
++| 2. Parity | Read-only. | — |
++| 3. Dual read | `for t in ${STAGED//,/ }; do live_tables -X POST -d '{"dual_read":false}' "$LT/$t" >/dev/null; done`. It never changed an answer. | `status`: `dual_read` false |
++| 4. Flip | `live_tables -X POST -d '{"authority":"live","by":"'"$USER"'"}' "$LT/$T" \| jq`: Chat sends its queued changes of the table to Live first, then Live writes it again. Refused while any change is still queued (`not in Live yet`): look at `curl -s 127.0.0.1:4400/ready \| jq .mirror` and run it again. | answer `table.authority` / `chat_authority` `live`; `outbox_pending` counting Live's new writes, then draining |
++| 4, Chat down | 1) send Chat's queue to Live: `cd /opt/openvibe.chat && sudo -u ubuntu env $(sudo cat /etc/openvibe/chat.env \| xargs) CHAT_DB_PATH=/var/lib/openvibe-chat/chat.db node scripts/mirror-flush.js` (exit 0 = drained; Live takes the rows only while the table is still `chat` there, so this comes first); 2) Chat's side: `chat_tables set $T live --force`; 3) Live's side: `live_tables -X POST -d '{"authority":"live","force":true}' "$LT/$T"`. | `status`: `authority` `live`; once Chat runs again `chat_authority` `live`, no `disagree` |
++| 5. Live stops writing | Same as step 4: the flip back makes Live the writer again. | as step 4 |
++| Wave 6 rollback | Before `docs/cutover.md` "Rollback", flip every table back (row 4). If Chat cannot run, its mirror flush there (step 2 of that rollback) carries the staged tables at `chat` too; then set each table back with the "Chat down" row. | `status` |
++
++If `status` shows `disagree` for a table (one side `chat`, the other `live`, after a failed or forced
++flip), run the flip that states the side you want once more; Live's relay keeps any change Chat
++refused in the meantime and sends it once both sides agree.
ddiff --git a/scripts/import-from-live.js b/scripts/import-from-live.js
iindex 2501696..0d7ae73 100644
---- a/scripts/import-from-live.js
++++ b/scripts/import-from-live.js
@@@ -2,11 +2,15 @@
/**
* Copy OpenVibe.Live's chat tables into OpenVibe.Chat's database.
*
-- * node scripts/import-from-live.js --live-db <copy of live.db> [--dry-run] [--chat-db <path>]
-- * [--headroom <n>] [--tables a,b] [--no-projections]
++ * node scripts/import-from-live.js --live-db <copy of live.db> [--apply [--backup <path> | --no-backup]]
++ * [--chat-db <path>] [--headroom <n>] [--tables a,b] [--no-projections]
*
* Run it against a SNAPSHOT (sqlite3 live.db ".backup /tmp/live-snapshot.db"), never the live file.
*
++ * - A dry run unless --apply: the report says what a run would do and nothing is written. --apply
++ * first copies Chat's database (VACUUM INTO <chat db>.pre-import-<time>, or --backup <path>) and
++ * says where in the report.
++ *
* - Idempotent: a row already in Chat is recognised by its primary key and never written twice;
* running it again after the cutover copies only what Live wrote since the first run.
* - Ids are kept (a message, DM or sound keeps its Live id). Chat's own id sequences start
@@@ -16,7 +20,10 @@
* an id already used by a DIFFERENT row, a constraint the row breaks, rows of Live's
* transient *_new migration tables.
* - Chat's tables: an existing Chat row always wins (after the cutover Chat is the authority
-- * and may have edited it). Staged tables (Live still writes them in W6): refreshed from Live.
++ * and may have edited it). Staged tables (C-04) follow their table_authority: at 'live' this copy
++ * is made equal to Live's — rows refreshed, and rows Live no longer has removed (`pruned`: a
++ * moderator removed in Live must not keep powers here); at 'chat' Chat is the authority and the
++ * run only reports how Live's copy differs (`differs`, `live_only`, `chat_only`).
* - New columns: *subject_id is filled from Live's linked_accounts (the Network subject).
* - media_requests / media_request_settings stay in Live (decision: docs/cutover.md) and are
* only reported.
@@@ -32,12 +39,15 @@ const path = require('path');
const Database = require('better-sqlite3');
function parseArgs(argv) {
-- const out = { dryRun: false, headroom: 10000, projections: true, tables: null };
++ const out = { dryRun: true, backup: true, headroom: 10000, projections: true, tables: null };
for (let i = 0; i < argv.length; i++) {
const a = argv[i];
if (a === '--live-db') out.liveDb = argv[++i];
else if (a === '--chat-db') out.chatDb = argv[++i];
else if (a === '--dry-run') out.dryRun = true;
++ else if (a === '--apply') out.dryRun = false;
++ else if (a === '--backup') out.backupPath = argv[++i];
++ else if (a === '--no-backup') out.backup = false;
else if (a === '--headroom') out.headroom = Math.max(0, parseInt(argv[++i], 10) || 0);
else if (a === '--tables') out.tables = String(argv[++i] || '').split(',').map((s) => s.trim()).filter(Boolean);
else if (a === '--no-projections') out.projections = false;
@@@ -74,17 +84,17 @@ const SUBJECT_COLUMNS = {
};
const TRANSIENT = ['chat_messages_new', 'emotes_new', 'channel_sounds_new'];
const NOT_MOVED = ['media_requests', 'media_request_settings'];
--const STAGED_PKS = {
-- channel_moderators: ['id'],
-- channel_moderation_settings: ['channel_id'],
-- emotes: ['id'],
-- user_tags: ['id'],
-- chat_ai_summaries: ['id'],
-- chat_timeline_events: ['id'],
--};
--
class Rollback extends Error {}
++/** Copy Chat's database before a write run (VACUUM INTO: consistent, compact, the WAL included). */
++function backupChat(chat, target) {
++ const fs = require('fs');
++ const file = path.resolve(target || `${chat.name}.pre-import-${new Date().toISOString().replace(/[:.]/g, '-')}`);
++ if (fs.existsSync(file)) throw new Error(`backup ${file} already exists`);
++ chat.prepare('VACUUM INTO ?').run(file);
++ return file;
++}
++
function run(opts) {
if (!opts.liveDb) throw new Error('--live-db <snapshot> is required');
if (opts.chatDb) process.env.CHAT_DB_PATH = opts.chatDb;
@@@ -97,9 +107,10 @@ function run(opts) {
const liveTables = new Set(live.prepare("SELECT name FROM sqlite_master WHERE type = 'table'").all().map((r) => r.name));
const colsOf = (conn, t) => conn.prepare(`PRAGMA table_info(${t})`).all().map((c) => c.name);
++ // A staged table Chat already writes is compared, never written (Chat is its authority).
const plan = [
...Object.entries(db.CHAT_TABLES).map(([t, pk]) => ({ table: t, pk, kind: 'chat' })),
-- ...Object.entries(STAGED_PKS).map(([t, pk]) => ({ table: t, pk, kind: 'staged' })),
++ ...Object.entries(db.STAGED_KEYS).map(([t, pk]) => ({ table: t, pk, kind: db.tableAuthority(t) === 'chat' ? 'staged-chat' : 'staged' })),
].filter((p) => !opts.tables || opts.tables.includes(p.table));
// Schema drift check first: nothing is written when Live has columns Chat cannot hold.
@@@ -126,12 +137,35 @@ function run(opts) {
const holdStmt = chat.prepare('INSERT OR IGNORE INTO import_hold (source_table, source_pk, reason, row_json) VALUES (?, ?, ?, ?)');
const hold = (table, pkObj, reason, row) => holdStmt.run(table, JSON.stringify(pkObj), String(reason).slice(0, 300), JSON.stringify(row)).changes;
-- const report = { live_db: path.resolve(opts.liveDb), dry_run: !!opts.dryRun, tables: {} };
++ // A staged table Chat writes: count how Live's copy (its mirror) differs, write nothing.
++ function compareOnly(p, r) {
++ const chatCols = new Set(colsOf(chat, p.table));
++ const cols = colsOf(live, p.table).filter((c) => chatCols.has(c));
++ const getChat = chat.prepare(`SELECT ${cols.join(', ')} FROM ${p.table} WHERE ${p.pk.map((k) => `${k} = ?`).join(' AND ')}`);
++ const seen = new Set();
++ for (const row of live.prepare(`SELECT ${cols.join(', ')} FROM ${p.table}`).iterate()) {
++ r.live++;
++ seen.add(JSON.stringify(p.pk.map((k) => row[k])));
++ const mine = getChat.get(...p.pk.map((k) => row[k]));
++ if (!mine) r.live_only++;
++ else if (cols.every((c) => mine[c] === row[c] || (mine[c] == null && row[c] == null))) r.identical++;
++ else r.differs++;
++ }
++ for (const row of chat.prepare(`SELECT ${p.pk.join(', ')} FROM ${p.table}`).iterate()) {
++ if (!seen.has(JSON.stringify(p.pk.map((k) => row[k])))) r.chat_only++;
++ }
++ }
++
++ const report = { live_db: path.resolve(opts.liveDb), dry_run: !!opts.dryRun, backup: null, tables: {} };
++ if (!opts.dryRun && opts.backup) report.backup = backupChat(chat, opts.backupPath);
for (const p of plan) {
-- const r = { live: 0, inserted: 0, identical: 0, refreshed: 0, chat_kept: 0, held: 0 };
++ const r = p.kind === 'staged-chat'
++ ? { authority: 'chat', live: 0, identical: 0, differs: 0, live_only: 0, chat_only: 0 }
++ : { live: 0, inserted: 0, identical: 0, refreshed: 0, chat_kept: 0, held: 0, ...(p.kind === 'staged' ? { authority: 'live', pruned: 0 } : {}) };
report.tables[p.table] = r;
if (!liveTables.has(p.table)) { r.missing_in_live = true; continue; }
++ if (p.kind === 'staged-chat') { compareOnly(p, r); continue; }
const chatCols = new Set(colsOf(chat, p.table));
const cols = colsOf(live, p.table).filter((c) => chatCols.has(c));
const subjectCols = (SUBJECT_COLUMNS[p.table] || []).filter(([sc]) => chatCols.has(sc));
@@@ -145,6 +179,16 @@ function run(opts) {
const orderBy = p.pk.join(', ');
const apply = () => {
++ // A staged table at 'live' is a copy: what Live no longer has goes first (a row Live
++ // removed and added again under a new id would otherwise collide on its unique columns).
++ if (p.kind === 'staged') {
++ const keyOf = (row) => JSON.stringify(p.pk.map((k) => row[k]));
++ const inLive = new Set(live.prepare(`SELECT ${p.pk.join(', ')} FROM ${p.table}`).all().map(keyOf));
++ const del = chat.prepare(`DELETE FROM ${p.table} WHERE ${where}`);
++ for (const row of chat.prepare(`SELECT ${p.pk.join(', ')} FROM ${p.table}`).all()) {
++ if (!inLive.has(keyOf(row))) r.pruned += del.run(...p.pk.map((k) => row[k])).changes;
++ }
++ }
for (const row of live.prepare(`SELECT ${cols.join(', ')} FROM ${p.table} ORDER BY ${orderBy}`).iterate()) {
r.live++;
const pkVals = p.pk.map((k) => row[k]);
@@@ -167,8 +211,12 @@ function run(opts) {
continue;
}
if (p.kind === 'staged' && refresh) {
-- refresh.run(...upd.map((c) => row[c]), ...pkVals);
-- r.refreshed++;
++ try {
++ refresh.run(...upd.map((c) => row[c]), ...pkVals);
++ r.refreshed++;
++ } catch (err) {
++ r.held += hold(p.table, pkObj, `refresh failed: ${err.message}`, row);
++ }
} else {
r.chat_kept++;
}
@@@ -207,7 +255,8 @@ function run(opts) {
// Chat's new ids start above Live's highest id plus headroom (never lowered).
chat.transaction(() => {
for (const [t] of Object.entries(db.CHAT_TABLES)) {
-- if (!liveTables.has(t) || !colsOf(live, t).includes('id')) continue;
++ // (only the tables this run imported: --tables leaves the others as they are)
++ if (!report.tables[t] || !liveTables.has(t) || !colsOf(live, t).includes('id')) continue;
const liveMax = live.prepare(`SELECT COALESCE(MAX(id), 0) AS m FROM ${t}`).get().m;
const chatMax = chat.prepare(`SELECT COALESCE(MAX(id), 0) AS m FROM ${t}`).get().m;
const target = Math.max(liveMax + opts.headroom, chatMax);
@@@ -252,7 +301,7 @@ if (require.main === module) {
let opts;
try { opts = parseArgs(process.argv.slice(2)); } catch (err) { console.error(err.message); process.exit(2); }
if (opts.help || !opts.liveDb) {
-- console.log('usage: node scripts/import-from-live.js --live-db <snapshot> [--dry-run] [--chat-db <path>] [--headroom <n>] [--tables a,b] [--no-projections]');
++ console.log('usage: node scripts/import-from-live.js --live-db <snapshot> [--apply [--backup <path> | --no-backup]] [--chat-db <path>] [--headroom <n>] [--tables a,b] [--no-projections]');
process.exit(opts.help ? 0 : 2);
}
try {
ddiff --git a/scripts/parity-check.js b/scripts/parity-check.js
iindex d23e4c3..3123220 100644
---- a/scripts/parity-check.js
++++ b/scripts/parity-check.js
@@@ -5,6 +5,14 @@
* node scripts/parity-check.js --live https://openvibe.live --chat http://127.0.0.1:4401 \
* [--before "2026-09-22 10:00:00"] [--stream 123]... [--channel 45]... [--token <jwt>]
*
++ * node scripts/parity-check.js --tables --live-db <live.db copy> --chat-db <chat.db or a copy> \
++ * [--table emotes]...
++ *
++ * --tables compares the staged tables (C-04, docs/staged-tables-cutover.md) row for row: per table
++ * the row count and a sha256 over every row (the columns both copies have, ordered by key — the hash
++ * Chat's dual read uses, database.js sliceHash), and on a difference the first keys that are only in
++ * one copy or differ. Both files are opened read-only. Exit 1 on any difference.
++ *
* --before pins history pages to messages older than the snapshot, so rows written since do not
* count as differences. --token (a Network JWT of a test account) adds the DM routes. Prints one
* line per path (same / DIFFERENT with the first differing keys) and exits 1 on any difference.
@@@ -14,15 +22,52 @@
'use strict';
function parseArgs(argv) {
-- const o = { streams: [], channels: [] };
++ const o = { streams: [], channels: [], only: [] };
for (let i = 0; i < argv.length; i++) {
const a = argv[i], v = argv[i + 1];
-- if (a === '--live') { o.live = v; i++; } else if (a === '--chat') { o.chat = v; i++; } else if (a === '--before') { o.before = v; i++; } else if (a === '--stream') { o.streams.push(v); i++; } else if (a === '--channel') { o.channels.push(v); i++; } else if (a === '--token') { o.token = v; i++; } else throw new Error(`unknown argument ${a}`);
++ if (a === '--live') { o.live = v; i++; } else if (a === '--chat') { o.chat = v; i++; } else if (a === '--before') { o.before = v; i++; } else if (a === '--stream') { o.streams.push(v); i++; } else if (a === '--channel') { o.channels.push(v); i++; } else if (a === '--token') { o.token = v; i++; } else if (a === '--tables') { o.tables = true; } else if (a === '--live-db') { o.liveDb = v; i++; } else if (a === '--chat-db') { o.chatDb = v; i++; } else if (a === '--table') { o.only.push(v); i++; } else throw new Error(`unknown argument ${a}`);
}
-- if (!o.live || !o.chat) throw new Error('--live and --chat are required');
++ if (o.tables) { if (!o.liveDb || !o.chatDb) throw new Error('--tables needs --live-db and --chat-db'); } else if (!o.live || !o.chat) throw new Error('--live and --chat are required');
return o;
}
++/** --tables: the staged tables of two database files, row for row. → number of tables that differ */
++function tableParity(o, log = console.log) {
++ const path = require('path');
++ const Database = require('better-sqlite3');
++ const { STAGED_KEYS, sliceHash } = require('../server/db/database');
++ const live = new Database(path.resolve(o.liveDb), { readonly: true, fileMustExist: true });
++ const chat = new Database(path.resolve(o.chatDb), { readonly: true, fileMustExist: true });
++ const colsOf = (conn, t) => conn.prepare(`PRAGMA table_info(${t})`).all().map((c) => c.name);
++ let authority = {};
++ try { authority = Object.fromEntries(chat.prepare('SELECT table_name, authority FROM table_authority').all().map((r) => [r.table_name, r.authority])); } catch { /* an old copy */ }
++ let bad = 0;
++ for (const [t, pk] of Object.entries(STAGED_KEYS)) {
++ if (o.only.length && !o.only.includes(t)) continue;
++ const lc = colsOf(live, t), cc = colsOf(chat, t);
++ if (!lc.length || !cc.length) { bad++; log(`MISSING ${t}: ${!lc.length ? 'not in the Live copy' : 'not in the Chat copy'}`); continue; }
++ const cols = lc.filter((c) => cc.includes(c)).sort();
++ const q = `SELECT ${cols.join(', ')} FROM ${t} ORDER BY ${pk.join(', ')}`;
++ const a = live.prepare(q).all(), b = chat.prepare(q).all();
++ const ha = sliceHash(cols, a), hb = sliceHash(cols, b);
++ const label = `${t.padEnd(28)} (${authority[t] || '?'})`;
++ if (a.length === b.length && ha === hb) { log(`same ${label} rows ${a.length} sha256 ${ha.slice(0, 16)}`); continue; }
++ bad++;
++ const keyOf = (r) => JSON.stringify(pk.map((k) => r[k]));
++ const rowOf = (r) => JSON.stringify(cols.map((c) => (r[c] === undefined ? null : r[c])));
++ const ma = new Map(a.map((r) => [keyOf(r), rowOf(r)])), mb = new Map(b.map((r) => [keyOf(r), rowOf(r)]));
++ const onlyLive = [...ma.keys()].filter((k) => !mb.has(k));
++ const onlyChat = [...mb.keys()].filter((k) => !ma.has(k));
++ const differ = [...ma.keys()].filter((k) => mb.has(k) && mb.get(k) !== ma.get(k));
++ log(`DIFFERENT ${label} rows ${a.length} ≠ ${b.length} sha256 ${ha.slice(0, 16)} ≠ ${hb.slice(0, 16)}`);
++ const show = (name, list) => { if (list.length) log(` ${name} (${list.length}): ${list.slice(0, 5).join(' ')}${list.length > 5 ? ' …' : ''}`); };
++ show('only in Live', onlyLive); show('only in Chat', onlyChat); show('differ', differ);
++ }
++ live.close(); chat.close();
++ log(bad ? `\n${bad} table(s) differ` : '\nall staged tables are the same');
++ return bad;
++}
++
const IGNORE = new Set(['request_id', 'trace_id', 'latest_id']);
function diff(a, b, at = '', out = []) {
if (out.length > 5) return out;
@@@ -48,6 +93,7 @@ async function get(base, p, token) {
async function main() {
const o = parseArgs(process.argv.slice(2));
++ if (o.tables) process.exit(tableParity(o) ? 1 : 0);
const before = o.before ? `&before=${encodeURIComponent(o.before)}` : '';
const paths = [
`/api/chat/global/history?limit=200${before}`,
@@@ -72,4 +118,6 @@ async function main() {
process.exit(bad ? 1 : 0);
}
--main().catch((err) => { console.error(err.message); process.exit(2); });
++if (require.main === module) main().catch((err) => { console.error(err.message); process.exit(2); });
++
++module.exports = { parseArgs, tableParity };
ddiff --git a/scripts/table-authority.js b/scripts/table-authority.js
nnew file mode 100644
iindex 0000000..9094241
---- /dev/null
++++ b/scripts/table-authority.js
@@@ -0,0 +1,51 @@
++#!/usr/bin/env node
++/**
++ * Who writes each staged table (roadmap C-04, docs/staged-tables-cutover.md), from Chat's side.
++ *
++ * node scripts/table-authority.js → { tables: { t: { authority, rows, mirror_pending } } }
++ * node scripts/table-authority.js set <table> <live|chat> --force
++ *
++ * The normal way to move a table is Live's handoff (POST /internal/chat-tables/:table on Live), which
++ * drains both sides and moves Chat and Live together. `set` changes only Chat's table_authority, for
++ * when Live cannot run the handoff (Live down during a rollback); afterwards make Live agree with the
++ * same handoff call. Back to 'live' is refused while changes of the table still wait for the Live
++ * mirror (scripts/mirror-flush.js sends them). Uses the service's env (CHAT_DB_PATH).
++ */
++'use strict';
++
++const db = require('../server/db/database');
++
++function status() {
++ const tables = {};
++ for (const t of Object.keys(db.STAGED_KEYS)) {
++ tables[t] = { authority: db.tableAuthority(t), rows: db.get(`SELECT COUNT(*) AS n FROM ${t}`).n, mirror_pending: db.mirrorPending(t) };
++ }
++ return { tables };
++}
++
++function set(table, authority) {
++ if (!db.STAGED_KEYS[table]) throw new Error(`${table} is not a staged table (${Object.keys(db.STAGED_KEYS).join(', ')})`);
++ if (authority !== 'live' && authority !== 'chat') throw new Error('authority is live or chat');
++ const pending = db.mirrorPending(table);
++ if (authority === 'live' && pending) throw new Error(`${pending} change(s) to ${table} have not reached Live yet: run scripts/mirror-flush.js first`);
++ return { table, before: db.tableAuthority(table), after: db.setTableAuthority(table, authority) };
++}
++
++if (require.main === module) {
++ const [cmd, table, authority] = process.argv.slice(2).filter((a) => !a.startsWith('--'));
++ const force = process.argv.includes('--force');
++ try {
++ db.initDb({ captureMirror: false });
++ if (!cmd || cmd === 'status') console.log(JSON.stringify(status(), null, 2));
++ else if (cmd === 'set') {
++ if (!force) throw new Error('set changes Chat alone; the handoff is Live\'s POST /internal/chat-tables/:table. Add --force if Live cannot run it.');
++ console.log(JSON.stringify(set(table, authority), null, 2));
++ } else throw new Error('usage: table-authority.js [status] | set <table> <live|chat> --force');
++ db.close();
++ } catch (err) {
++ console.error(err.message);
++ process.exit(2);
++ }
++}
++
++module.exports = { status, set };
ddiff --git a/server/app.js b/server/app.js
iindex b271267..2fdacfa 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -145,6 +145,8 @@ function createApp({ chatServer, bridge, mirror, relay, events = null, callServe
connections: chatServer.getTotalConnections(),
live: { last_sync_at: ctx.stats.lastSyncAt, last_success_at: ctx.stats.lastSuccessAt, failures: ctx.stats.failures, last_error: ctx.stats.lastError },
mirror: mirror ? { enabled: config.live.mirror, pending, last_error: mirror.lastError() } : null,
++ // Who writes each staged table (C-04, docs/staged-tables-cutover.md).
++ table_authority: checks.db.status === 'ok' ? (() => { try { return db.stagedAuthorities(); } catch { return null; } })() : null,
events: {
enabled: !!config.events.url,
consumer: events ? { enabled: events.enabled, ...events.stats } : null,
ddiff --git a/server/bridge/live-bridge.js b/server/bridge/live-bridge.js
iindex ef4ed7a..053d04a 100644
---- a/server/bridge/live-bridge.js
++++ b/server/bridge/live-bridge.js
@@@ -24,6 +24,19 @@
* without it that op alone is refused ({ ok: false, code: 'capability.denied' }) and the
* rest of the batch still runs.
*
++ * Staged tables (roadmap C-04, docs/staged-tables-cutover.md), same endpoint and capability:
++ * db ops addChannelModerator … addChatTimelineEvents Live's writers once a table is at
++ * 'chat' (awaited one by one, with an idempotency key); each answers { value, mirror }:
++ * Live's own return value and the rows as they are now, which Live applies to its copy
++ * at once. Refused while the table is at 'live'.
++ * stagedApply [changes] Live's captured changes while a table is at 'live' (this copy
++ * stays current; refused per change once Chat writes the table)
++ * stagedSlice [table, where, columns] { count, hash, … } for Live's dual read
++ * tableAuthority [] { table: 'live' | 'chat' }
++ * setTableAuthority [table, authority] Chat's half of the handoff Live runs (never
++ * automatic): to 'chat' needs the Live mirror on; back to 'live' first sends every
++ * queued change of the table to Live and refuses while any is left.
++ *
* GET /internal/live/presence capability chat.presence.read
* Who is connected where (counts, slow modes, user/anon → ip), for Live's synchronous
* reads (getTotalConnections, getStreamViewerCount, getConnectedUserIp, findClientByAnonId,
@@@ -47,6 +60,10 @@ const DB_OPS = new Set([
'logModerationAction', 'recordFirstChat', 'setTtsVoiceOverride', 'deleteTtsVoiceOverride',
'createChannelSound', 'setChannelSoundEmote', 'deleteChannelSound', 'renameChannelSoundCommand',
'updateChannelSoundEmoteRefs',
++ // Staged tables at 'chat' (each refuses while its table is at 'live').
++ 'addChannelModerator', 'removeChannelModerator', 'upsertChannelModerationSettings', 'setChannelAlertSound',
++ 'createEmote', 'updateEmote', 'deleteEmote', 'setEmoteMedia', 'grantUserTag', 'revokeUserTag',
++ 'upsertChatAiSummary', 'addChatTimelineEvents',
]);
// ChatServer methods Live's modules call, and where their stream id argument sits (the stream is
@@@ -102,7 +119,35 @@ const REF_TTL_MS = 10 * 60 * 1000; // in memory
const REF_KEEP_MS = 24 * 3600 * 1000; // in bridge_refs
const REF_MIN = -(2 ** 40); // Live's placeholders are ≤ this
--function createBridge({ chatServer }) {
++/**
++ * Chat's half of a staged table's handoff (Live's server/chat/chat-tables.js runs it, its writers
++ * waiting meanwhile). To 'chat': Chat writes the table from now on and the mirror copies it to Live —
++ * so the mirror must be on. Back to 'live': every change Chat made to the table reaches Live first;
++ * the last check and the switch run in one tick, so no Chat write lands between them.
++ */
++async function handOver(table, authority, { mirror, config }) {
++ if (!db.STAGED_KEYS[table]) throw new Error(`${table} is not a staged table`);
++ if (authority !== 'live' && authority !== 'chat') throw new Error('authority is live or chat');
++ if (authority === 'chat') {
++ if (!mirror || !config.live.mirror) throw new Error('LIVE_MIRROR is off: Live would never get this table back');
++ db.setTableAuthority(table, 'chat');
++ console.log(`[Bridge] ${table}: Chat writes it now (table_authority chat)`);
++ return { table, authority: 'chat', mirror_pending: db.mirrorPending(table) };
++ }
++ const until = Date.now() + 15000;
++ while (db.mirrorPending(table) > 0 && mirror && config.live.mirror && Date.now() < until) {
++ const r = await mirror.flush();
++ if (r && r.error) break;
++ if (r && r.busy) await new Promise((res) => setTimeout(res, 100));
++ }
++ const left = db.mirrorPending(table);
++ if (left) throw new Error(`${left} change(s) to ${table} not in Live yet${mirror && mirror.lastError() ? ` (${mirror.lastError()})` : ''}`);
++ db.setTableAuthority(table, 'live');
++ console.log(`[Bridge] ${table}: back to Live (table_authority live)`);
++ return { table, authority: 'live', mirror_pending: 0 };
++}
++
++function createBridge({ chatServer, mirror = null, config = require('../config') }) {
const router = express.Router();
const refs = new Map(); // `${boot}|${ref}` → { id, at }
@@@ -210,6 +255,15 @@ function createBridge({ chatServer }) {
else throw new Error(`invalidate: unknown kind ${kind}`);
return null;
}
++ // Staged tables (C-04).
++ case 'stagedApply':
++ return db.applyStagedChanges(args[0]);
++ case 'stagedSlice':
++ return db.stagedSlice(String(args[0] || ''), args[1] && typeof args[1] === 'object' ? args[1] : {}, Array.isArray(args[2]) ? args[2] : null);
++ case 'tableAuthority':
++ return db.stagedAuthorities();
++ case 'setTableAuthority':
++ return handOver(String(args[0] || ''), String(args[1] || ''), { mirror, config });
default:
throw new Error(`unknown op ${op}`);
}
@@@ -262,4 +316,4 @@ function createBridge({ chatServer }) {
return router;
}
--module.exports = { createBridge, DB_OPS, SERVER_OPS, MESSAGE_SEND, sendsMessage };
++module.exports = { createBridge, handOver, DB_OPS, SERVER_OPS, MESSAGE_SEND, sendsMessage };
ddiff --git a/server/bridge/live-mirror.js b/server/bridge/live-mirror.js
iindex 6e31aca..4d5b0c4 100644
---- a/server/bridge/live-mirror.js
++++ b/server/bridge/live-mirror.js
@@@ -3,8 +3,9 @@
*
* After the cutover Chat is the only writer of its tables, but Live still reads some of them in
* place (home-page stats, recaps, AI chat context, VOD chat replay, /api/mod queues, analytics),
-- * and rollback means Live becomes the authority again. So every change Chat makes to its tables
-- * is copied into Live's tables of the same name, same ids:
++ * and rollback means Live becomes the authority again. So every change Chat makes to its tables —
++ * and to each staged table once it is at 'chat' (C-04, docs/staged-tables-cutover.md) — is copied
++ * into Live's tables of the same name, same ids:
*
* - this connection's TEMP triggers (db.initDb({ captureMirror: true })) record each change in
* live_mirror_outbox — writes by the importer (rows that came from Live) are never recorded;
@@@ -39,7 +40,8 @@ function createMirror({ config, fetchImpl = (...a) => globalThis.fetch(...a), lo
for (const r of rows) byKey.set(`${r.tbl}|${r.pk}`, r);
const changes = [];
for (const r of byKey.values()) {
-- const cols = db.CHAT_TABLES[r.tbl];
++ // A staged table (C-04) is queued only while Chat writes it (its triggers check).
++ const cols = db.CHAT_TABLES[r.tbl] || db.STAGED_KEYS[r.tbl];
if (!cols) continue;
let pk;
try { pk = JSON.parse(r.pk); } catch { continue; }
ddiff --git a/server/chat/chat-server.js b/server/chat/chat-server.js
iindex 4d3cb5f..88b5115 100644
---- a/server/chat/chat-server.js
++++ b/server/chat/chat-server.js
@@@ -1679,7 +1679,7 @@ class ChatServer {
// Per-stream slow mode (not global)
if (client.streamId) {
this.slowModeByStream.set(client.streamId, seconds > 0 ? seconds * 1000 : 0);
-- // Persist (channel_moderation_settings is written by Live in this wave)
++ // Persist (by Live, or here once Chat writes channel_moderation_settings: live-context effects)
try {
const stream = ctx.getStreamById(client.streamId);
if (stream?.channel_id) {
ddiff --git a/server/chat/sounds-routes.js b/server/chat/sounds-routes.js
iindex 2a31b3b..86f532d 100644
---- a/server/chat/sounds-routes.js
++++ b/server/chat/sounds-routes.js
@@@ -450,7 +450,8 @@ router.post('/alert/:kind', requireAuth, soundUpload.single('sound'), async (req
const ext = path.extname(finalPath).toLowerCase();
const mime = EXT_TO_MIME[ext] || 'audio/mpeg';
-- // The alert columns are on Live's channel_moderation_settings; Live's alerts read them.
++ // The alert columns are on channel_moderation_settings (Live's until the C-04 handoff, then
++ // Chat's and mirrored); Live's alerts read them.
await ctx.effects.setChannelAlertSound(channel.id, kind, finalPath, mime, req.user.id);
res.json({ set: true, kind, url: `/api/sounds/file/${path.basename(finalPath)}` });
} catch (err) {
ddiff --git a/server/db/database.js b/server/db/database.js
iindex 3c227e1..83fd6ef 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -36,14 +36,27 @@ const CHAT_TABLES = {
stream_first_chats: ['chatter_key', 'channel_user_id'],
moderation_actions: ['id'],
};
--// Chat-target tables Live still writes in this wave: imported here, read through live-context.
++// Chat-target tables whose writer moves per table (roadmap C-04, docs/staged-tables-cutover.md).
++// table_authority says who writes each one: 'live' (the default) — Live writes, and every change
++// it makes reaches this copy over the bridge (applyStagedChanges); 'chat' — the staged writes
++// below are the only writers, Live's writers call them over the bridge and the Live mirror copies
++// the table back. The notes say who wrote them before the move.
const STAGED_TABLES = {
-- channel_moderators: 'written by Live /api/channels (channel-mod-routes) until that route moves',
-- channel_moderation_settings: 'written by Live /api/channels and the dashboard; /slow and alert sounds go through Live',
-- emotes: 'written by Live /api/emotes and its Media asset-sync',
-- user_tags: 'written by Live game/tags (shop); chat reads tags through live-context',
-- chat_ai_summaries: 'written by Live server/ai/chat-ai.js',
-- chat_timeline_events: 'written by Live server/ai/chat-ai.js',
++ channel_moderators: 'Live /api/channels (channel-mod-routes) while at live',
++ channel_moderation_settings: 'Live /api/channels, the dashboard, /slow and alert sounds while at live',
++ emotes: 'Live /api/emotes and its Media asset-sync while at live',
++ user_tags: 'no writer since Live’s game tags went read-only; data only',
++ chat_ai_summaries: 'Live server/ai/chat-ai.js while at live',
++ chat_timeline_events: 'Live server/ai/chat-ai.js while at live',
++};
++// Their primary keys (the mirror and Live's changes address rows by them).
++const STAGED_KEYS = {
++ channel_moderators: ['id'],
++ channel_moderation_settings: ['channel_id'],
++ emotes: ['id'],
++ user_tags: ['id'],
++ chat_ai_summaries: ['id'],
++ chat_timeline_events: ['id'],
};
function getDb() {
@@@ -118,21 +131,30 @@ function initDb({ captureMirror = false } = {}) {
d.exec('CREATE INDEX IF NOT EXISTS idx_ctx_users_subject ON ctx_users(subject_id)');
const upsertAuth = d.prepare('INSERT INTO table_authority (table_name, authority, note) VALUES (?, ?, ?) ON CONFLICT(table_name) DO UPDATE SET authority = excluded.authority, note = excluded.note');
for (const t of Object.keys(CHAT_TABLES)) upsertAuth.run(t, 'chat', 'Chat writes; Live keeps a read mirror');
-- for (const [t, note] of Object.entries(STAGED_TABLES)) upsertAuth.run(t, 'live', note);
++ // A staged table keeps the authority it was handed (a restart never moves it back).
++ const seedStaged = d.prepare("INSERT INTO table_authority (table_name, authority, note) VALUES (?, 'live', ?) ON CONFLICT(table_name) DO UPDATE SET note = excluded.note");
++ for (const [t, note] of Object.entries(STAGED_TABLES)) seedStaged.run(t, note);
++ _authority.at = 0;
if (captureMirror) installMirrorTriggers();
return d;
}
function installMirrorTriggers() {
const d = getDb();
-- for (const [table, pk] of Object.entries(CHAT_TABLES)) {
++ const tables = [
++ ...Object.entries(CHAT_TABLES).map(([table, pk]) => [table, pk, '']),
++ // A staged table is mirrored only while Chat writes it (read at each change, so a handoff
++ // made by another process counts at once).
++ ...Object.entries(STAGED_KEYS).map(([table, pk]) => [table, pk, `WHEN (SELECT authority FROM main.table_authority WHERE table_name = '${table}') = 'chat'`]),
++ ];
++ for (const [table, pk, when] of tables) {
const obj = (alias) => `json_object(${pk.map((c) => `'${c}', ${alias}.${c}`).join(', ')})`;
d.exec(`
-- CREATE TEMP TRIGGER IF NOT EXISTS mirror_${table}_ins AFTER INSERT ON main.${table}
++ CREATE TEMP TRIGGER IF NOT EXISTS mirror_${table}_ins AFTER INSERT ON main.${table} ${when}
BEGIN INSERT INTO live_mirror_outbox (tbl, op, pk) VALUES ('${table}', 'upsert', ${obj('NEW')}); END;
-- CREATE TEMP TRIGGER IF NOT EXISTS mirror_${table}_upd AFTER UPDATE ON main.${table}
++ CREATE TEMP TRIGGER IF NOT EXISTS mirror_${table}_upd AFTER UPDATE ON main.${table} ${when}
BEGIN INSERT INTO live_mirror_outbox (tbl, op, pk) VALUES ('${table}', 'upsert', ${obj('NEW')}); END;
-- CREATE TEMP TRIGGER IF NOT EXISTS mirror_${table}_del AFTER DELETE ON main.${table}
++ CREATE TEMP TRIGGER IF NOT EXISTS mirror_${table}_del AFTER DELETE ON main.${table} ${when}
BEGIN INSERT INTO live_mirror_outbox (tbl, op, pk) VALUES ('${table}', 'delete', ${obj('OLD')}); END;
`);
}
@@@ -819,6 +841,386 @@ function updateChannelSoundEmoteRefs(ownerId, oldCode, newCode) {
[newCode || '', ownerId, oldCode]);
}
++// ── Staged tables (roadmap C-04) ──────────────────────────────
++// Who writes each one is table_authority (STAGED_TABLES above). The reads are cached for a second:
++// a handoff made by another process (scripts/table-authority.js) counts within that.
++const _authority = { at: 0, map: new Map() };
++function tableAuthority(table) {
++ if (Date.now() - _authority.at > 1000) {
++ _authority.map = new Map(all('SELECT table_name, authority FROM table_authority').map((r) => [r.table_name, r.authority]));
++ _authority.at = Date.now();
++ }
++ return _authority.map.get(table) || (CHAT_TABLES[table] ? 'chat' : 'live');
++}
++function setTableAuthority(table, authority) {
++ if (!STAGED_KEYS[table]) throw new Error(`${table} is not a staged table`);
++ if (authority !== 'live' && authority !== 'chat') throw new Error('authority is live or chat');
++ run('UPDATE table_authority SET authority = ? WHERE table_name = ?', [authority, table]);
++ _authority.at = 0;
++ return tableAuthority(table);
++}
++/** { table: authority } for the staged tables. */
++function stagedAuthorities() {
++ return Object.fromEntries(Object.keys(STAGED_KEYS).map((t) => [t, tableAuthority(t)]));
++}
++/** Changes of this table still queued for Live (the mirror drains them). */
++function mirrorPending(table) {
++ return get('SELECT COUNT(*) AS n FROM live_mirror_outbox WHERE tbl = ?', [table])?.n || 0;
++}
++
++const _colCache = new Map();
++function _columns(table) {
++ if (!_colCache.has(table)) _colCache.set(table, new Set(getDb().prepare(`PRAGMA table_info(${table})`).all().map((c) => c.name)));
++ return _colCache.get(table);
++}
++
++/** A staged write refuses while Live still writes the table: one writer at a time. */
++function _assertChatWrites(table) {
++ if (tableAuthority(table) === 'chat') return;
++ const err = new Error(`${table} is written by Live (table_authority live)`);
++ err.code = 'table.not_chat';
++ throw err;
++}
++
++/**
++ * What a staged write returns over the bridge: `value` — what Live's function of the same name
++ * returns (its callers keep working), and `mirror` — the rows as they are now, in the Live mirror's
++ * change shape, so Live's copy is right at once (the mirror sends the same rows again later).
++ */
++function _staged(value, table, rows = [], deletedPks = []) {
++ const plainValue = value && typeof value === 'object' && 'changes' in value && 'lastInsertRowid' in value
++ ? { changes: value.changes, lastInsertRowid: Number(value.lastInsertRowid) } : value;
++ return {
++ value: plainValue === undefined ? null : plainValue,
++ mirror: [
++ ...deletedPks.map((pk) => ({ table, op: 'delete', pk })),
++ ...rows.filter(Boolean).map((row) => ({ table, op: 'upsert', row })),
++ ],
++ };
++}
++
++// Channel moderators, settings and alert sounds (Live's /api/channels, the dashboard, /slow).
++function addChannelModerator(channelId, userId, addedBy) {
++ _assertChatWrites('channel_moderators');
++ const out = transaction(() => {
++ const res = run('INSERT OR IGNORE INTO channel_moderators (channel_id, user_id, added_by) VALUES (?, ?, ?)', [channelId, userId, addedBy]);
++ return _staged(res, 'channel_moderators', all('SELECT * FROM channel_moderators WHERE channel_id = ? AND user_id = ?', [channelId, userId]));
++ });
++ _ctx().invalidateChannel(channelId);
++ return out;
++}
++
++function removeChannelModerator(channelId, userId) {
++ _assertChatWrites('channel_moderators');
++ const out = transaction(() => {
++ const gone = all('SELECT id FROM channel_moderators WHERE channel_id = ? AND user_id = ?', [channelId, userId]);
++ const res = run('DELETE FROM channel_moderators WHERE channel_id = ? AND user_id = ?', [channelId, userId]);
++ return _staged(res, 'channel_moderators', [], gone.map((r) => ({ id: r.id })));
++ });
++ _ctx().invalidateChannel(channelId);
++ return out;
++}
++
++// Live's upsertChannelModerationSettings, clamps and all.
++function upsertChannelModerationSettings(channelId, fields) {
++ _assertChatWrites('channel_moderation_settings');
++ const out = transaction(() => {
++ const existing = get('SELECT 1 FROM channel_moderation_settings WHERE channel_id = ?', [channelId]);
++ if (existing) {
++ const updates = [];
++ const params = [];
++ const set = (col, v) => { updates.push(`${col} = ?`); params.push(v); };
++ if (fields.slow_mode_seconds !== undefined) set('slow_mode_seconds', fields.slow_mode_seconds);
++ if (fields.followers_only !== undefined) set('followers_only', fields.followers_only ? 1 : 0);
++ if (fields.emote_only !== undefined) set('emote_only', fields.emote_only ? 1 : 0);
++ if (fields.allow_anonymous !== undefined) set('allow_anonymous', fields.allow_anonymous ? 1 : 0);
++ if (fields.links_allowed !== undefined) set('links_allowed', fields.links_allowed ? 1 : 0);
++ if (fields.gifs_enabled !== undefined) set('gifs_enabled', fields.gifs_enabled ? 1 : 0);
++ if (fields.account_age_gate_hours !== undefined) set('account_age_gate_hours', Number(fields.account_age_gate_hours) || 0);
++ if (fields.caps_percentage_limit !== undefined) set('caps_percentage_limit', Number(fields.caps_percentage_limit) || 0);
++ if (fields.aggressive_filter !== undefined) set('aggressive_filter', fields.aggressive_filter ? 1 : 0);
++ if (fields.max_message_length !== undefined) set('max_message_length', Math.max(50, Number(fields.max_message_length) || 500));
++ if (fields.slur_filter_enabled !== undefined) set('slur_filter_enabled', fields.slur_filter_enabled ? 1 : 0);
++ if (fields.slur_filter_use_builtin !== undefined) set('slur_filter_use_builtin', fields.slur_filter_use_builtin ? 1 : 0);
++ if (fields.slur_filter_terms !== undefined) set('slur_filter_terms', String(fields.slur_filter_terms || '').slice(0, 4000));
++ if (fields.slur_filter_regexes !== undefined) set('slur_filter_regexes', String(fields.slur_filter_regexes || '').slice(0, 8000));
++ if (fields.slur_filter_nudge_message !== undefined) set('slur_filter_nudge_message', String(fields.slur_filter_nudge_message || '').slice(0, 800));
++ if (fields.slur_filter_disabled_categories !== undefined) set('slur_filter_disabled_categories', String(fields.slur_filter_disabled_categories || '[]').slice(0, 200));
++ if (fields.ip_approval_mode !== undefined) set('ip_approval_mode', fields.ip_approval_mode ? 1 : 0);
++ if (fields.soundboard_enabled !== undefined) set('soundboard_enabled', fields.soundboard_enabled ? 1 : 0);
++ if (fields.soundboard_allow_pitch !== undefined) set('soundboard_allow_pitch', fields.soundboard_allow_pitch ? 1 : 0);
++ if (fields.soundboard_allow_speed !== undefined) set('soundboard_allow_speed', fields.soundboard_allow_speed ? 1 : 0);
++ if (fields.soundboard_banned_ids !== undefined) set('soundboard_banned_ids', String(fields.soundboard_banned_ids || '').slice(0, 4000));
++ if (fields.viewer_auto_delete_enabled !== undefined) set('viewer_auto_delete_enabled', fields.viewer_auto_delete_enabled ? 1 : 0);
++ if (fields.viewer_delete_all_enabled !== undefined) set('viewer_delete_all_enabled', fields.viewer_delete_all_enabled ? 1 : 0);
++ if (fields.custom_emotes_enabled !== undefined) set('custom_emotes_enabled', fields.custom_emotes_enabled ? 1 : 0);
++ if (fields.custom_sounds_enabled !== undefined) set('custom_sounds_enabled', fields.custom_sounds_enabled ? 1 : 0);
++ if (fields.max_sound_seconds !== undefined) set('max_sound_seconds', Math.min(30, Math.max(1, Number(fields.max_sound_seconds) || 10)));
++ if (fields.uploads_mods_only !== undefined) set('uploads_mods_only', fields.uploads_mods_only ? 1 : 0);
++ if (fields.mods_can_edit_about !== undefined) set('mods_can_edit_about', fields.mods_can_edit_about ? 1 : 0);
++ if (fields.emote_scale !== undefined) set('emote_scale', Math.min(300, Math.max(50, Number(fields.emote_scale) || 100)));
++ if (fields.emote_size_min !== undefined) set('emote_size_min', Math.min(200, Math.max(25, Number(fields.emote_size_min) || 50)));
++ if (fields.emote_size_max !== undefined) set('emote_size_max', Math.min(400, Math.max(50, Number(fields.emote_size_max) || 200)));
++ if (fields.sounds_mods_only !== undefined) set('sounds_mods_only', fields.sounds_mods_only ? 1 : 0);
++ if (fields.sound_min_speed !== undefined) set('sound_min_speed', Math.min(1, Math.max(0.1, Number(fields.sound_min_speed) || 0.5)));
++ if (fields.sound_max_speed !== undefined) set('sound_max_speed', Math.min(5, Math.max(1, Number(fields.sound_max_speed) || 3.0)));
++ if (fields.sound_min_pitch_cents !== undefined) set('sound_min_pitch_cents', Math.min(0, Math.max(-2400, Math.round(Number(fields.sound_min_pitch_cents) || -1200))));
++ if (fields.sound_max_pitch_cents !== undefined) set('sound_max_pitch_cents', Math.max(0, Math.min(2400, Math.round(Number(fields.sound_max_pitch_cents) || 1200))));
++ if (updates.length > 0) {
++ updates.push('updated_at = CURRENT_TIMESTAMP');
++ params.push(channelId);
++ run(`UPDATE channel_moderation_settings SET ${updates.join(', ')} WHERE channel_id = ?`, params);
++ }
++ } else {
++ const b = (v, dflt) => (v !== undefined ? (v ? 1 : 0) : dflt);
++ run(
++ `INSERT INTO channel_moderation_settings (
++ channel_id, slow_mode_seconds, followers_only, emote_only,
++ allow_anonymous, links_allowed, gifs_enabled, account_age_gate_hours,
++ caps_percentage_limit, aggressive_filter, max_message_length,
++ slur_filter_enabled, slur_filter_use_builtin, slur_filter_terms, slur_filter_regexes, slur_filter_nudge_message, slur_filter_disabled_categories,
++ ip_approval_mode, soundboard_enabled, soundboard_allow_pitch, soundboard_allow_speed, soundboard_banned_ids,
++ viewer_auto_delete_enabled, viewer_delete_all_enabled,
++ custom_emotes_enabled, custom_sounds_enabled, max_sound_seconds, uploads_mods_only, emote_scale,
++ emote_size_min, emote_size_max, sounds_mods_only, mods_can_edit_about
++ ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
++ [
++ channelId,
++ fields.slow_mode_seconds || 0,
++ fields.followers_only ? 1 : 0,
++ fields.emote_only ? 1 : 0,
++ b(fields.allow_anonymous, 1),
++ b(fields.links_allowed, 1),
++ b(fields.gifs_enabled, 1),
++ Number(fields.account_age_gate_hours) || 0,
++ Number(fields.caps_percentage_limit) || 0,
++ fields.aggressive_filter ? 1 : 0,
++ Math.max(50, Number(fields.max_message_length) || 500),
++ fields.slur_filter_enabled ? 1 : 0,
++ b(fields.slur_filter_use_builtin, 1),
++ String(fields.slur_filter_terms || '').slice(0, 4000),
++ String(fields.slur_filter_regexes || '').slice(0, 8000),
++ String(fields.slur_filter_nudge_message || '').slice(0, 800),
++ String(fields.slur_filter_disabled_categories || '[]').slice(0, 200),
++ fields.ip_approval_mode ? 1 : 0,
++ b(fields.soundboard_enabled, 1),
++ b(fields.soundboard_allow_pitch, 1),
++ b(fields.soundboard_allow_speed, 1),
++ String(fields.soundboard_banned_ids || '').slice(0, 4000),
++ b(fields.viewer_auto_delete_enabled, 1),
++ b(fields.viewer_delete_all_enabled, 1),
++ b(fields.custom_emotes_enabled, 1),
++ b(fields.custom_sounds_enabled, 1),
++ Math.min(30, Math.max(1, Number(fields.max_sound_seconds) || 10)),
++ fields.uploads_mods_only ? 1 : 0,
++ Math.min(300, Math.max(50, Number(fields.emote_scale) || 100)),
++ Math.min(200, Math.max(25, Number(fields.emote_size_min) || 50)),
++ Math.min(400, Math.max(50, Number(fields.emote_size_max) || 200)),
++ fields.sounds_mods_only ? 1 : 0,
++ fields.mods_can_edit_about ? 1 : 0,
++ ]
++ );
++ }
++ // tts_max_length, as Live: after the UPDATE or the fresh INSERT.
++ if (fields.tts_max_length !== undefined) {
++ run('UPDATE channel_moderation_settings SET tts_max_length = ? WHERE channel_id = ?', [Math.min(1000, Math.max(10, Number(fields.tts_max_length) || 200)), channelId]);
++ }
++ const row = get('SELECT * FROM channel_moderation_settings WHERE channel_id = ?', [channelId]);
++ return _staged(row, 'channel_moderation_settings', [row]);
++ });
++ _ctx().invalidateChannel(channelId);
++ return out;
++}
++
++// Donation / goal alert sounds live on the settings row (url = the file's path in the sounds dir).
++function setChannelAlertSound(channelId, kind, url, mime) {
++ _assertChatWrites('channel_moderation_settings');
++ const out = transaction(() => {
++ if (!get('SELECT 1 FROM channel_moderation_settings WHERE channel_id = ?', [channelId])) {
++ run('INSERT INTO channel_moderation_settings (channel_id) VALUES (?)', [channelId]);
++ }
++ const col = kind === 'goal' ? 'goal_sound' : 'donation_sound';
++ const res = run(`UPDATE channel_moderation_settings SET ${col}_url = ?, ${col}_mime = ? WHERE channel_id = ?`, [url || null, mime || null, channelId]);
++ return _staged(res, 'channel_moderation_settings', [get('SELECT * FROM channel_moderation_settings WHERE channel_id = ?', [channelId])]);
++ });
++ _ctx().invalidateChannel(channelId);
++ return out;
++}
++
++// Emotes (Live's /api/emotes; media_url/media_asset_id from its Media asset-sync).
++function createEmote({ user_id, code, url, animated = false, width = 28, height = 28, is_global = false, channel_owner_id = null, size = 100 }) {
++ _assertChatWrites('emotes');
++ return transaction(() => {
++ const res = run(
++ `INSERT INTO emotes (user_id, code, url, animated, width, height, is_global, channel_owner_id, size)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
++ [user_id, code, url, animated ? 1 : 0, width, height, is_global ? 1 : 0, channel_owner_id || null, Math.min(400, Math.max(25, parseInt(size, 10) || 100))]
++ );
++ return _staged(res, 'emotes', [get('SELECT * FROM emotes WHERE id = ?', [res.lastInsertRowid])]);
++ });
++}
++
++function updateEmote(id, { code, size } = {}) {
++ _assertChatWrites('emotes');
++ const sets = [];
++ const params = [];
++ if (code !== undefined) { sets.push('code = ?'); params.push(code); }
++ if (size !== undefined) { sets.push('size = ?'); params.push(Math.min(400, Math.max(25, parseInt(size, 10) || 100))); }
++ if (!sets.length) return _staged({ changes: 0 }, 'emotes');
++ return transaction(() => {
++ const res = run(`UPDATE emotes SET ${sets.join(', ')} WHERE id = ?`, [...params, id]);
++ return _staged(res, 'emotes', [get('SELECT * FROM emotes WHERE id = ?', [id])]);
++ });
++}
++
++function deleteEmote(id) {
++ _assertChatWrites('emotes');
++ return transaction(() => {
++ const had = get('SELECT id FROM emotes WHERE id = ?', [id]);
++ const res = run('DELETE FROM emotes WHERE id = ?', [id]);
++ return _staged(res, 'emotes', [], had ? [{ id: had.id }] : []);
++ });
++}
++
++/** The emote's copy on OpenVibe.Media (Live's asset-sync). */
++function setEmoteMedia(id, mediaUrl, mediaAssetId) {
++ _assertChatWrites('emotes');
++ return transaction(() => {
++ const res = run('UPDATE emotes SET media_url = ?, media_asset_id = ? WHERE id = ?', [mediaUrl || null, mediaAssetId || null, id]);
++ return _staged(res, 'emotes', [get('SELECT * FROM emotes WHERE id = ?', [id])]);
++ });
++}
++
++// User tags (owned chat tags; Live has had no writer since its game tags went read-only).
++function grantUserTag(userId, tagId, source = 'shop') {
++ _assertChatWrites('user_tags');
++ return transaction(() => {
++ const res = run('INSERT OR IGNORE INTO user_tags (user_id, tag_id, source) VALUES (?, ?, ?)', [userId, String(tagId), source || 'shop']);
++ return _staged(res, 'user_tags', [get('SELECT * FROM user_tags WHERE user_id = ? AND tag_id = ?', [userId, String(tagId)])]);
++ });
++}
++
++function revokeUserTag(userId, tagId) {
++ _assertChatWrites('user_tags');
++ return transaction(() => {
++ const gone = all('SELECT id FROM user_tags WHERE user_id = ? AND tag_id = ?', [userId, String(tagId)]);
++ const res = run('DELETE FROM user_tags WHERE user_id = ? AND tag_id = ?', [userId, String(tagId)]);
++ return _staged(res, 'user_tags', [], gone.map((r) => ({ id: r.id })));
++ });
++}
++
++// Chat AI (Live's server/ai/chat-ai.js): rolling summaries and the append-only timeline.
++function upsertChatAiSummary(sfx) {
++ _assertChatWrites('chat_ai_summaries');
++ const {
++ scope, subject_id = 0, window, overview = '', memory_json = '', timeline_json = '[]',
++ message_count = 0, window_message_count = 0, last_message_id = 0,
++ window_label = '', window_start = null, window_end = null,
++ } = sfx || {};
++ return transaction(() => {
++ const res = run(
++ `INSERT INTO chat_ai_summaries
++ (scope, subject_id, window, overview, memory_json, timeline_json, message_count,
++ window_message_count, last_message_id, window_label, window_start, window_end, updated_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
++ ON CONFLICT(scope, subject_id, window) DO UPDATE SET
++ overview = excluded.overview,
++ memory_json = excluded.memory_json,
++ timeline_json = excluded.timeline_json,
++ message_count = excluded.message_count,
++ window_message_count = excluded.window_message_count,
++ last_message_id = excluded.last_message_id,
++ window_label = excluded.window_label,
++ window_start = excluded.window_start,
++ window_end = excluded.window_end,
++ updated_at = CURRENT_TIMESTAMP`,
++ [scope, subject_id || 0, window, overview, memory_json, timeline_json, message_count,
++ window_message_count, last_message_id, window_label, window_start, window_end]
++ );
++ return _staged(res, 'chat_ai_summaries', [get('SELECT * FROM chat_ai_summaries WHERE scope = ? AND subject_id = ? AND window = ?', [scope, subject_id || 0, window])]);
++ });
++}
++
++function addChatTimelineEvents(scope, subjectId, events) {
++ _assertChatWrites('chat_timeline_events');
++ if (!Array.isArray(events) || !events.length) return _staged(0, 'chat_timeline_events');
++ return transaction(() => {
++ let n = 0;
++ const ids = [];
++ for (const e of events) {
++ if (!e || !e.label || !e.ts) continue;
++ try {
++ const res = run('INSERT OR IGNORE INTO chat_timeline_events (scope, subject_id, ts, label, detail) VALUES (?, ?, ?, ?, ?)',
++ [scope || 'global', subjectId || 0, e.ts, String(e.label).slice(0, 120), String(e.detail || '').slice(0, 400)]);
++ if (res.changes) ids.push(Number(res.lastInsertRowid));
++ n++; // Live counts every event it tried, the duplicates included
++ } catch { /* */ }
++ }
++ return _staged(n, 'chat_timeline_events', ids.map((id) => get('SELECT * FROM chat_timeline_events WHERE id = ?', [id])));
++ });
++}
++
++/**
++ * Live's changes to a staged table it still writes (its capture, relayed over the bridge): the rows
++ * as they are in Live now, or a delete. Applied only while the table is at 'live' — once Chat writes
++ * it, Live's copy is the mirror and never flows back. REPLACE: the authority's row wins over
++ * whatever holds its key or one of its unique columns here.
++ */
++function applyStagedChanges(changes) {
++ const out = { applied: 0, skipped: [] };
++ transaction(() => {
++ for (const c of Array.isArray(changes) ? changes : []) {
++ const pk = STAGED_KEYS[c && c.table];
++ if (!pk) { out.skipped.push({ table: c && c.table, reason: 'not a staged table' }); continue; }
++ if (tableAuthority(c.table) !== 'live') { out.skipped.push({ table: c.table, reason: 'Chat writes this table (table_authority chat)' }); continue; }
++ const have = _columns(c.table);
++ try {
++ if (c.op === 'delete' && c.pk) {
++ run(`DELETE FROM ${c.table} WHERE ${pk.map((k) => `${k} = ?`).join(' AND ')}`, pk.map((k) => c.pk[k]));
++ } else if (c.op === 'upsert' && c.row && pk.every((k) => c.row[k] != null)) {
++ const cols = Object.keys(c.row).filter((k) => have.has(k) && /^[a-z_]+$/.test(k));
++ run(`INSERT OR REPLACE INTO ${c.table} (${cols.join(', ')}) VALUES (${cols.map(() => '?').join(', ')})`, cols.map((k) => c.row[k]));
++ } else { out.skipped.push({ table: c.table, reason: 'bad change' }); continue; }
++ out.applied++;
++ } catch (err) {
++ out.skipped.push({ table: c.table, pk: c.pk || (c.row && Object.fromEntries(pk.map((k) => [k, c.row[k]]))), reason: err.message });
++ }
++ }
++ });
++ for (const c of Array.isArray(changes) ? changes : []) {
++ if (c && (c.table === 'channel_moderators' || c.table === 'channel_moderation_settings')) {
++ const ch = (c.row && c.row.channel_id) || (c.pk && c.pk.channel_id);
++ if (ch) _ctx().invalidateChannel(ch);
++ }
++ }
++ return out;
++}
++
++/**
++ * Hash of a list of rows over `columns` (Live's server/chat/chat-tables-sync.js computes the same):
++ * sha256 of the JSON array of rows, each row the array of its values in column order.
++ */
++function sliceHash(columns, rows) {
++ const body = JSON.stringify(rows.map((r) => columns.map((c) => (r[c] === undefined ? null : r[c]))));
++ return require('crypto').createHash('sha256').update(body).digest('hex');
++}
++
++/**
++ * One slice of a staged table for Live's dual read: the rows whose columns equal `where` (null-safe),
++ * ordered by key → { count, hash, columns, rows (when 50 or fewer) }. `columns` limits the hash to
++ * the ones Live has too.
++ */
++function stagedSlice(table, where = {}, columns = null) {
++ const pk = STAGED_KEYS[table];
++ if (!pk) throw new Error(`${table} is not a staged table`);
++ const have = _columns(table);
++ const keys = Object.keys(where || {});
++ if (keys.some((k) => !have.has(k))) throw new Error(`unknown column in ${table}`);
++ const cols = (Array.isArray(columns) && columns.length ? columns.filter((c) => have.has(c)) : [...have]).sort();
++ const rows = all(`SELECT ${cols.join(', ')} FROM ${table} WHERE ${keys.map((k) => `${k} IS ?`).join(' AND ') || '1'} ORDER BY ${pk.join(', ')}`, keys.map((k) => where[k]));
++ return { count: rows.length, hash: sliceHash(cols, rows), columns: cols, rows: rows.length <= 50 ? rows : undefined };
++}
++
// ── Meta ─────────────────────────────────────────────────────
function getMeta(key) { return get('SELECT value FROM chat_meta WHERE key = ?', [key])?.value ?? null; }
function setMeta(key, value) { return run('INSERT INTO chat_meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value', [key, value == null ? null : String(value)]); }
@@@ -826,6 +1228,7 @@ function setMeta(key, value) { return run('INSERT INTO chat_meta (key, value) VA
module.exports = {
CHAT_TABLES,
STAGED_TABLES,
++ STAGED_KEYS,
getDb,
initDb,
installMirrorTriggers,
@@@ -887,4 +1290,24 @@ module.exports = {
deleteChannelSound,
renameChannelSoundCommand,
updateChannelSoundEmoteRefs,
++ // staged tables (C-04): authority, Live's changes, dual-read slices, and the writes once Chat owns them
++ tableAuthority,
++ setTableAuthority,
++ stagedAuthorities,
++ mirrorPending,
++ applyStagedChanges,
++ stagedSlice,
++ sliceHash,
++ addChannelModerator,
++ removeChannelModerator,
++ upsertChannelModerationSettings,
++ setChannelAlertSound,
++ createEmote,
++ updateEmote,
++ deleteEmote,
++ setEmoteMedia,
++ grantUserTag,
++ revokeUserTag,
++ upsertChatAiSummary,
++ addChatTimelineEvents,
};
ddiff --git a/server/index.js b/server/index.js
iindex a776b26..44648b8 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -36,8 +36,8 @@ async function start() {
const server = http.createServer();
chatServer.init(server);
callServer.init(server);
-- const bridge = createBridge({ chatServer });
const mirror = createMirror({ config });
++ const bridge = createBridge({ chatServer, mirror, config });
const relay = createRelay({ config });
const events = createEventsConsumer({ chatServer, secrets: config.events.secrets });
const { app, handleUpgrade } = createApp({ chatServer, bridge, mirror, relay, events, callServer });
ddiff --git a/server/live-context.js b/server/live-context.js
iindex b1e60a6..8e2f21a 100644
---- a/server/live-context.js
++++ b/server/live-context.js
@@@ -338,11 +338,20 @@ const _policy = new Swr(TTL.policy, async (channelId) => {
};
});
function _policyFor(channelId) { return channelId ? _policy.peek(Number(channelId)) : undefined; }
++// Once Chat writes a staged table (table_authority 'chat', roadmap C-04) its rows here are the truth:
++// read in place, like Live's own readers (an indexed row, no network, no cache to go stale).
++const _chatWrites = (table) => db.tableAuthority(table) === 'chat';
function getChannelModerationSettings(channelId) {
++ if (_chatWrites('channel_moderation_settings')) {
++ return (channelId && db.get('SELECT * FROM channel_moderation_settings WHERE channel_id = ?', [Number(channelId)])) || defaultModerationSettings(channelId);
++ }
const p = _policyFor(channelId);
return (p && p.settings) || defaultModerationSettings(channelId);
}
function isChannelModerator(userId, channelId) {
++ if (_chatWrites('channel_moderators')) {
++ return !!(userId && channelId && db.get('SELECT 1 FROM channel_moderators WHERE user_id = ? AND channel_id = ?', [Number(userId), Number(channelId)]));
++ }
const p = _policyFor(channelId);
return !!(p && p.moderators.has(Number(userId)));
}
@@@ -778,8 +787,31 @@ const effects = {
viewerSnapshot(streamId, count, chatActivity) { _snapshots.push({ stream_id: streamId, viewer_count: count, chat_messages_5m: chatActivity }); },
setUserColor: (userId, color) => effect('user-color', { user_id: userId, color }).then((r) => { db.run('UPDATE ctx_users SET profile_color = ? WHERE id = ?', [color, userId]); return r; }),
ban: (body) => effect('ban', body).then((r) => { invalidateBans(); return r; }),
-- updateChannelModerationSettings: (channelId, fields, actorUserId) => effect('channel-settings', { channel_id: channelId, fields, actor_user_id: actorUserId }).then((r) => { invalidateChannel(channelId); return r; }),
-- setChannelAlertSound: (channelId, kind, url, mime, actorUserId) => effect('alert-sound', { channel_id: channelId, kind, url, mime, actor_user_id: actorUserId }).then((r) => { invalidateChannel(channelId); return r; }),
++ // /slow and alert sounds: written by Live while it owns channel_moderation_settings, here once Chat
++ // does (C-04; the callers already checked the moderator / the channel owner, as Live re-checks).
++ updateChannelModerationSettings: (channelId, fields, actorUserId) => {
++ if (!_chatWrites('channel_moderation_settings')) return effect('channel-settings', { channel_id: channelId, fields, actor_user_id: actorUserId }).then((r) => { invalidateChannel(channelId); return r; });
++ return Promise.resolve().then(() => {
++ if (!channelId || !fields || fields.slow_mode_seconds === undefined) throw new LiveError(400, 'no chat-settable fields');
++ db.upsertChannelModerationSettings(Number(channelId), { slow_mode_seconds: Math.max(0, parseInt(fields.slow_mode_seconds, 10) || 0) });
++ return { ok: true };
++ });
++ },
++ setChannelAlertSound: (channelId, kind, url, mime, actorUserId) => {
++ if (!_chatWrites('channel_moderation_settings')) return effect('alert-sound', { channel_id: channelId, kind, url, mime, actor_user_id: actorUserId }).then((r) => { invalidateChannel(channelId); return r; });
++ return Promise.resolve().then(() => {
++ const file = url ? require('path').resolve(String(url)) : null;
++ if (file) {
++ // Live's rule: alert sounds are files in the shared sounds directory.
++ const fs = require('fs');
++ let inside = false;
++ try { inside = require('path').dirname(fs.realpathSync(file)) === fs.realpathSync(require('path').resolve(config.sounds.path)); } catch { inside = false; }
++ if (!inside) throw new LiveError(400, 'alert sounds live in the sounds directory');
++ }
++ db.setChannelAlertSound(Number(channelId), kind === 'goal' ? 'goal' : 'donation', file, file ? String(mime || 'audio/mpeg') : null);
++ return { ok: true };
++ });
++ },
// One call per real chat line: coins chat bonus, AI viewers, PowerChat relay (Live decides each).
chatMessage: (body) => effect('chat-message', body),
aiModCommand: (channelUserId, streamId, args, opts) => effect('ai/mod-command', { channel_user_id: channelUserId, stream_id: streamId, args, by: opts && opts.by }).then((r) => r.reply),
ddiff --git a/test/import.test.js b/test/import.test.js
iindex b3c54ab..bdd8e87 100644
---- a/test/import.test.js
++++ b/test/import.test.js
@@@ -1,11 +1,13 @@
'use strict';
/**
* scripts/import-from-live.js against a fake Live snapshot built from Live's real table
-- * definitions (test/fixtures/live-chat-schema.sql): counts per table, dry run writes nothing,
-- * ids kept, Network subjects filled in, id headroom, projections seeded, idempotent re-runs,
-- * Chat's edits win on chat tables while staged tables refresh from Live, conflicting rows and
-- * transient-table rows held (never dropped), rows Live wrote after the first pass brought over,
-- * schema drift refused.
++ * definitions (test/fixtures/live-chat-schema.sql): counts per table, a dry run unless --apply (and
++ * it writes nothing), --apply backs Chat's database up first, ids kept, Network subjects filled in,
++ * id headroom, projections seeded, idempotent re-runs, Chat's edits win on chat tables while staged
++ * tables at 'live' are made equal to Live's (refreshed, and pruned of rows Live removed) and staged
++ * tables at 'chat' are only compared, conflicting rows and transient-table rows held (never
++ * dropped), rows Live wrote after the first pass brought over, schema drift refused; and
++ * scripts/parity-check.js --tables (row counts and a content hash per staged table).
*/
const assert = require('assert');
const fs = require('fs');
@@@ -69,10 +71,15 @@ t('build a Live snapshot with every chat table', () => {
live.exec("CREATE TABLE chat_messages_new AS SELECT * FROM chat_messages WHERE id = 101");
});
--t('dry run reports and writes nothing', () => {
-- const r = importer(['--dry-run']);
-- assert.strictEqual(r.code, 0, r.stderr);
-- assert.strictEqual(r.report.dry_run, true);
++t('dry run reports and writes nothing (the default; --dry-run says it too)', () => {
++ for (const args of [[], ['--dry-run']]) {
++ const r = importer(args);
++ assert.strictEqual(r.code, 0, r.stderr);
++ assert.strictEqual(r.report.dry_run, true);
++ assert.strictEqual(r.report.backup, null);
++ }
++ const r = importer([]);
++ assert.strictEqual(r.report.tables.channel_moderators.authority, 'live');
assert.strictEqual(r.report.tables.chat_messages.inserted, 5);
assert.strictEqual(r.report.tables.chat_messages_new.held, 1);
const c = new Database(chatDb, { readonly: true });
@@@ -83,8 +90,12 @@ t('dry run reports and writes nothing', () => {
});
t('import copies every table with ids kept, subjects filled, headroom set', () => {
-- const r = importer(['--headroom', '1000']);
++ const r = importer(['--apply', '--headroom', '1000']);
assert.strictEqual(r.code, 0, r.stderr);
++ assert.ok(r.report.backup && fs.existsSync(r.report.backup), 'backed up first');
++ const b = new Database(r.report.backup, { readonly: true });
++ assert.strictEqual(b.prepare('SELECT COUNT(*) AS n FROM chat_messages').get().n, 0, 'the backup is the database before the run');
++ b.close();
const T = r.report.tables;
const expect = { chat_messages: 5, dm_conversations: 1, dm_participants: 2, dm_messages: 1, dm_blocks: 1, tts_voice_overrides: 1, channel_sounds: 1, relay_users: 1, hidden_relay_users: 1, pending_ip_messages: 1, stream_first_chats: 1, moderation_actions: 1, channel_moderators: 1, channel_moderation_settings: 1, emotes: 1, user_tags: 1, chat_ai_summaries: 1, chat_timeline_events: 1 };
for (const [tbl, n] of Object.entries(expect)) {
@@@ -117,7 +128,8 @@ t('import copies every table with ids kept, subjects filled, headroom set', () =
});
t('re-running is idempotent', () => {
-- const r = importer(['--headroom', '1000']);
++ const r = importer(['--apply', '--no-backup', '--headroom', '1000']);
++ assert.strictEqual(r.report.backup, null);
assert.strictEqual(r.code, 0, r.stderr);
for (const [tbl, v] of Object.entries(r.report.tables)) {
if (v.inserted !== undefined) assert.strictEqual(v.inserted, 0, `${tbl} inserted again`);
@@@ -134,7 +146,7 @@ t('after the cutover: Chat’s edits win, staged tables refresh, conflicts are h
live.prepare('UPDATE channel_moderation_settings SET slow_mode_seconds = 10 WHERE channel_id = 5').run();
live.prepare("INSERT INTO chat_messages (id, user_id, username, message, timestamp) VALUES (106, 2, 'Alice', 'a different 106', '2026-09-03 00:00:00')").run();
live.prepare("INSERT INTO chat_messages (id, user_id, username, message, timestamp) VALUES (107, 2, 'Alice', 'written in Live after the first pass', '2026-09-03 00:00:01')").run();
-- const r = importer(['--headroom', '1000']);
++ const r = importer(['--apply', '--no-backup', '--headroom', '1000']);
assert.strictEqual(r.code, 0, r.stderr);
const T = r.report.tables;
assert.strictEqual(T.chat_messages.chat_kept, 1);
@@@ -153,6 +165,62 @@ t('after the cutover: Chat’s edits win, staged tables refresh, conflicts are h
d.close();
});
++function parity(args) {
++ const r = spawnSync(process.execPath, [path.join(__dirname, '..', 'scripts', 'parity-check.js'), '--tables', '--live-db', liveDb, '--chat-db', chatDb, ...args], {
++ cwd: path.join(__dirname, '..'), encoding: 'utf8', env: { ...process.env, NODE_ENV: 'test' },
++ });
++ return { code: r.status, out: r.stdout, err: r.stderr };
++}
++
++t('parity --tables: after an import every staged table has the same rows and hash', () => {
++ const r = parity([]);
++ assert.strictEqual(r.code, 0, r.out + r.err);
++ for (const tbl of ['channel_moderators', 'channel_moderation_settings', 'emotes', 'user_tags', 'chat_ai_summaries', 'chat_timeline_events']) {
++ assert.match(r.out, new RegExp(`^same\\s+${tbl}\\s+\\(live\\) rows 1\\b`, 'm'), tbl);
++ }
++ assert.match(r.out, /all staged tables are the same/);
++});
++
++t('staged tables at live follow Live: a removed moderator goes, a re-added one comes back; parity sees each step', () => {
++ live.prepare('DELETE FROM channel_moderators WHERE channel_id = 5 AND user_id = 2').run();
++ live.prepare('INSERT INTO channel_moderators (channel_id, user_id, added_by) VALUES (5, 3, 1)').run();
++ let p = parity(['--table', 'channel_moderators']);
++ assert.strictEqual(p.code, 1);
++ assert.match(p.out, /DIFFERENT channel_moderators\s+\(live\) rows 1 ≠ 1/);
++ assert.match(p.out, /only in Live \(1\): \[2\]/);
++ assert.match(p.out, /only in Chat \(1\): \[1\]/);
++ const dry = importer(['--tables', 'channel_moderators']);
++ assert.strictEqual(dry.code, 0, dry.stderr + dry.stdout);
++ assert.deepStrictEqual([dry.report.tables.channel_moderators.pruned, dry.report.tables.channel_moderators.inserted], [1, 1], 'the dry run counts it');
++ assert.strictEqual(parity(['--table', 'channel_moderators']).code, 1, 'and changes nothing');
++ const r = importer(['--apply', '--no-backup', '--tables', 'channel_moderators']);
++ assert.strictEqual(r.code, 0, r.stderr + r.stdout);
++ assert.deepStrictEqual([r.report.tables.channel_moderators.pruned, r.report.tables.channel_moderators.inserted], [1, 1]);
++ p = parity(['--table', 'channel_moderators']);
++ assert.strictEqual(p.code, 0, p.out);
++ // The same user removed and added again under a new id: the old row goes first, no collision.
++ live.prepare('DELETE FROM channel_moderators WHERE user_id = 3').run();
++ live.prepare('INSERT INTO channel_moderators (channel_id, user_id, added_by) VALUES (5, 3, 1)').run();
++ const again = importer(['--apply', '--no-backup', '--tables', 'channel_moderators']);
++ assert.deepStrictEqual([again.report.tables.channel_moderators.pruned, again.report.tables.channel_moderators.inserted, again.report.tables.channel_moderators.held], [1, 1, 0]);
++ assert.strictEqual(parity(['--table', 'channel_moderators']).code, 0);
++});
++
++t('a staged table at chat is only compared: Chat is its authority', () => {
++ const c = new Database(chatDb);
++ c.prepare("UPDATE table_authority SET authority = 'chat' WHERE table_name = 'emotes'").run();
++ c.prepare("UPDATE emotes SET code = 'pogchat' WHERE code = 'pog'").run();
++ c.close();
++ live.prepare("INSERT INTO emotes (user_id, code, url, channel_owner_id) VALUES (1, 'kek', '/e/kek.png', 1)").run();
++ const r = importer(['--apply', '--no-backup', '--tables', 'emotes']);
++ assert.strictEqual(r.code, 0, r.stderr);
++ assert.deepStrictEqual(r.report.tables.emotes, { authority: 'chat', live: 2, identical: 0, differs: 1, live_only: 1, chat_only: 0 });
++ const d = new Database(chatDb, { readonly: true });
++ assert.deepStrictEqual(d.prepare('SELECT code FROM emotes ORDER BY id').all().map((x) => x.code), ['pogchat'], 'nothing written');
++ d.close();
++ assert.match(parity(['--table', 'emotes']).out, /DIFFERENT emotes\s+\(chat\)/);
++});
++
t('schema drift is refused before anything is written', () => {
live.exec('ALTER TABLE chat_messages ADD COLUMN surprise TEXT');
live.prepare("INSERT INTO chat_messages (id, username, message, surprise) VALUES (108, 'x', 'y', 'z')").run();
ddiff --git a/test/staged-tables.test.js b/test/staged-tables.test.js
nnew file mode 100644
iindex 0000000..6c6002a
---- /dev/null
++++ b/test/staged-tables.test.js
@@@ -0,0 +1,239 @@
++'use strict';
++/**
++ * The staged tables (roadmap C-04, docs/staged-tables-cutover.md): table_authority per table, kept
++ * across restarts; at 'live' the staged writes refuse and Live's captured changes keep this copy
++ * current (never mirrored back); the handoff over the bridge (to 'chat' needs the mirror, back to
++ * 'live' waits until the mirror has sent the table's changes); at 'chat' every write Live's writers
++ * make — moderators, settings and alert sounds, emote CRUD with the Media fields, user tags, AI
++ * summaries and timeline events — answers Live's own return value plus the rows, once per key, and
++ * reaches Live through the mirror; Chat's chat server then reads the moderation tables in place and
++ * writes /slow itself; the dual-read slice; scripts/table-authority.js.
++ */
++const assert = require('assert');
++const path = require('path');
++const fs = require('fs');
++const { spawnSync } = require('child_process');
++const { boot, suite } = require('./helpers');
++
++const t = suite('staged-tables');
++let h, BRIDGE, streamer, mod, viewer, channelId, streamId;
++const STAGED = ['channel_moderators', 'channel_moderation_settings', 'emotes', 'user_tags', 'chat_ai_summaries', 'chat_timeline_events'];
++let seq = 0;
++
++const calls = (ops) => h.http('POST', '/internal/live/calls', { token: BRIDGE, body: { boot: 'live-boot-staged', ops: ops.map((o) => ({ seq: ++seq, ...o })) } });
++/** One op, its result (throws with Chat's error when it was refused). */
++async function op(name, args, { key } = {}) {
++ const r = await calls([{ op: name, args, key }]);
++ assert.strictEqual(r.status, 200, r.text);
++ const res = r.body.results[0];
++ if (!res.ok) throw Object.assign(new Error(res.error), { refused: true });
++ return res.result;
++}
++const dbOp = (fn, args, opts) => op('db', [fn, ...args], opts);
++const count = (table, where = '1', params = []) => h.db.get(`SELECT COUNT(*) AS n FROM ${table} WHERE ${where}`, params).n;
++const stagedPending = () => h.db.get(`SELECT COUNT(*) AS n FROM live_mirror_outbox WHERE tbl IN (${STAGED.map(() => '?').join(',')})`, STAGED).n;
++
++t('boot with the Live mirror on', async () => {
++ h = await boot({ env: { LIVE_MIRROR: '1', LIVE_MIRROR_INTERVAL_MS: '3600000' } });
++ BRIDGE = h.serviceToken(['chat.live_bridge.write']);
++ streamer = h.addUser('streamer', { role: 'streamer' });
++ mod = h.addUser('moddy');
++ viewer = h.addUser('viewer');
++ channelId = h.addChannel(streamer.id, { settings: { channel_id: 0, slow_mode_seconds: 99 } });
++ streamId = h.addStream(streamer.id, channelId);
++ await h.ctx.sync();
++});
++
++t('every staged table starts at live; a restart keeps the authority it was handed', async () => {
++ assert.deepStrictEqual(await op('tableAuthority', []), Object.fromEntries(STAGED.map((s) => [s, 'live'])));
++ h.db.setTableAuthority('emotes', 'chat');
++ h.db.initDb();
++ assert.strictEqual(h.db.tableAuthority('emotes'), 'chat', 'initDb never moves a table back');
++ h.db.setTableAuthority('emotes', 'live');
++ assert.strictEqual(h.db.tableAuthority('chat_messages'), 'chat', 'Chat’s own tables stay Chat’s');
++ assert.throws(() => h.db.setTableAuthority('chat_messages', 'live'), /not a staged table/);
++});
++
++t('at live: every staged write refuses, one writer at a time', async () => {
++ const tries = [
++ ['addChannelModerator', [channelId, mod.id, streamer.id]],
++ ['upsertChannelModerationSettings', [channelId, { slow_mode_seconds: 1 }]],
++ ['createEmote', [{ user_id: streamer.id, code: 'nope', url: '/e/nope.png' }]],
++ ['grantUserTag', [viewer.id, 'og']],
++ ['upsertChatAiSummary', [{ scope: 'global', subject_id: 0, window: 'rolling', overview: 'x' }]],
++ ['addChatTimelineEvents', ['global', 0, [{ ts: '2026-09-25 10:00:00', label: 'x' }]]],
++ ];
++ for (const [fn, args] of tries) await assert.rejects(dbOp(fn, args), /written by Live \(table_authority live\)/, fn);
++ for (const s of STAGED) assert.strictEqual(count(s), 0, `${s} untouched`);
++});
++
++t('at live: Live’s changes keep this copy current, idempotently, and never go back to Live', async () => {
++ const changes = [
++ { table: 'channel_moderators', op: 'upsert', row: { id: 7, channel_id: channelId, user_id: mod.id, added_by: streamer.id, created_at: '2026-09-25 10:00:00' } },
++ { table: 'channel_moderation_settings', op: 'upsert', row: { channel_id: channelId, slow_mode_seconds: 4, emote_scale: 120 } },
++ { table: 'emotes', op: 'upsert', row: { id: 3, user_id: streamer.id, code: 'pog', url: '/e/pog.png', channel_owner_id: streamer.id, media_url: null } },
++ { table: 'user_tags', op: 'upsert', row: { id: 5, user_id: viewer.id, tag_id: 'legacy', source: 'migration' } },
++ { table: 'chat_ai_summaries', op: 'upsert', row: { id: 9, scope: 'global', subject_id: 0, window: 'rolling', overview: 'busy' } },
++ { table: 'chat_timeline_events', op: 'upsert', row: { id: 11, scope: 'global', subject_id: 0, ts: '2026-09-25 10:00:00', label: 'raid' } },
++ { table: 'chat_messages', op: 'upsert', row: { id: 1, message: 'not staged' } },
++ ];
++ let r = await op('stagedApply', [changes]);
++ assert.strictEqual(r.applied, 6);
++ assert.deepStrictEqual(r.skipped.map((s) => s.reason), ['not a staged table']);
++ r = await op('stagedApply', [changes.slice(0, 6)]);
++ assert.strictEqual(r.applied, 6, 'again: the same rows');
++ for (const s of STAGED) assert.strictEqual(count(s), 1, s);
++ assert.strictEqual(h.db.get('SELECT emote_scale FROM channel_moderation_settings WHERE channel_id = ?', [channelId]).emote_scale, 120);
++ // Removed in Live and added again under a new id: the new row replaces the old one.
++ r = await op('stagedApply', [[{ table: 'channel_moderators', op: 'upsert', row: { id: 8, channel_id: channelId, user_id: mod.id, added_by: streamer.id } }]]);
++ assert.deepStrictEqual(h.db.all('SELECT id FROM channel_moderators').map((x) => x.id), [8]);
++ r = await op('stagedApply', [[{ table: 'user_tags', op: 'delete', pk: { id: 5 } }]]);
++ assert.strictEqual(count('user_tags'), 0);
++ assert.strictEqual(stagedPending(), 0, 'Live’s own rows are never mirrored back');
++ // Chat still reads the moderation tables through Live while Live writes them.
++ await h.ctx.ensurePolicy(channelId);
++ assert.strictEqual(h.ctx.getChannelModerationSettings(channelId).slow_mode_seconds, 99);
++ assert.strictEqual(h.ctx.isChannelModerator(mod.id, channelId), false);
++});
++
++t('the dual-read slice: count and hash over the columns asked for, rows when few', async () => {
++ const s = await op('stagedSlice', ['channel_moderators', { channel_id: channelId }, ['id', 'channel_id', 'user_id', 'added_by', 'created_at', 'not_a_column']]);
++ assert.strictEqual(s.count, 1);
++ assert.deepStrictEqual(s.columns, ['added_by', 'channel_id', 'created_at', 'id', 'user_id']);
++ assert.strictEqual(s.hash, h.db.sliceHash(s.columns, s.rows));
++ assert.strictEqual((await op('stagedSlice', ['emotes', { channel_owner_id: null }])).count, 0, 'null-safe equality');
++ await assert.rejects(op('stagedSlice', ['emotes', { 'code; DROP TABLE emotes': 1 }]), /unknown column/);
++ await assert.rejects(op('stagedSlice', ['chat_messages', {}]), /not a staged table/);
++});
++
++t('the handoff: to chat needs the Live mirror', async () => {
++ const { handOver } = require('../server/bridge/live-bridge');
++ await assert.rejects(handOver('emotes', 'chat', { mirror: h.mirrorRelay, config: { live: { mirror: false } } }), /LIVE_MIRROR is off/);
++ assert.strictEqual(h.db.tableAuthority('emotes'), 'live');
++ await assert.rejects(op('setTableAuthority', ['chat_messages', 'live']), /not a staged table/);
++ for (const s of STAGED) assert.deepStrictEqual(await op('setTableAuthority', [s, 'chat']), { table: s, authority: 'chat', mirror_pending: 0 });
++ assert.strictEqual((await h.http('GET', '/ready')).body.table_authority.emotes, 'chat');
++});
++
++let emoteId;
++t('at chat: moderators and settings — Live’s return values, the rows, once per key', async () => {
++ let r = await dbOp('addChannelModerator', [channelId, viewer.id, streamer.id], { key: 'live:k1' });
++ assert.strictEqual(r.value.changes, 1);
++ assert.deepStrictEqual(r.mirror.map((c) => [c.table, c.op, c.row.user_id]), [['channel_moderators', 'upsert', viewer.id]]);
++ r = await dbOp('addChannelModerator', [channelId, viewer.id, streamer.id], { key: 'live:k1' });
++ assert.strictEqual(count('channel_moderators', 'user_id = ?', [viewer.id]), 1, 'a retried key applies once');
++ r = await dbOp('removeChannelModerator', [channelId, viewer.id]);
++ assert.strictEqual(r.value.changes, 1);
++ assert.strictEqual(r.mirror[0].op, 'delete');
++ assert.strictEqual(count('channel_moderators', 'user_id = ?', [viewer.id]), 0);
++ r = await dbOp('upsertChannelModerationSettings', [channelId, { slow_mode_seconds: 7, emote_scale: 999, tts_max_length: 5, followers_only: true }]);
++ assert.deepStrictEqual([r.value.slow_mode_seconds, r.value.emote_scale, r.value.tts_max_length, r.value.followers_only], [7, 300, 10, 1], 'Live’s clamps');
++ r = await dbOp('upsertChannelModerationSettings', [channelId + 1000, { slow_mode_seconds: 2 }]);
++ assert.deepStrictEqual([r.value.channel_id, r.value.slow_mode_seconds, r.value.allow_anonymous, r.value.max_message_length], [channelId + 1000, 2, 1, 500], 'a new row with Live’s defaults');
++ r = await dbOp('setChannelAlertSound', [channelId, 'goal', '/sounds/goal.mp3', 'audio/mpeg']);
++ assert.strictEqual(r.mirror[0].row.goal_sound_url, '/sounds/goal.mp3');
++ assert.strictEqual(r.mirror[0].row.slow_mode_seconds, 7, 'the whole row travels');
++});
++
++t('at chat: emote CRUD with the Media fields, user tags, AI summaries and timeline', async () => {
++ let r = await dbOp('createEmote', [{ user_id: streamer.id, code: 'hype', url: '/data/emotes/hype.png', animated: true, channel_owner_id: streamer.id, size: 5 }]);
++ emoteId = r.value.lastInsertRowid;
++ assert.ok(emoteId > 3);
++ assert.deepStrictEqual([r.mirror[0].row.code, r.mirror[0].row.animated, r.mirror[0].row.size], ['hype', 1, 25]);
++ await assert.rejects(dbOp('createEmote', [{ user_id: streamer.id, code: 'hype', url: '/x.png', channel_owner_id: streamer.id }]), /UNIQUE/, 'one code per channel');
++ r = await dbOp('updateEmote', [emoteId, { code: 'hype2', size: 1000 }]);
++ assert.deepStrictEqual([r.value.changes, r.mirror[0].row.code, r.mirror[0].row.size], [1, 'hype2', 400]);
++ assert.deepStrictEqual((await dbOp('updateEmote', [emoteId, {}])).value, { changes: 0 });
++ r = await dbOp('setEmoteMedia', [emoteId, 'https://openvibe.media/a/77', 77]);
++ assert.deepStrictEqual([r.mirror[0].row.media_url, r.mirror[0].row.media_asset_id], ['https://openvibe.media/a/77', 77]);
++ r = await dbOp('deleteEmote', [3]);
++ assert.deepStrictEqual(r.mirror, [{ table: 'emotes', op: 'delete', pk: { id: 3 } }]);
++ r = await dbOp('grantUserTag', [viewer.id, 'founder', 'grant']);
++ assert.strictEqual(r.mirror[0].row.source, 'grant');
++ await dbOp('grantUserTag', [viewer.id, 'founder']);
++ assert.strictEqual(count('user_tags'), 1, 'a tag is owned once');
++ r = await dbOp('revokeUserTag', [viewer.id, 'founder']);
++ assert.strictEqual(r.value.changes, 1);
++ r = await dbOp('upsertChatAiSummary', [{ scope: 'user', subject_id: viewer.id, window: 'rolling', overview: 'first', last_message_id: 10 }]);
++ r = await dbOp('upsertChatAiSummary', [{ scope: 'user', subject_id: viewer.id, window: 'rolling', overview: 'second', last_message_id: 20 }]);
++ assert.deepStrictEqual([r.mirror[0].row.overview, r.mirror[0].row.last_message_id], ['second', 20]);
++ assert.strictEqual(count('chat_ai_summaries', "scope = 'user'"), 1);
++ r = await dbOp('addChatTimelineEvents', ['global', 0, [{ ts: '2026-09-25 11:00:00', label: 'clip' }, { ts: '2026-09-25 11:00:00', label: 'clip' }, { label: 'no ts' }]]);
++ assert.strictEqual(r.value, 2, 'Live counts the duplicate it tried');
++ assert.strictEqual(r.mirror.length, 1);
++ // Live's changes no longer apply to a table Chat writes.
++ const refused = await op('stagedApply', [[{ table: 'emotes', op: 'delete', pk: { id: emoteId } }]]);
++ assert.deepStrictEqual([refused.applied, refused.skipped[0].reason], [0, 'Chat writes this table (table_authority chat)']);
++ assert.strictEqual(count('emotes', 'id = ?', [emoteId]), 1);
++});
++
++t('at chat: every change reaches Live through the mirror, newest state per row', async () => {
++ assert.ok(stagedPending() > 0);
++ await h.mirrorRelay.flush();
++ assert.strictEqual(stagedPending(), 0);
++ const got = (table) => h.live.mirror.filter((c) => c.table === table);
++ assert.deepStrictEqual(got('emotes').find((c) => c.row && c.row.id === emoteId).row.code, 'hype2');
++ assert.ok(got('emotes').some((c) => c.op === 'delete' && c.pk.id === 3));
++ assert.ok(got('channel_moderators').some((c) => c.op === 'delete'));
++ assert.strictEqual(got('channel_moderation_settings').find((c) => c.row.channel_id === channelId).row.goal_sound_url, '/sounds/goal.mp3');
++ for (const s of ['user_tags', 'chat_ai_summaries', 'chat_timeline_events']) assert.ok(got(s).length >= 1, s);
++});
++
++t('at chat: Chat’s chat server reads the moderation tables in place and persists /slow itself', async () => {
++ await dbOp('addChannelModerator', [channelId, mod.id, streamer.id]);
++ assert.strictEqual(h.ctx.isChannelModerator(mod.id, channelId), true, 'Live’s policy lists no moderators; the local row counts');
++ assert.strictEqual(h.ctx.getChannelModerationSettings(channelId).slow_mode_seconds, 7, 'the local row, not Live’s 99');
++ assert.strictEqual(h.ctx.getChannelModerationSettings(424242).allow_anonymous, 1, 'defaults when there is no row');
++ const effectsBefore = h.live.effects.length;
++ const ws = await h.ws({ ip: '198.51.100.60', token: mod.token, stream: streamId });
++ ws.sendJson({ type: 'join', streamId, token: mod.token });
++ await ws.next((m) => m.type === 'auth');
++ ws.sendJson({ type: 'chat', message: '/slow 12' });
++ assert.strictEqual((await ws.next((m) => m.type === 'slowmode')).seconds, 12, 'a moderator by the local table');
++ await h.sleep(100);
++ assert.strictEqual(h.db.get('SELECT slow_mode_seconds FROM channel_moderation_settings WHERE channel_id = ?', [channelId]).slow_mode_seconds, 12);
++ assert.ok(!h.live.effects.slice(effectsBefore).some((e) => e.name === 'channel-settings'), 'Live is not asked to write it');
++ ws.close();
++ // Alert sounds: files in the sounds directory only, written here.
++ await assert.rejects(h.ctx.effects.setChannelAlertSound(channelId, 'donation', '/etc/hostname', 'audio/mpeg', streamer.id), /sounds directory/);
++ const snd = path.join(h.tmp, 'sounds', 'alert.mp3');
++ fs.writeFileSync(snd, 'ID3');
++ await h.ctx.effects.setChannelAlertSound(channelId, 'donation', snd, 'audio/mpeg', streamer.id);
++ assert.strictEqual(h.ctx.getChannelAlertSoundsByUser(streamer.id).donation_sound_url, snd);
++ assert.ok(!h.live.effects.slice(effectsBefore).some((e) => e.name === 'alert-sound'));
++});
++
++t('back to live: waits for the mirror, refused while Live refuses it, then Live writes again', async () => {
++ await dbOp('addChannelModerator', [channelId, viewer.id, streamer.id]);
++ assert.ok(h.db.mirrorPending('channel_moderators') > 0);
++ h.live.mirrorStatus = 409;
++ await assert.rejects(op('setTableAuthority', ['channel_moderators', 'live']), /not in Live yet/);
++ assert.strictEqual(h.db.tableAuthority('channel_moderators'), 'chat', 'nothing moved');
++ h.live.mirrorStatus = 200;
++ assert.deepStrictEqual(await op('setTableAuthority', ['channel_moderators', 'live']), { table: 'channel_moderators', authority: 'live', mirror_pending: 0 });
++ assert.ok(h.live.mirror.some((c) => c.table === 'channel_moderators' && c.row && c.row.user_id === viewer.id), 'Chat’s newest row reached Live first');
++ await assert.rejects(dbOp('addChannelModerator', [channelId, streamer.id, streamer.id]), /written by Live/);
++ assert.strictEqual((await op('stagedApply', [[{ table: 'channel_moderators', op: 'delete', pk: { id: 8 } }]])).applied, 1, 'Live’s changes apply again');
++ assert.strictEqual(stagedPending(), 0, 'and are not mirrored back');
++});
++
++t('scripts/table-authority.js: status; set only with --force; never back to live with changes queued', async () => {
++ const run = (...args) => spawnSync(process.execPath, [path.join(__dirname, '..', 'scripts', 'table-authority.js'), ...args], { encoding: 'utf8', env: { ...process.env } });
++ let r = run();
++ assert.strictEqual(r.status, 0, r.stderr);
++ const st = JSON.parse(r.stdout).tables;
++ assert.deepStrictEqual([st.channel_moderators.authority, st.emotes.authority], ['live', 'chat']);
++ assert.strictEqual(run('set', 'emotes', 'live').status, 2, 'needs --force');
++ await dbOp('createEmote', [{ user_id: viewer.id, code: 'late', url: '/e/late.png' }]);
++ r = run('set', 'emotes', 'live', '--force');
++ assert.strictEqual(r.status, 2);
++ assert.match(r.stderr, /have not reached Live yet/);
++ await h.mirrorRelay.flush();
++ r = run('set', 'emotes', 'live', '--force');
++ assert.strictEqual(r.status, 0, r.stderr);
++ assert.deepStrictEqual(JSON.parse(r.stdout), { table: 'emotes', before: 'chat', after: 'live' });
++ await h.sleep(1100); // the service's authority cache
++ await assert.rejects(dbOp('createEmote', [{ user_id: viewer.id, code: 'later', url: '/e/later.png' }]), /written by Live/);
++});
++
++t.run(async () => { if (h) await h.close(); });