openvibers/openvibe.chat Commits

Retire the Live chat bridge receiver and bridge deploy notices

OP OpenVibers committed 4 Oct 2026 · a861c94c20 · 23 files, +145 −741

Asked: T3 step 1, first half (J3a) — OpenVibe.Chat: retire the Live chat bridge receiver and the C-84 bridge entry of deploy notices. Gate met: Live's J2 is merged AND deployed (Live #20 routed every producer through Chat's ingress; Live #21 c1bd342 'Deliver chat only through Chat's ingress and drop the b…
Harness job chat-t3-step-1-first-half-221009 (plan T3).

Changes

ddiff --git a/README.md b/README.md
iindex 50869b3..4c74406 100644
---- a/README.md
++++ b/README.md
@@@ -57,8 +57,9 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
waits on a network read. Live pushes invalidations when it changes cached data.
- **Live's side** is `docs/live-patch.diff` (applies to Live `main` with `git apply`):
`/internal/chat-context/*` and `/internal/chat-effects/*`, and `CHAT_AUTHORITY=chat`, which stops
-- Live's chat server and routes and turns `require('./chat/chat-server')` into a proxy that forwards
-- Live's own chat calls here (`POST /internal/live/calls`).
++ Live's chat server and routes. Live's own chat calls reach Chat through the Chat-owned ingress
++ (`/internal/chat/*`, `docs/chat-ingress.md`); the bridge proxy at `POST /internal/live/calls` is
++ retired (T3 J3a).
- **Own database on PostgreSQL** (ADR-035, plan T3) with Live's tables and ids, the Network subject on new
rows, a transactional events outbox and a read mirror back into Live. `server/db/database.js` serves
through `openvibe-sdk/db`: production needs `DATABASE_URL` (the runtime role, DML only, through
@@@ -76,8 +77,8 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
a row of Chat's tables (`import_hold`), reports counts per table.
- **The six chat tables** (`channel_moderators`, `channel_moderation_settings`, `emotes`, `user_tags`,
`chat_ai_summaries`, `chat_timeline_events`): Chat's own since the C-04 cutover — Chat is their only
-- writer, there is no authority switch any more. Live's current release's writers still call Chat over
-- the bridge (op `db`) until Live N+1 is deployed.
++ writer, there is no authority switch any more. Live's writers go through Chat's APIs below and the
++ Chat-owned ingress (`docs/chat-ingress.md`); the Live bridge they used before is retired (T3 J3a).
- **The features on those tables, served by Chat** (plan T3, plan step 1) — built here so Live can
stop touching them at the flip:
- `GET /api/emotes/global`, `/channel/:userId`, `/mine`, `/defaults`, `/sources` (GET/PUT),
@@@ -95,8 +96,8 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
`GET /internal/moderation/channels/:channelId/emote-count` — loopback + service token,
capability `chat.moderation.read` (`chat.channel-moderation-result@1`,
`chat.moderated-channels-result@1`, `chat.emote-count-result@1`). Live caches each answer 30 s.
-- - Alert sounds: the bridge op `playAlertSound [streamerId, streamId, kind]` makes Chat resolve the
-- channel's own settings row, read the clip and broadcast it, so Live only names the alert.
++ - Alert sounds: `playAlertSound(chatServer, streamerId, streamId, kind)` (`server/chat/alert-sounds.js`)
++ resolves the channel's own settings row, reads the clip and broadcasts it to the channel room.
- **The chat-AI summaries** (plan T3 step 2, decision 5) — Live's `server/ai/chat-ai.js` job moved
here: a background poller (off unless `CHAT_AI_ENABLED=1`) that folds Chat's own messages into
rolling global/per-user/per-relay/per-anon insights and the append-only timeline, writing
@@@ -112,10 +113,10 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
| Concern | Where it lives | How Chat reaches it |
| --- | --- | --- |
--| Messages, DMs, channel sounds, TTS voice overrides, relay users, first chats, IP-approval queue, moderation log | **Chat** (authority from the cutover) | its own database (`CHAT_DB_PATH` SQLite today; PostgreSQL per plan T3 — `migrations/`, `scripts/import-sqlite-to-pg.js`); Live keeps a read mirror (`POST /internal/chat-effects/mirror`) |
++| Messages, DMs, channel sounds, TTS voice overrides, relay users, first chats, IP-approval queue, moderation log | **Chat** (authority from the cutover) | its own database on PostgreSQL since the 2026-09-23 cutover (`DATABASE_URL` through `openvibe-sdk/db`, `migrations/`, the one-time `scripts/import-sqlite-to-pg.js`; `CHAT_DB_PATH` is read only by that importer); Live keeps a read mirror (`POST /internal/chat-effects/mirror`) |
| Accounts, roles, streams, channels, bans, IP approvals, follows, cosmetics, tags, site settings | **Live** (Network for identity) | `server/live-context.js` → `GET/POST /internal/chat-context/*` (service token, `live.chat_context.read`) |
| Coins, AI viewers, arena, media queue, hardware, pastes, translation, PowerChat, notifications | **Live** | `server/live-context.js` → `POST /internal/chat-effects/*` (`live.chat_effects.write`) |
--| Live's own chat pushes and writes (AI viewers, relays, donations, `/api/mod`, recaps, calls) | **Live → Chat** | `POST /internal/live/calls` (`chat.live_bridge.write`), presence `GET /internal/live/presence` (`chat.presence.read`) |
++| Live's own chat pushes and writes (AI viewers, relays, donations, `/api/mod`, recaps, calls) | **Live → Chat** | the Chat-owned ingress `/internal/chat/*` (`docs/chat-ingress.md`), presence `GET /internal/chat/presence` (`chat.presence.read`); the bridge `/internal/live/*` is retired (T3 J3a) |
| The six chat tables (moderators, moderation settings, emotes, tags, AI summaries, timeline) | **Chat** (C-04 done) | Chat's own database; Live reads through `GET /internal/moderation/*` (`chat.moderation.read`), 30 s cached |
| Emote image bytes | **OpenVibe.Media** (namespace `chat`) | `server/media/client.js` (openvibe-sdk `createObjectsClient`), Chat's service token (`media.object.upload` / `.delete`); the row keeps `media_url` + `media_asset_id` |
| Chat-AI summaries (global / per-chatter insight + timeline) | **OpenVibe.AI** (namespace `chat.*`) | `server/ai/client.js` (openvibe-sdk `createAiClient`), Chat's service token (`ai.run.create` / `ai.run.read`); workflows `chat.global` / `chat.profile`; extractive fallback when AI does not answer |
@@@ -161,16 +162,15 @@ Chat subscribes (consumer `chat`) to these topics (and `network.user.token_valid
| Event | From | What Chat does |
| --- | --- | --- |
--| `live.release.deployed` | Live (`server/events/release-events.js`), subject `{ type: release, id: <head commit> }` | stores or folds the deploy card in global chat, exactly as the bridge op `deployNotice` does (`server/chat/deploy-notice.js`: one rolling card, folded while nobody has spoken in any room and within 3 hours; the broadcast carries the row id; late joiners get it once) |
++| `live.release.deployed` | Live (`server/events/release-events.js`), subject `{ type: release, id: <head commit> }` | stores or folds the deploy card in global chat (`server/chat/deploy-notice.js`: one rolling card, folded while nobody has spoken in any room and within 3 hours; the broadcast carries the row id; late joiners get it once) |
| `network.module.updated` | Network (`server/identity/module-events.js`) | for `chat.preferences`, a revision newer than the cached copy drops it (`prefs.handleEvent()`); an older or equal revision and other namespaces change nothing |
| `network.block.changed` | Network (`server/identity/blocks.js`, platform blocks) | keeps the newest revision per (blocker, blocked) in `network_blocks` (`server/chat/network-blocks.js`); while a block is active neither person can start a DM with the other, message them in a 1:1, add them to a group or call them, and the DM user search hides them, exactly like `dm_blocks` (same errors) |
--- **One card per deploy, whichever path comes first** (compatibility register C-84). Until the bridge
-- op is removed, Live sends each deploy twice: `deployNotice` over the bridge (commits newest first,
-- so the first is the head) and the event. Both claim the head commit in `deploy_releases` in the
-- transaction that stores or folds the card; the second finds it claimed and changes nothing but its
-- `bridge_at` / `event_at`. A retried bridge op, a redelivered event and Live re-announcing after a
-- crash are the same card. A redeploy of a head already announced (a rollback to it) says nothing.
++- **One card per deploy** (compatibility register C-84, closed). The event is the only path: the
++ bridge op `deployNotice` is retired (T3 J3a); `deploy_releases.first_via` / `bridge_at` keep what
++ it recorded. The event claims the head commit in `deploy_releases` in the transaction that stores
++ or folds the card; a later delivery for that head finds it claimed and changes nothing but its
++ `event_at`. A redelivered event and Live re-announcing after a crash are the same card. A redeploy of a head already announced (a rollback to it) says nothing.
A release event older than 6 hours (an operator replay) is `ignored:stale`.
- **Exactly once.** The openvibe-sdk inbox (`chat_event_inbox`, pruned after 35 days) claims
`(chat, event_id)` in the same transaction as the change, and broadcasts run only after it
@@@ -223,8 +223,8 @@ cutover** (`CHAT_CALLS=1`, `docs/calls-cutover.md`):
`GET/PUT /api/streams/:id/call` (the streamer, checked through `live-context`).
- **Live's stream hooks** (`server/calls/internal.js`, Live `CALLS_AUTHORITY=chat`):
`POST /internal/calls/stream-channel { stream_id, mode, user_id }` and
-- `DELETE /internal/calls/stream-channel/:streamId` — the bridge's service token and capability
-- (`chat.live_bridge.write`), loopback only.
++ `DELETE /internal/calls/stream-channel/:streamId` — a Live service token with capability
++ `chat.live_bridge.write` (the name is from the retired chat bridge; Live holds the grant), loopback only.
- **Lifecycle** (`server/calls/lifecycle.js`, table `calls`): a ring is `direct`, `pending → ringing →
active → ended`, or `missed` (no answer within `CALL_RING_TIMEOUT_MS`, 45 s; the caller is told),
`declined` (declined or busy) or `failed` (the invite could not be delivered, with the reason); a
@@@ -404,8 +404,8 @@ rollback — is `docs/cutover.md`.
The REST API also limits who calls it (`server/net/actor-limits.js`, openvibe-sdk/limits, roadmap
WS-R task 4). A person counts as `user:usr_…` whether they call with their Network token or an `hbt_`
API token (a bot counts as its owner); anyone else by their address. Browsers call these routes
--themselves, so no service speaks for many visitors here; Live's service calls (the bridge at
--`/internal/live`, the stream hooks at `/internal/calls`) carry every viewer's chat and are never
++themselves, so no service speaks for many visitors here; Live's service calls (the ingress at
++`/internal/chat`, the stream hooks at `/internal/calls`) carry every viewer's chat and are never
limited. The per-address `/api/` limit and the chat flood controls (the socket's, DMs 10 a minute and
5 new conversations an hour, rooms 6 every 10 s, 6 rings a minute) stay. Past a limit: `429`
problem+json `rate_limited` with `Retry-After`, one `[Limits]` log line and
@@@ -426,7 +426,7 @@ problem+json `rate_limited` with `Retry-After`, one `[Limits]` log line and
| Calls: voice channel create and delete, stream call switch 20 / 200; ring 20 / 200; answer 30 / 300 | as listed |
| Chat settings save (`PUT /api/chat/preferences` and the other modules) | 30 / 300 |
--Never limited: `/health`, `/ready`, `/release.json`, `/metrics`, `/internal/*` (the bridge, the call
++Never limited: `/health`, `/ready`, `/release.json`, `/metrics`, `/internal/*` (the ingress, the call
hooks and the signed Events deliveries, which carry account deletions and merges), `/ws/chat` and
`/ws/call`, the openvibe.chat pages, and the media files players fetch (`/api/tts/audio/…`,
`/api/sounds/file/…`, `/api/emotes/file/…`), which only the per-address limit bounds. `test/actor-limits.test.js`.
@@@ -439,7 +439,7 @@ server/app.js Live's guards for these routes: CORS, /api rate limit
server/live-context.js the only module that talks to Live (interface in its header)
server/chat/ moved from Live: chat-server, dm, dm-routes, routes, history-store, tts-*, sounds-*, soundboard, moderation-utils, word-filter, deploy-notice; plus the T3 APIs: emotes-routes, channel-mod-routes, internal-moderation (Live's read API), alert-sounds
server/auth/ token resolution through Live; the chat subset of Live's permissions
--server/bridge/ Live → Chat calls + presence (live-bridge.js); Chat → Live read mirror (live-mirror.js)
++server/bridge/ Chat → Live read mirror (live-mirror.js)
server/events/outbox.js events.event-envelope@1 outbox and relay
server/events/consumer.js POST /internal/events: Chat's Events subscriptions (live.release.deployed, network.module.updated)
server/events/subscriptions.js creates them at boot when missing; list/disable/enable for scripts/subscribe-events.js
@@@ -475,13 +475,13 @@ queue are still Live's and reached through `live-context`. Calls are ported with
Introduced here and registered in `openvibe-contracts` v0.13.0:
`chat.live_bridge.write`, `chat.presence.read` (owner chat) and `live.chat_context.read`,
--`live.chat_effects.write`, `live.chat_mirror.write` (owner live). Enforced here as well:
--`chat.message.send` — a service or app principal whose `/internal/live/calls` op sends a chat
--message (a `saveChatMessage` write, the deploy notice, or a pushed `chat`/`dm` frame) must hold it
--beside `chat.live_bridge.write`; without it that op is refused (`capability.denied`) and the rest of
--the batch runs. It is owned by `chat` and listed in the chat manifest since `openvibe-contracts`
--v0.30.2 (so is the `chat.preferences` user module since v0.32.0); the check names it literally, so
--the contracts check enforces it. Planned families:
++`live.chat_effects.write`, `live.chat_mirror.write` (owner live). `chat.live_bridge.write` now only
++guards Live's stream hooks for calls (`/internal/calls/stream-channel`) and `chat.presence.read` the
++presence snapshot (`GET /internal/chat/presence`); the bridge routes they were made for are retired
++(T3 J3a). Enforced here as well: `chat.message.send` (`POST /internal/chat/messages`). It is owned
++by `chat` and listed in the chat manifest since `openvibe-contracts` v0.30.2 (so is the
++`chat.preferences` user module since v0.32.0); the check names it literally, so the contracts check
++enforces it. Planned families:
`chat.room.*`, `chat.message.*`, `chat.dm.*`, `chat.moderation.*`, `chat.tts.*`, `chat.call.*`.
Events: `chat.message.created`, `chat.message.deleted`, `chat.dm.created`, `chat.moderation.action`,
@@@ -514,7 +514,7 @@ Called elsewhere, as the service principal `chat` ([server/net/service-auth.js](
## Acceptance (must be true before "done")
- stream/global/DM histories preserved on import; old Live URLs and WS messages keep working through the adapter — *done on production data: 70,860 messages, 11 conversations and 1,522 DMs imported with 0 held (two passes, identical), 15/15 read paths identical at the cutover*
--- restart Live without losing Chat; restart Chat's delivery plane and resume persisted messages — *seen in production (Live restarted several times on 2026-09-23 while Chat stayed up; messages and a queued outbox row survived Chat restarts); `test/restart-resume.test.js` restarts Chat as a real process (SIGTERM, new process on the same database) and proves stream, global and channel readers resume from their `after_id` cursor with no gap and no duplicate, including Live bridge placeholders that straddle the restart (kept in `bridge_refs`)*
++- restart Live without losing Chat; restart Chat's delivery plane and resume persisted messages — *seen in production (Live restarted several times on 2026-09-23 while Chat stayed up; messages and a queued outbox row survived Chat restarts); `test/restart-resume.test.js` restarts Chat as a real process (SIGTERM, new process on the same database) and proves stream, global and channel readers resume from their `after_id` cursor with no gap and no duplicate, including an ingress message Live sends again with its key after the restart (one row, the first id)*
- a call row without a working signalling/media path is not parity — *Live's `/ws/call` protocol and call routes run in Chat with a `calls` row per call (`test/calls.test.js`: two signed-in sockets exchange offer/answer/ICE, limits, kick/ban, ringing → active/declined/missed/failed, stream channels); not switched over yet (`docs/calls-cutover.md`)*
- a paid TTS request is never duplicated by a retry — *forwarded writes are applied once per idempotency key; paid TTS does not exist yet*
- after a reconnect, deleted and blocked state converge; browser parity (join, send, DM, `/tts`, moderation, popout) — *`scripts/parity.js` and `test/parity.test.js` (`docs/parity.md`): a reader that was away converges on what a connected reader saw (cursor reads carry `deleted_ids`); gaps: sub-only mode does not exist, public chat does not apply blocks*
ddiff --git a/docs/capabilities-proposal/chat.live_bridge.write.json b/docs/capabilities-proposal/chat.live_bridge.write.json
iindex 7929c30..e61536b 100644
---- a/docs/capabilities-proposal/chat.live_bridge.write.json
++++ b/docs/capabilities-proposal/chat.live_bridge.write.json
@@@ -4,7 +4,7 @@
"owner": "chat",
"status": "planned",
"visibility": "internal",
-- "description": "Live's remaining modules (AI viewers, chat relays, donations and alerts, /api/mod, recaps, news, calls, emotes, deploy notices) push to chat clients and write chat rows through OpenVibe.Chat: ChatServer calls in order (broadcasts, DMs with the participant check, TTS, channel sounds, disconnects, user updates, cache invalidations) and allow-listed chat-table writes, applied once per idempotency key, with placeholder ids mapped per Live boot. Transitional: shrinks as those modules move out of Live.",
++ "description": "Live's stream lifecycle hooks for calls (CALLS_AUTHORITY=chat): create, change or end a stream's voice channel. Named for the Live chat bridge (POST /internal/live/calls), retired in T3 J3a; Live already holds this grant, so the call hooks kept it.",
"permissions": [
"chat:write"
],
@@@ -12,11 +12,9 @@
"none"
],
"quotaClass": "internal-write",
-- "events": [
-- "chat.message.created",
-- "chat.moderation.action"
-- ],
++ "events": [],
"implementedBy": [
-- "POST /internal/live/calls"
++ "POST /internal/calls/stream-channel",
++ "DELETE /internal/calls/stream-channel/:streamId"
]
}
ddiff --git a/docs/capabilities-proposal/chat.presence.read.json b/docs/capabilities-proposal/chat.presence.read.json
iindex c1f632d..6f25015 100644
---- a/docs/capabilities-proposal/chat.presence.read.json
++++ b/docs/capabilities-proposal/chat.presence.read.json
@@@ -14,7 +14,6 @@
"quotaClass": "internal-read",
"events": [],
"implementedBy": [
-- "GET /internal/live/presence",
"GET /internal/chat/presence"
]
}
ddiff --git a/docs/chat-ingress.md b/docs/chat-ingress.md
iindex a6dc591..0761a44 100644
---- a/docs/chat-ingress.md
++++ b/docs/chat-ingress.md
@@@ -1,7 +1,7 @@
# Chat-owned internal ingress (T3 J2 prerequisite)
--These routes accept an RS256 service token for `openvibe.chat` on loopback only. The existing
--`/internal/live/*` bridge remains available during the Live rollout. All POST bodies require a
++These routes accept an RS256 service token for `openvibe.chat` on loopback only. They replaced the
++`/internal/live/*` bridge, which is retired (T3 J3a; Live #20/#21). All POST bodies require a
stable `key` (1–160 letters, digits, `:._-`). A retry with the same principal, route and body
returns the first result. Reusing the key with another body returns 409. If delivery fails after
the database write, retry the same key: Chat retries the broadcast without writing another row.
@@@ -18,7 +18,7 @@ a stream that does not exist is 404: do not retry a 4xx.
| `POST /internal/chat/events` | `chat.event.publish` | `{key,target:{kind,id?},frame:{type,...}}`, returns `{ok:true}`. Target kinds: `stream`, `channel`, `owner-streams`, `user` (positive `id`), `global`, `all` (no `id`). Only the event frame types allowlisted in `internal-ingress.js` are accepted; `dm` and delete frames are refused, and `chat` only as the news card. |
| `POST /internal/chat/moderation` | `chat.moderation.write` | `{key,action,...}`; actions: `delete-message`, `delete-user-messages`, `delete-anon-messages`, `delete-relay-messages`, `delete-by-range`, `review-pending-ip`, `approve-ip-messages`, `deny-ip-messages`, `relay-hide`, `relay-unhide`, `relay-record`, `tts-voice-override`, `disconnect`, `log`. Deletes return `ids` and broadcast `delete-messages` to every affected surface. |
| `POST /internal/chat/invalidate` | `chat.cache.invalidate` | `{key,user?,user_data?,approvals?,bans?,channel?}`; a cache hint, returns `{ok:true}`. Include a typed `user_data` object when Live edited the account, so Chat updates sockets and stored message names. |
--| `GET /internal/chat/presence` | `chat.presence.read` | The same snapshot as `/internal/live/presence`. |
++| `GET /internal/chat/presence` | `chat.presence.read` | Who is connected where: counts, per-stream viewers, slow modes, users/anons with their addresses (`server/chat/presence.js`). |
`/internal/chat/events` is a narrow transient event ingress, not a generic ChatServer call.
`live.release.deployed` continues through OpenVibe.Events; there is no deploy-notice endpoint.
@@@ -39,18 +39,17 @@ Before Live repoints, the separate **OpenVibe.Contracts** repository must regist
`chat.event.publish`, `chat.moderation.write`, and `chat.cache.invalidate` and add them to
`manifests/services/chat.json`. Network's `principal_grants` and `DEFAULT_GRANTS` must grant
`live chat.event.publish openvibe.chat`, `live chat.moderation.write openvibe.chat`, and
--`live chat.cache.invalidate openvibe.chat`. Keep the existing Live grants for
--`chat.message.send`, `chat.presence.read`, and `chat.live_bridge.write` until Live no longer calls
--the bridge. Chat's local proposals are in `docs/capabilities-proposal/`.
++`live chat.cache.invalidate openvibe.chat`. Keep the Live grants for `chat.message.send`
++(`messages`), `chat.presence.read` (`/internal/chat/presence`) and `chat.live_bridge.write` (the call
++hooks at `/internal/calls/stream-channel`): the bridge is gone, but each still guards a route. Chat's
++local proposals are in `docs/capabilities-proposal/`.
--Then Live J2 replaces its bridge writer with a typed client: repoint message producers and persisted DMs
++Live J2 (done: Live #20/#21) replaced its bridge writer with a typed client: repoint message producers and persisted DMs
to `messages` (with `role`/`profile_color` etc. in `frame`), card, sound, media-queue, redemption
and vibe-coding producers to `events`, the call invite/response `sendDm` calls to `events` with a
`user` target, moderation writes and disconnects to
`moderation`, cache hints to `invalidate`, and presence reads to `/internal/chat/presence`.
--Await the real message id; use one stable key per operation. Drain Live's bridge outbox before
--removing its sender. Live J4b moves arena-command replies into the Chat-effects response; Chat sends
--each returned reply to the socket that ran the command (on the bridge path the response carries none
--and Live answers the connection itself).
++Await the real message id; use one stable key per operation. Live J4b moves arena-command replies
++into the Chat-effects response; Chat sends each returned reply to the socket that ran the command.
For deploy notices, Live publishes `live.release.deployed` through Events. If Events outbox
initialization fails, Live must leave `deploy_last_announced` unchanged so the next boot retries.
ddiff --git a/docs/cutover.md b/docs/cutover.md
iindex 6000ea2..cc00b74 100644
---- a/docs/cutover.md
++++ b/docs/cutover.md
@@@ -10,6 +10,12 @@ until a later wave retires them.
The cutover itself ran on 2026-09-23; the rehearsal, flip and check steps below record that deploy.
++**The Live chat bridge is retired (T3 J3a).** Live #20/#21 route every producer through Chat's own
++ingress (`/internal/chat/*`, `docs/chat-ingress.md`) and dropped Live's bridge outbox, so Chat no
++longer mounts `POST /internal/live/calls` or `GET /internal/live/presence`, and deploy notices arrive
++only as `live.release.deployed`. Where the steps below mention the bridge, they record the cutover as
++it ran. Its tables `bridge_applied` / `bridge_refs` stay as data until a later migration (J3b).
++
## Rehearsal
`ov rehearse` loads this seed after main's migrations, then this branch's migrations:
@@@ -71,14 +77,14 @@ writers are repointed; deploy the three new capability grants before that Live r
| 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` | **Chat** (C-04 done) | Chat's own tables; Chat is their only writer. Live reads what it needs through `GET /internal/moderation/*` (`chat.moderation.read`), and Live's current release's writers call Chat over the bridge (op `db`) until Live N+1 is deployed. There is no authority switch any more. |
++| `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 (the bridge at the cutover; Chat's ingress since T3 J3a). |
++| `channel_moderators`, `channel_moderation_settings`, `emotes`, `user_tags`, `chat_ai_summaries`, `chat_timeline_events` | **Chat** (C-04 done) | Chat's own tables; Chat is their only writer. Live reads what it needs through `GET /internal/moderation/*` (`chat.moderation.read`), and Live's writers call Chat's APIs and ingress (the bridge op `db` they used is retired, T3 J3a). There is no authority switch any more. |
| `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`. |
Chat also keeps `ctx_users`, `ctx_streams`, `ctx_managed_streams`, `ctx_channels` — projections of
Live data maintained by `live-context` (never authority), `events_outbox`, `live_mirror_outbox`,
--`bridge_applied`, `import_hold`, `import_runs`.
++`bridge_applied` (the retired bridge's, kept as data), `import_hold`, `import_runs`.
## Prerequisites
@@@ -102,8 +108,9 @@ Live data maintained by `live-context` (never authority), `events_outbox`, `live
`chat live.chat_effects.write openvibe.live`, `chat live.chat_mirror.write openvibe.live`,
`live chat.live_bridge.write openvibe.chat`, `live chat.message.send openvibe.chat`,
`live chat.presence.read openvibe.chat` (`chat events.event.publish openvibe.events` is already
-- a default). Without `chat.message.send`, Chat refuses the bridge ops that send a message (AI
-- viewer, relay and donation lines, deploy notices). Add them to
++ a default). Without `chat.message.send`, Chat refused the bridge ops that sent a message (AI
++ viewer, relay and donation lines, deploy notices); since T3 J3a it guards
++ `POST /internal/chat/messages`. Add them to
`DEFAULT_GRANTS` in `server/identity/principals.js` so every boot keeps them.
VIP member badges (after this cutover, any time): `chat vip.entitlement.check openvibe.vip`;
without it no message carries a badge (README, "VIP member badges").
@@@ -249,8 +256,8 @@ unanswered, not served by Live. The nginx include stays in place through any rol
- **Live unreachable.** Chat keeps serving rooms from its caches; new sign-ins resolve as anonymous
and new anonymous visitors get a temporary number (`anon9xxxxxxxx`) until Live answers; effects are
skipped (coins, AI viewers) or answer an error (`/color`, media commands).
--- **Deploy notices** are still Live's (its commits, announced in chat through the bridge); Chat's own
-- deploys are not announced.
++- **Deploy notices** are still Live's (its commits, announced in chat through the OpenVibe.Events
++ event `live.release.deployed`; the bridge op is retired, T3 J3a); Chat's own deploys are not announced.
- **Responses** carry the same fields as before; Chat's `*subject_id` columns are stripped from API
answers.
@@@ -272,8 +279,8 @@ Found while moving; deliberately not fixed in this wave (fix after the cutover,
## PostgreSQL switch (plan T3, ADR-035)
--Chat's own database moves from the SQLite file to PostgreSQL; the Live read mirror and the bridge are
--unchanged in meaning. This is a runbook for that one deploy — the schema is `migrations/0001_initial.sql`
++Chat's own database moves from the SQLite file to PostgreSQL; the Live read mirror and the bridge (since
++retired, T3 J3a) were unchanged in meaning. This is a runbook for that one deploy — the schema is `migrations/0001_initial.sql`
(applied by `openvibe-sdk/db` at boot, owner role) and the data moves once.
**Prerequisites.** `DATABASE_URL` (the runtime role, through PgBouncer — DML only, no session state) and
ddiff --git a/server/app.js b/server/app.js
iindex 10f2101..22f9341 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -61,7 +61,7 @@ function getAllowedOrigins() {
return allowed;
}
--function createApp({ chatServer, bridge, mirror, relay, events = null, callServer = null }) {
++function createApp({ chatServer, mirror, relay, events = null, callServer = null }) {
const allowedOrigins = getAllowedOrigins();
const app = express();
app.disable('x-powered-by');
@@@ -101,8 +101,6 @@ function createApp({ chatServer, bridge, mirror, relay, events = null, callServe
}
// ── Internal (loopback; never routed by nginx) ─────────────
-- // Alert sounds travel as base64 in broadcasts (uploads up to MAX_SOUND_SIZE_KB): room for them.
-- app.use('/internal/live', express.json({ limit: '16mb' }), bridge);
app.use('/internal/chat', express.json({ limit: '64kb' }), createInternalIngress({ chatServer }));
// OpenVibe.Events deliveries (live.release.deployed, network.module.updated): signature v2 over
// the raw body, so this router reads the body itself (server/events/consumer.js).
ddiff --git a/server/bridge/live-bridge.js b/server/bridge/live-bridge.js
ddeleted file mode 100644
iindex 7f73d9d..0000000
---- a/server/bridge/live-bridge.js
++++ /dev/null
@@@ -1,268 +0,0 @@
--/**
-- * OpenVibe.Chat — the bridge Live uses while it still hosts code that talks to chat.
-- *
-- * With CHAT_AUTHORITY=chat Live stops running the chat server; its `require('./chat/chat-server')`
-- * becomes a proxy (Live server/chat/chat-remote.js, docs/live-patch.diff) that sends every call
-- * here, in order, in small batches:
-- *
-- * POST /internal/live/calls capability chat.live_bridge.write
-- * { boot, ops: [{ seq, op, args, ref? }] } → { ok, results: [{ seq, ok, result?, error? }] }
-- * op = a ChatServer method Live's modules call (broadcastToStream, forwardToGlobal, sendDm,
-- * synthesizeAndBroadcastTTS, triggerChannelSound, disconnectUser, sendUserUpdate, …), a few
-- * bridge-only ops (sendToConn, deployNotice, playAlertSound, userChanged, invalidate, broadcastAllRaw), or `db` — one of
-- * the chat-table writes Live's un-moved code still makes (AI viewers, relays, donations,
-- * /api/mod), executed on Chat's database, which is the authority.
-- *
-- * Pending ids: Live's saveChatMessage cannot wait for Chat's id, so it hands its caller a
-- * placeholder (a large negative number, `ref`) and sends the insert with it. Chat maps the
-- * placeholder to the real id, and every later op in that Live boot that carries it — a
-- * broadcast `{ id }`, a TTS key `m<ref>`, a reply_to_id — is rewritten before it runs. The
-- * map is kept in bridge_refs as well, so it survives a Chat restart between the batch that
-- * acknowledged the insert and a later batch that carries the placeholder.
-- *
-- * An op that sends a chat message also needs chat.message.send in the same token (below);
-- * without it that op alone is refused ({ ok: false, code: 'capability.denied' }) and the
-- * rest of the batch still runs.
-- *
-- * The six tables Chat owns (channel moderators/settings, emotes, user tags, chat-AI
-- * summaries/timeline), same endpoint and capability: db ops addChannelModerator …
-- * addChatTimelineEvents. Chat is their only writer since the C-04 cutover (there is no
-- * authority switch any more); Live's current release still calls these ops until its next
-- * deploy, each answering { value, mirror } — Live's own return value and the rows as they are
-- * now, which Live applies to its copy at once.
-- *
-- * 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,
-- * slowModeByStream). Internal only: it carries addresses.
-- */
--'use strict';
--
--const express = require('express');
--const { capabilities } = require('openvibe-contracts');
--const db = require('../db/database');
--const ctx = require('../live-context');
--const serviceAuth = require('../net/service-auth');
--
--// Chat-table writes Live's remaining modules make (function names and arguments of Live's
--// database.js, which Chat's database.js keeps).
--const DB_OPS = new Set([
-- 'saveChatMessage', 'mergeChatMessageMetadata', 'deleteChatMessage', 'deleteUserChatMessages',
-- 'deleteAnonChatMessages', 'deleteRelayUserMessages', 'deleteChatMessagesByTimeRange',
-- 'holdMessageForApproval', 'reviewPendingIpMessage', 'approveAllFromIp', 'denyAllFromIp',
-- 'recordRelayUser', 'hideRelayUser', 'unhideRelayUser', 'unhideRelayUserByIdentity',
-- 'logModerationAction', 'recordFirstChat', 'setTtsVoiceOverride', 'deleteTtsVoiceOverride',
-- 'createChannelSound', 'setChannelSoundEmote', 'deleteChannelSound', 'renameChannelSoundCommand',
-- 'updateChannelSoundEmoteRefs',
-- // The six tables Chat owns (C-04 done): Live's current release's writers, kept as thin calls
-- // to the same database functions until Live N+1 is deployed.
-- '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
--// loaded into the projection first, so labels like source_channel are right on a fresh stream).
--const SERVER_OPS = {
-- broadcastToStream: [0],
-- broadcastToChannelRoom: [1],
-- broadcastGlobal: [],
-- broadcastAll: [],
-- forwardToGlobal: [0],
-- forwardToGlobalByChannel: [],
-- forwardToStreamerRooms: [0],
-- broadcastToOwnerStreams: [],
-- sendDm: [],
-- sendUserUpdate: [],
-- disconnectUser: [],
-- synthesizeAndBroadcastTTS: [0],
-- triggerChannelSound: [],
-- sendToConn: [],
--};
--
--// chat.message.send: a service or app principal sending a chat message through this API must hold
--// it, on top of chat.live_bridge.write. Sending a message = persisting one (db.saveChatMessage, any
--// message_type), the deploy notice (it writes one), or pushing a chat line (a frame of type 'chat')
--// or a DM (a frame of type 'dm', through sendDm) to browsers. Other frames (system notices, counts,
--// alerts, call invites through sendDm), TTS, channel sounds, moderation writes and invalidations
--// need only the bridge capability.
--//
--// chat.message.send is chat's own capability (openvibe-contracts >= 0.30.2 lists it in the chat
--// manifest), so the check below names it literally and openvibe-contracts-check enforces it.
--const MESSAGE_SEND = 'chat.message.send';
--const FRAME_OPS = new Set([
-- 'broadcastToStream', 'broadcastToChannelRoom', 'broadcastGlobal', 'broadcastAll', 'forwardToGlobal',
-- 'forwardToGlobalByChannel', 'forwardToStreamerRooms', 'broadcastToOwnerStreams', 'sendToConn', 'sendDm',
--]);
--const MESSAGE_FRAMES = new Set(['chat', 'dm']);
--
--function isMessageFrame(frame) {
-- return !!frame && typeof frame === 'object' && MESSAGE_FRAMES.has(frame.type);
--}
--
--/** Does this bridge op send a chat message (and so need chat.message.send)? */
--function sendsMessage(op, args) {
-- const a = Array.isArray(args) ? args : [];
-- if (op === 'db') return a[0] === 'saveChatMessage';
-- if (op === 'deployNotice') return true;
-- if (FRAME_OPS.has(op)) return isMessageFrame(a[a.length - 1]); // the frame is each of these ops' last argument
-- if (op === 'broadcastAllRaw') { try { return isMessageFrame(JSON.parse(String(a[0]))); } catch { return false; } }
-- return false;
--}
--
--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, mirror = null, config = require('../config') }) {
-- const router = express.Router();
-- const refs = new Map(); // `${boot}|${ref}` → { id, at }
--
-- async function sweepRefs() {
-- const cutoff = Date.now() - REF_TTL_MS;
-- for (const [k, v] of refs) if (v.at < cutoff) refs.delete(k);
-- try { await db.run('DELETE FROM bridge_applied WHERE applied_at < ?', [Date.now() - 7 * 24 * 3600 * 1000]); } catch { /* */ }
-- try { await db.run('DELETE FROM bridge_refs WHERE at < ?', [Date.now() - REF_KEEP_MS]); } catch { /* */ }
-- }
--
-- async function rememberRef(boot, ref, id) {
-- const at = Date.now();
-- refs.set(`${boot}|${ref}`, { id, at });
-- await db.run('INSERT INTO bridge_refs (boot, ref, id, at) VALUES (?, ?, ?, ?) ON CONFLICT(boot, ref) DO UPDATE SET id = excluded.id, at = excluded.at', [boot, ref, id, at]);
-- }
--
-- /** The real id for a placeholder of this Live boot: memory first, then bridge_refs (a restart). */
-- async function lookupRef(boot, ref) {
-- const hit = refs.get(`${boot}|${ref}`);
-- if (hit) return hit.id;
-- let row = null;
-- try { row = await db.get('SELECT id FROM bridge_refs WHERE boot = ? AND ref = ?', [boot, ref]); } catch { row = null; }
-- if (!row) return null;
-- refs.set(`${boot}|${ref}`, { id: row.id, at: Date.now() });
-- return row.id;
-- }
-- const sweep = setInterval(sweepRefs, 60_000);
-- if (sweep.unref) sweep.unref();
--
-- async function mapRefs(value, boot) {
-- if (typeof value === 'number') {
-- if (value <= REF_MIN) { const id = await lookupRef(boot, value); return id != null ? id : value; }
-- return value;
-- }
-- if (typeof value === 'string') {
-- const m = /^(m|ov-)(-\d{13,})$/.exec(value);
-- if (m) { const id = await lookupRef(boot, Number(m[2])); return id != null ? `${m[1]}${id}` : value; }
-- return value;
-- }
-- if (Array.isArray(value)) return (await Promise.all(value.map(async (v) => await mapRefs(v, boot))));
-- if (value && typeof value === 'object') {
-- const out = {};
-- for (const [k, v] of Object.entries(value)) out[k] = await mapRefs(v, boot);
-- return out;
-- }
-- return value;
-- }
--
-- const plain = (r) => {
-- if (r && typeof r === 'object' && 'changes' in r && 'lastInsertRowid' in r) return { changes: r.changes, lastInsertRowid: Number(r.lastInsertRowid) };
-- return r === undefined ? null : r;
-- };
--
-- async function runOp(op, rawArgs, boot, ref, key) {
-- const args = await mapRefs(Array.isArray(rawArgs) ? rawArgs : [], boot);
-- if (op === 'db') {
-- const [fn, ...fnArgs] = args;
-- if (!DB_OPS.has(fn) || typeof db[fn] !== 'function') throw new Error(`db.${fn} is not a bridge write`);
-- // saveChatMessage derives the channel from the stream: have the stream first.
-- if (fn === 'saveChatMessage' && fnArgs[0] && fnArgs[0].stream_id) await ctx.ensureStream(fnArgs[0].stream_id);
-- const remember = async (result) => {
-- if (ref != null && result && result.lastInsertRowid != null) await rememberRef(boot, ref, Number(result.lastInsertRowid));
-- return result;
-- };
-- if (!key) return plain(await remember(await db[fn](...fnArgs)));
-- // Applied once per idempotency key, in the same transaction as the write.
-- return await db.tx(async () => {
-- const done = await db.get('SELECT result FROM bridge_applied WHERE key = ?', [key]);
-- if (done) { const prev = done.result ? JSON.parse(done.result) : null; return await remember(prev); }
-- const result = plain(await db[fn](...fnArgs));
-- await db.run('INSERT INTO bridge_applied (key, result, applied_at) VALUES (?, ?, ?)', [key, JSON.stringify(result === undefined ? null : result), Date.now()]);
-- return await remember(result);
-- });
-- }
-- if (Object.prototype.hasOwnProperty.call(SERVER_OPS, op)) {
-- for (const i of SERVER_OPS[op]) if (args[i]) await ctx.ensureStream(args[i]);
-- if (op === 'disconnectUser') await ctx.invalidateBans(); // a ban was just written in Live
-- if (op === 'triggerChannelSound') {
-- // (ws, client, stream, command, args, relay) — Live passes no socket.
-- const [, client, stream, command, soundArgs, relay] = args;
-- return plain(await chatServer.triggerChannelSound(null, client || {}, stream, command, soundArgs || [], relay || null));
-- }
-- return plain(await chatServer[op](...args));
-- }
-- switch (op) {
-- case 'broadcastAllRaw': {
-- // Live code that iterated chatServer.clients and sent a pre-serialized payload.
-- let data;
-- try { data = JSON.parse(String(args[0])); } catch { throw new Error('broadcastAllRaw: not JSON'); }
-- return plain(await chatServer.broadcastAll(data));
-- }
-- case 'deployNotice':
-- return await require('../chat/deploy-notice').announceCommits({ db, chatServer, commits: args[0] });
-- case 'playAlertSound': {
-- // Live's donation / goal-reached alerts: Live says "play the alert for channel X"
-- // (streamerId, streamId, kind), Chat resolves the sound from its own settings row and
-- // broadcasts it to the channel room. Replaces Live reading the file + sending base64.
-- const [streamerId, streamId, kind] = args;
-- return await require('../chat/alert-sounds').playAlertSound(chatServer, Number(streamerId) || 0, streamId != null ? Number(streamId) : null, kind === 'goal' ? 'goal' : 'donation');
-- }
-- case 'userChanged':
-- ctx.invalidateUser(args[0]);
-- await ctx.ensureUsers([args[0]]).catch(() => {});
-- return null;
-- case 'invalidate': {
-- // Live wrote data Chat caches (channel moderators/settings, IP approvals, bans, a user).
-- const [kind, id] = args;
-- if (kind === 'channel') await ctx.invalidateChannel(id);
-- else if (kind === 'approvals') ctx.invalidateApprovals(id);
-- else if (kind === 'bans') await ctx.invalidateBans();
-- else if (kind === 'user') ctx.invalidateUser(id);
-- else throw new Error(`invalidate: unknown kind ${kind}`);
-- return null;
-- }
-- default:
-- throw new Error(`unknown op ${op}`);
-- }
-- }
--
-- router.post('/calls', serviceAuth.guard('chat.live_bridge.write'), async (req, res) => {
-- const boot = String((req.body && req.body.boot) || '').slice(0, 64) || 'unknown';
-- const ops = Array.isArray(req.body && req.body.ops) ? req.body.ops.slice(0, 500) : [];
-- const results = [];
-- const mayMessage = capabilities.check({ cap: req.principal && req.principal.cap }, 'chat.message.send');
-- for (const o of ops) {
-- if (!mayMessage.allowed && sendsMessage(String(o.op || ''), o.args)) {
-- console.warn(`[Bridge] ${o.op}${o.op === 'db' && Array.isArray(o.args) ? `.${o.args[0]}` : ''} refused: ${req.principal && req.principal.sub} lacks ${MESSAGE_SEND}`);
-- results.push({ seq: o.seq, ok: false, code: mayMessage.code || 'capability.denied', error: `${MESSAGE_SEND} not granted` });
-- continue;
-- }
-- try {
-- const result = await runOp(String(o.op || ''), o.args, boot, typeof o.ref === 'number' ? o.ref : null, typeof o.key === 'string' ? o.key.slice(0, 80) : null);
-- results.push({ seq: o.seq, ok: true, result });
-- } catch (err) {
-- console.warn(`[Bridge] ${o.op}${o.op === 'db' && Array.isArray(o.args) ? `.${o.args[0]}` : ''} failed: ${err.message}`);
-- results.push({ seq: o.seq, ok: false, error: err.message });
-- }
-- }
-- res.json({ ok: true, results });
-- });
--
-- router.get('/presence', serviceAuth.guard('chat.presence.read'), async (req, res) => {
-- res.json(await require('../chat/presence').snapshot(chatServer));
-- });
--
-- router._mapRefs = mapRefs;
-- router._refs = refs;
-- return router;
--}
--
--module.exports = { createBridge, DB_OPS, SERVER_OPS, MESSAGE_SEND, sendsMessage };
ddiff --git a/server/calls/internal.js b/server/calls/internal.js
iindex 209ac9d..3baecaf 100644
---- a/server/calls/internal.js
++++ b/server/calls/internal.js
@@@ -1,7 +1,7 @@
/**
* OpenVibe.Chat — Live's stream lifecycle hooks for calls (Live CALLS_AUTHORITY=chat,
* server/streaming/calls-authority.js). Loopback only, never routed by nginx; the same service-token
-- * guard and capability as Live's other calls into Chat (bridge/live-bridge.js):
++ * guard as Live's other calls into Chat, and the capability Live already holds for them:
*
* POST /internal/calls/stream-channel capability chat.live_bridge.write
* { stream_id, mode, user_id } → { ok, channel } go-live with call mode, or the mode changed
ddiff --git a/server/chat/deploy-notice.js b/server/chat/deploy-notice.js
iindex b3eb887..90a964f 100644
---- a/server/chat/deploy-notice.js
++++ b/server/chat/deploy-notice.js
@@@ -4,16 +4,13 @@
// (Moved from OpenVibe.Live server/chat/deploy-notice.js, W6.)
//
// Live still decides WHAT shipped — its git HEAD and the `deploy_last_announced` site setting — and
--// tells Chat two ways while the Chat bridge is being retired (compatibility register C-84):
--// • the bridge op deployNotice (Live server/chat/chat-remote.js), commits newest first, so the
--// first one is the head Live deployed → announceCommits();
--// • the durable OpenVibe.Events event live.release.deployed (Live server/events/release-events.js),
--// subject { type: 'release', id: <head> } → applyReleaseEvent(), from ../events/consumer.js.
--// Both claim the release by its head commit in deploy_releases, in the SAME transaction that stores
--// or folds the card: the first to arrive makes the card, the other finds the head claimed and only
--// records when it saw it. So the same head never makes two cards, whichever path comes first, and a
--// repeat of either (a bridge retry, an Events redelivery, Live re-announcing after a crash) is one
--// card. Once the event path has carried a clean deploy, the bridge op can go.
++// tells Chat through the durable OpenVibe.Events event live.release.deployed (Live
++// server/events/release-events.js), subject { type: 'release', id: <head> } → applyReleaseEvent(),
++// from ../events/consumer.js. The release is claimed by its head commit in deploy_releases, in the
++// SAME transaction that stores or folds the card, so a repeat (an Events redelivery, Live
++// re-announcing after a crash) is one card. The bridge op deployNotice that carried the same notice
++// during compatibility register C-84 is retired (T3 step 1); deploy_releases.first_via/bridge_at
++// keep what it recorded.
//
// Chat keeps the part that is chat:
// • Consecutive deploys fold into ONE stored message: while the newest chat row in ANY room is a
@@@ -77,8 +74,8 @@ async function persist(db, commits) {
}
/**
-- * Claim the release (its head commit) and store or fold its card, in ONE transaction. The first path
-- * to arrive makes the card; the other finds the head claimed, notes when it saw it, and changes
++ * Claim the release (its head commit) and store or fold its card, in ONE transaction. The first
++ * delivery makes the card; a repeat finds the head claimed, notes when it saw it, and changes
* nothing else. Returns { duplicate: true, id } or { duplicate: false, saved }.
*/
async function claimRelease(db, { head, commits, via, eventId = null, now = Date.now() }) {
@@@ -112,28 +109,6 @@ function broadcast({ chatServer, saved, list, log = console }) {
ATTEMPTS_MS.forEach((ms, i) => { const t = setTimeout(() => { const n = push(); if (n) log.log(`[Deploy notice] ${list.length} commit(s) → ${n} client(s) (pass ${i + 1})`); }, ms); if (t.unref) t.unref(); });
}
--/**
-- * The bridge path: commits Live just shipped (Live filtered them against deploy_last_announced),
-- * newest first, so the first is the head. A head already announced (by the event, or a retried
-- * bridge call) announces nothing.
-- * @returns {{ announced: number, id?: number, duplicate?: boolean }}
-- */
--async function announceCommits({ db, chatServer, commits, log = console }) {
-- const list = cleanCommits(commits);
-- if (!list.length) return { announced: 0 };
-- const head = list[0].hash;
--
-- let r;
-- try { r = await claimRelease(db, { head, commits: list, via: 'bridge' }); }
-- catch (err) { log.warn('[Deploy notice] not saved:', err.message); return { announced: 0 }; }
-- if (r.duplicate) {
-- log.log(`[Deploy notice] ${head.slice(0, 7)} already announced (card ${r.id}); the bridge copy changes nothing`);
-- return { announced: 0, id: r.id, duplicate: true };
-- }
-- broadcast({ chatServer, saved: r.saved, list, log });
-- return { announced: list.length, id: r.saved.id };
--}
--
/** The release a live.release.deployed envelope announces ({ head, commits }), or an 'ignored:*' outcome. No I/O. */
function releaseFrom(event, { now = Date.now(), maxAgeMs = RELEASE_MAX_AGE_MS } = {}) {
if (!event || event.event_type !== 'live.release.deployed') return 'ignored:type';
@@@ -153,7 +128,7 @@ function releaseFrom(event, { now = Date.now(), maxAgeMs = RELEASE_MAX_AGE_MS }
* The Events path, called inside the consumer's inbox transaction (synchronous). Returns an
* 'ignored:*' outcome, or { outcome, detail, after } where after() broadcasts once the transaction
* has committed. outcome: announced (a new card) | folded (into the last card) | duplicate:release
-- * (the bridge, or an earlier event with this head, made the card already).
++ * (an earlier event with this head made the card already).
*/
async function applyReleaseEvent({ db, chatServer, event, now = Date.now(), maxAgeMs = RELEASE_MAX_AGE_MS, log = console }) {
const rel = releaseFrom(event, { now, maxAgeMs });
@@@ -176,4 +151,4 @@ function replayTo(ws) {
return false;
}
--module.exports = { replayTo, announceCommits, applyReleaseEvent, releaseFrom, claimRelease, persist, plainText, parseLog, cleanCommits, RELEASE_MAX_AGE_MS };
++module.exports = { replayTo, applyReleaseEvent, releaseFrom, claimRelease, persist, plainText, parseLog, cleanCommits, RELEASE_MAX_AGE_MS };
ddiff --git a/server/index.js b/server/index.js
iindex 9720a32..8546536 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -17,7 +17,6 @@ const db = require('./db/database');
const ctx = require('./live-context');
const chatServer = require('./chat/chat-server');
const callServer = require('./calls/call-server');
--const { createBridge } = require('./bridge/live-bridge');
const { createMirror } = require('./bridge/live-mirror');
const { createRelay } = require('./events/outbox');
const { createEventsConsumer } = require('./events/consumer');
@@@ -57,10 +56,9 @@ async function start() {
chatServer.init(server);
callServer.init(server);
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 });
++ const { app, handleUpgrade } = createApp({ chatServer, mirror, relay, events, callServer });
server.on('request', app);
server.on('upgrade', (req, socket, head) => { handleUpgrade(req, socket, head).catch(() => { try { socket.destroy(); } catch { /* */ } }); });
mirror.start();
@@@ -105,7 +103,7 @@ async function start() {
],
close: [() => db.close()],
});
-- return { server, mirror, relay, bridge, events, callServer, subscriptions: subs, shutdown };
++ return { server, mirror, relay, events, callServer, subscriptions: subs, shutdown };
}
if (require.main === module) {
ddiff --git a/server/net/actor-limits.js b/server/net/actor-limits.js
iindex 163a61f..604fd3d 100644
---- a/server/net/actor-limits.js
++++ b/server/net/actor-limits.js
@@@ -9,7 +9,7 @@
* anyone else ip:<address>, the visitor's own address (Cloudflare → nginx → Chat, net/client-ip.js)
*
* Browsers call these routes themselves (nginx sends Live's chat prefixes here), so no service ever
-- * speaks for many visitors on them. Live's service calls (the bridge at /internal/live, the stream
++ * speaks for many visitors on them. Live's service calls (the ingress at /internal/chat, the stream
* hooks at /internal/calls) carry every viewer's chat and are never limited here: a refusal there
* would drop the whole site's chat, and the WebSocket keeps its own flood control.
*
ddiff --git a/server/net/service-auth.js b/server/net/service-auth.js
iindex ad7efab..9057113 100644
---- a/server/net/service-auth.js
++++ b/server/net/service-auth.js
@@@ -3,7 +3,7 @@
*
* headers(audience, scope) Chat → another service: cached client-credentials token from the
* Network's /oauth/token for client `chat`.
-- * guard(capability) another service → Chat: Express guard for /internal/live/* that
++ * guard(capability) another service → Chat: Express guard for /internal/* that
* verifies an RS256 service token for audience openvibe.chat and
* checks the capability in its `cap` claim.
*
ddiff --git a/test/actor-limits.test.js b/test/actor-limits.test.js
iindex b9e30a5..5443fc6 100644
---- a/test/actor-limits.test.js
++++ b/test/actor-limits.test.js
@@@ -78,7 +78,7 @@ t('health, ready, release.json, metrics and /internal/* are never limited', asyn
assert.strictEqual((await h.http('GET', '/release.json')).status, 200);
assert.strictEqual((await h.http('GET', '/metrics')).status, 200);
assert.notStrictEqual((await h.http('POST', '/internal/events', { body: {} })).status, 429);
-- assert.notStrictEqual((await h.http('POST', '/internal/live/calls', { body: { ops: [] } })).status, 429);
++ assert.notStrictEqual((await h.http('POST', '/internal/chat/messages', { body: {} })).status, 429);
}
});
ddiff --git a/test/audio-queue.test.js b/test/audio-queue.test.js
iindex fa03008..7546b23 100644
---- a/test/audio-queue.test.js
++++ b/test/audio-queue.test.js
@@@ -141,11 +141,9 @@ t('only the broadcaster and moderators skip: /skiptts skips the playing clip and
});
t('the queue over REST: list, skip by id, clear — moderators yes, others 403', async () => {
-- // Fill the room: cat plays, three more wait (through Live's bridge, keyed like chat messages).
++ // Fill the room: cat plays, three more wait (an AI viewer's lines, keyed like chat messages).
synthPlan = [5, 5, 5];
-- const BRIDGE = h.serviceToken(['chat.live_bridge.write', 'chat.message.send']);
-- const r = await h.http('POST', '/internal/live/calls', { token: BRIDGE, body: { boot: 'b', ops: [1, 2, 3].map((n) => ({ seq: n, op: 'synthesizeAndBroadcastTTS', args: [streamId, 'ChatBot', `bot line ${n}`, null, 'ai', 'ai:bot', null, `m9${n}`] })) } });
-- assert.ok(r.body.results.every((x) => x.ok), r.text);
++ for (const n of [1, 2, 3]) await h.chatServer.synthesizeAndBroadcastTTS(streamId, 'ChatBot', `bot line ${n}`, null, 'ai', 'ai:bot', null, `m9${n}`);
assert.strictEqual((await h.http('GET', `/api/tts/queue?stream_id=${streamId}`, { token: viewer.token })).status, 403);
assert.strictEqual((await h.http('GET', `/api/tts/queue?stream_id=${streamId}`)).status, 401);
@@@ -177,8 +175,7 @@ t('the queue over REST: list, skip by id, clear — moderators yes, others 403',
t('failures are recorded and the queue moves on', async () => {
synthPlan = [null, new Error('voice service down'), 6];
-- const BRIDGE = h.serviceToken(['chat.live_bridge.write', 'chat.message.send']);
-- await h.http('POST', '/internal/live/calls', { token: BRIDGE, body: { boot: 'b', ops: ['no audio', 'throws', 'fine'].map((l, i) => ({ seq: i + 1, op: 'synthesizeAndBroadcastTTS', args: [streamId, 'ChatBot', l, null, 'ai', `ai:bot${i}`, null, `mf${i}`] })) } });
++ for (const [i, l] of ['no audio', 'throws', 'fine'].entries()) await h.chatServer.synthesizeAndBroadcastTTS(streamId, 'ChatBot', l, null, 'ai', `ai:bot${i}`, null, `mf${i}`);
await listener.next((m) => m.type === 'tts-audio' && m.message === 'fine');
assert.strictEqual((await byLabel('no audio')).state, 'failed');
assert.strictEqual((await byLabel('no audio')).error, 'no audio');
ddiff --git a/test/bridge.test.js b/test/bridge.test.js
ddeleted file mode 100644
iindex 0e22a37..0000000
---- a/test/bridge.test.js
++++ /dev/null
@@@ -1,224 +0,0 @@
--'use strict';
--/**
-- * The bridge Live's remaining modules use (POST /internal/live/calls, GET /internal/live/presence):
-- * service-token guard, forwarded chat writes with placeholder ids mapped for later ops of the same
-- * Live boot, idempotent retries, only allow-listed writes, pushes to browsers (broadcasts, DMs with
-- * the participant check, disconnects, user updates with the stored-name rewrite, deploy notices),
-- * and the presence snapshot Live's synchronous reads use. Ops that send a chat message also need
-- * chat.message.send, for service and app principals alike.
-- */
--const assert = require('assert');
--const { boot, suite } = require('./helpers');
--
--const t = suite('bridge');
--let h, streamer, viewer, channelId, streamId, viewerWs, globalWs, BRIDGE;
--const REF = -(2 ** 40) - 1;
--
--const calls = (ops, { token = BRIDGE, boot: b = 'live-boot-1', headers } = {}) =>
-- h.http('POST', '/internal/live/calls', { token, headers, body: { boot: b, ops: ops.map((o, i) => ({ seq: i + 1, ...o })) } });
--
--t('boot', async () => {
-- h = await boot();
-- BRIDGE = h.serviceToken(['chat.live_bridge.write', 'chat.message.send']);
-- streamer = h.addUser('streamer', { role: 'streamer' });
-- viewer = h.addUser('viewer');
-- channelId = h.addChannel(streamer.id);
-- streamId = h.addStream(streamer.id, channelId);
-- await h.ctx.sync();
-- viewerWs = await h.ws({ ip: '198.51.100.50', token: viewer.token, stream: streamId });
-- viewerWs.sendJson({ type: 'join', streamId, token: viewer.token });
-- await viewerWs.next((m) => m.type === 'auth');
-- globalWs = await h.ws({ ip: '198.51.100.51' });
-- globalWs.sendJson({ type: 'join' });
-- await globalWs.next((m) => m.type === 'auth');
--});
--
--t('only Live’s service token, only from loopback', async () => {
-- assert.strictEqual((await h.http('POST', '/internal/live/calls', { body: { ops: [] } })).status, 401);
-- assert.strictEqual((await calls([], { token: h.serviceToken(['chat.presence.read']) })).status, 403);
-- assert.strictEqual((await calls([], { token: h.serviceToken(['chat.live_bridge.write'], { aud: 'openvibe.live' }) })).status, 401);
-- assert.strictEqual((await calls([], { headers: { 'X-Forwarded-For': '1.2.3.4' } })).status, 403);
-- assert.strictEqual((await h.http('GET', '/internal/live/presence', { token: BRIDGE })).status, 403);
--});
--
--t('sending a chat message needs chat.message.send; without it only those ops are refused', async () => {
-- const APP = 'app:app_01J8Z3Q4R5S6T7V8W9X0Y1Z2A3';
-- const before = (await h.db.get('SELECT COUNT(*) AS n FROM chat_messages')).n;
-- const outboxBefore = (await h.db.get('SELECT COUNT(*) AS n FROM events_outbox')).n;
-- const commit = { hash: 'f'.repeat(40), short: 'fffffff', date: '2026-09-22T10:00:00Z', subject: 'denied deploy' };
-- const sends = [
-- { op: 'db', ref: REF - 50, key: 'live:denied:1', args: ['saveChatMessage', { stream_id: streamId, username: 'Bot', message: 'denied insert', message_type: 'chat' }] },
-- { op: 'broadcastToStream', args: [streamId, { type: 'chat', id: 1, username: 'Bot', message: 'denied line' }] },
-- { op: 'forwardToGlobal', args: [streamId, { type: 'chat', id: 1, username: 'Bot', message: 'denied line' }] },
-- { op: 'broadcastAllRaw', args: [JSON.stringify({ type: 'chat', id: 1, username: 'Bot', message: 'denied line' })] },
-- { op: 'sendDm', args: [viewer.id, { type: 'dm', message: { id: 1, content: 'denied dm' } }] },
-- { op: 'deployNotice', args: [[commit]] },
-- ];
-- const others = [
-- { op: 'broadcastToStream', args: [streamId, { type: 'system', message: 'not a message send' }] },
-- { op: 'sendDm', args: [viewer.id, { type: 'call-ring', from: 'denied-test' }] },
-- ];
-- for (const [who, token] of [['service', h.serviceToken(['chat.live_bridge.write'])], ['app', h.serviceToken(['chat.live_bridge.write'], { sub: APP })]]) {
-- const r = await calls([...sends, ...others], { token });
-- assert.strictEqual(r.status, 200, r.text);
-- const res = r.body.results;
-- for (const x of res.slice(0, sends.length)) {
-- assert.strictEqual(x.ok, false, `${who}: seq ${x.seq} should be refused`);
-- assert.strictEqual(x.code, 'capability.denied');
-- assert.match(x.error, /chat\.message\.send/);
-- }
-- assert.ok(res.slice(sends.length).every((x) => x.ok), `${who}: other ops still run ${JSON.stringify(res)}`);
-- await viewerWs.next((m) => m.type === 'system' && m.message === 'not a message send');
-- await viewerWs.next((m) => m.type === 'call-ring' && m.from === 'denied-test');
-- }
-- const denied = (m) => /denied (insert|line|dm|deploy)/.test(JSON.stringify(m));
-- assert.ok(await viewerWs.none(denied), 'no refused message reached the stream room');
-- assert.ok(await globalWs.none(denied), 'no refused message reached global chat');
-- assert.strictEqual((await h.db.get('SELECT COUNT(*) AS n FROM chat_messages')).n, before, 'nothing persisted');
-- assert.strictEqual((await h.db.get('SELECT COUNT(*) AS n FROM events_outbox')).n, outboxBefore, 'no chat.message.created');
-- assert.strictEqual((await h.db.get('SELECT COUNT(*) AS n FROM bridge_applied WHERE key = ?', ['live:denied:1'])).n, 0, 'a refused write is not marked applied');
-- // The capability alone is not enough either: the bridge capability is still required.
-- assert.strictEqual((await calls(sends.slice(0, 1), { token: h.serviceToken(['chat.message.send']) })).status, 403);
-- // An app principal holding both may send.
-- const ok = await calls([{ op: 'broadcastToStream', args: [streamId, { type: 'chat', id: 2, username: 'App', message: 'app line' }] }], { token: h.serviceToken(['chat.live_bridge.write', 'chat.message.send'], { sub: APP }) });
-- assert.ok(ok.body.results[0].ok, ok.text);
-- await viewerWs.next((m) => m.type === 'chat' && m.message === 'app line');
--});
--
--let realId;
--t('a forwarded insert gets its real id, and later ops of the boot are rewritten', async () => {
-- const r = await calls([
-- { op: 'db', ref: REF, key: 'live:1', args: ['saveChatMessage', { stream_id: streamId, user_id: null, username: 'ChatBot', message: 'beep boop', message_type: 'chat', source_platform: 'ai', metadata: { bot: 1 } }] },
-- { op: 'broadcastToStream', args: [streamId, { type: 'chat', id: REF, username: 'ChatBot', message: 'beep boop', is_ai: true }] },
-- { op: 'forwardToGlobal', args: [streamId, { type: 'chat', id: REF, username: 'ChatBot', message: 'beep boop' }] },
-- ]);
-- assert.strictEqual(r.status, 200, r.text);
-- assert.ok(r.body.results.every((x) => x.ok), JSON.stringify(r.body.results));
-- realId = r.body.results[0].result.lastInsertRowid;
-- assert.ok(realId > 0);
-- const got = await viewerWs.next((m) => m.type === 'chat' && m.message === 'beep boop');
-- assert.strictEqual(got.id, realId, 'placeholder replaced by the real id');
-- const cross = await globalWs.next((m) => m.type === 'chat' && m.message === 'beep boop');
-- assert.strictEqual(cross.id, realId);
-- assert.strictEqual(cross.stream_channel, 'streamer');
-- const row = await h.db.getChatMessageById(realId);
-- assert.strictEqual(row.channel_user_id, streamer.id, 'channel derived from the stream');
-- assert.strictEqual(row.source_platform, 'ai');
--});
--
--t('a retried write (same key) is applied once and answers the first result', async () => {
-- const before = (await h.db.get('SELECT COUNT(*) AS n FROM chat_messages')).n;
-- const r = await calls([{ op: 'db', ref: REF, key: 'live:1', args: ['saveChatMessage', { stream_id: streamId, username: 'ChatBot', message: 'beep boop', message_type: 'chat' }] }]);
-- assert.strictEqual(r.body.results[0].result.lastInsertRowid, realId);
-- assert.strictEqual((await h.db.get('SELECT COUNT(*) AS n FROM chat_messages')).n, before);
--});
--
--t('placeholders are per Live boot', async () => {
-- const r = await calls([{ op: 'broadcastToStream', args: [streamId, { type: 'system', message: 'other boot', id: REF }] }], { boot: 'live-boot-2' });
-- assert.ok(r.body.results[0].ok);
-- const got = await viewerWs.next((m) => m.type === 'system' && m.message === 'other boot');
-- assert.strictEqual(got.id, REF, 'unknown placeholder left alone');
--});
--
--t('only allow-listed writes; unknown ops refused', async () => {
-- const r = await calls([
-- { op: 'db', args: ['run', 'DELETE FROM chat_messages'] },
-- { op: 'eval', args: ['1'] },
-- { op: 'db', args: ['deleteChatMessage', realId, streamer.id] },
-- ]);
-- assert.deepStrictEqual(r.body.results.map((x) => x.ok), [false, false, true]);
-- assert.ok((await h.db.get('SELECT COUNT(*) AS n FROM chat_messages')).n > 0);
-- assert.strictEqual((await h.db.getChatMessageById(realId)).is_deleted, 1);
-- // Moderation from Live's /api/mod lands in Chat's log, with an internal event.
-- const m = await calls([{ op: 'db', ref: REF - 1, key: 'live:2', args: ['logModerationAction', { scope_type: 'site', actor_user_id: streamer.id, action_type: 'message_delete', details: { id: realId } }] }]);
-- assert.ok(m.body.results[0].ok);
-- const ev = (await h.db.all("SELECT event FROM events_outbox WHERE event_type = 'chat.moderation.action'")).map((x) => JSON.parse(x.event)).pop();
-- assert.strictEqual(ev.payload.action_type, 'message_delete');
-- assert.strictEqual(ev.visibility, 'internal');
--});
--
--t('DMs through the bridge keep the participant check', async () => {
-- // viewer is not in conversation 999 (it does not exist): nothing is delivered.
-- await calls([{ op: 'sendDm', args: [viewer.id, { type: 'call-invite', conversation_id: 999 }] }]);
-- assert.ok(await viewerWs.none((m) => m.type === 'call-invite'));
-- // Without a conversation id (call ringing) it is delivered, as in Live.
-- await calls([{ op: 'sendDm', args: [viewer.id, { type: 'call-ring', from: 'streamer' }] }]);
-- await viewerWs.next((m) => m.type === 'call-ring');
--});
--
--t('a user update from Live refreshes sockets and rewrites stored names', async () => {
-- await h.db.saveChatMessage({ stream_id: streamId, user_id: viewer.id, username: 'Viewer', message: 'old name line', message_type: 'chat' });
-- await calls([{ op: 'sendUserUpdate', args: [viewer.id, { id: viewer.id, username: 'viewer', display_name: 'Viewer Renamed', role: 'user', avatar_url: null, profile_color: '#123456' }] }]);
-- const u = await viewerWs.next((m) => m.type === 'user-updated');
-- assert.strictEqual(u.user.display_name, 'Viewer Renamed');
-- assert.strictEqual((await h.db.get('SELECT username FROM chat_messages WHERE message = ?', ['old name line'])).username, 'Viewer Renamed');
-- await h.sleep(1100);
-- viewerWs.sendJson({ type: 'chat', message: 'with the new name' });
-- const own = await globalWs.next((m) => m.type === 'chat' && m.message === 'with the new name');
-- assert.strictEqual(own.username, 'Viewer Renamed');
--});
--
--t('broadcastAllRaw (Live code that looped over chatServer.clients) reaches everyone', async () => {
-- await calls([{ op: 'broadcastAllRaw', args: [JSON.stringify({ type: 'delete-messages', ids: [1, 2] })] }]);
-- await viewerWs.next((m) => m.type === 'delete-messages');
-- await globalWs.next((m) => m.type === 'delete-messages');
--});
--
--t('deploy notices: stored as one rolling global row, shown to everyone, folded on the next deploy', async () => {
-- const commit = (n) => ({ hash: String(n).repeat(40).slice(0, 40), short: String(n).repeat(7), date: '2026-09-22T10:00:00Z', subject: `change ${n}` });
-- let r = await calls([{ op: 'deployNotice', args: [[commit(1)]] }]);
-- const id1 = r.body.results[0].result.id;
-- const row = await h.db.getChatMessageById(id1);
-- assert.strictEqual(row.message_type, 'system');
-- assert.strictEqual(JSON.parse(row.metadata).kind, 'deploy');
-- r = await calls([{ op: 'deployNotice', args: [[commit(2)]] }]);
-- assert.strictEqual(r.body.results[0].result.id, id1, 'folded into the same row');
-- assert.strictEqual(JSON.parse((await h.db.getChatMessageById(id1)).metadata).deploys, 2);
-- // Someone speaks in a stream room (the global feed shows it): the next deploy is a new card,
-- // so no card's time range runs past a message under it.
-- await h.db.saveChatMessage({ stream_id: streamId, user_id: viewer.id, username: viewer.username, message: 'hello brother', message_type: 'chat', is_global: false });
-- r = await calls([{ op: 'deployNotice', args: [[commit(3)]] }]);
-- const id3 = r.body.results[0].result.id;
-- assert.notStrictEqual(id3, id1, 'a message since the last card starts a new card');
-- assert.strictEqual(JSON.parse((await h.db.getChatMessageById(id3)).metadata).deploys, 1);
-- assert.strictEqual(JSON.parse((await h.db.getChatMessageById(id1)).metadata).deploys, 2, 'the old card is left as it was');
-- // A late joiner gets it once.
-- const late = await h.ws({ ip: '198.51.100.52' });
-- late.sendJson({ type: 'join' });
-- const upd = await late.next((m) => m.type === 'update', 4000);
-- assert.strictEqual(upd.id, id3);
-- late.close();
--});
--
--t('the invalidate op is accepted for a channel and refuses an unknown kind', async () => {
-- assert.strictEqual(await h.ctx.isChannelModerator(viewer.id, channelId), false);
-- await h.db.addChannelModerator(channelId, viewer.id, streamer.id); // the moderator row is Chat's own (C-04)
-- const r = await calls([{ op: 'invalidate', args: ['channel', channelId] }, { op: 'invalidate', args: ['nonsense', 1] }]);
-- assert.deepStrictEqual(r.body.results.map((x) => x.ok), [true, false]);
-- assert.strictEqual(await h.ctx.isChannelModerator(viewer.id, channelId), true);
--});
--
--t('presence snapshot for Live’s synchronous reads', async () => {
-- const p = await h.http('GET', '/internal/live/presence', { token: h.serviceToken(['chat.presence.read']) });
-- assert.strictEqual(p.status, 200);
-- assert.ok(p.body.total >= 2);
-- assert.strictEqual(p.body.streams[streamId], 1);
-- assert.ok(p.body.users.some((u) => u.user_id === viewer.id && u.ip === '198.51.100.50'));
-- assert.ok(p.body.anons.some((a) => a.ip === '198.51.100.51'));
--});
--
--t('disconnectUser (a ban in Live) closes the sockets and refreshes the ban list', async () => {
-- const bye = await h.ws({ ip: '198.51.100.53', token: viewer.token });
-- bye.sendJson({ type: 'join', token: viewer.token });
-- await bye.next((m) => m.type === 'auth');
-- h.live.addBan({ user_id: viewer.id });
-- const closed = new Promise((r) => bye.on('close', r));
-- await calls([{ op: 'disconnectUser', args: [{ userId: viewer.id }] }]);
-- await bye.next((m) => m.type === 'system' && m.message === 'You have been banned.');
-- await closed;
-- assert.ok(h.ctx.isUserBanned(viewer.id, null), 'ban list refreshed');
-- h.live.clearBans();
-- await h.ctx.invalidateBans();
--});
--
--t.run(async () => { for (const w of [viewerWs, globalWs]) { try { w.close(); } catch { /* */ } } if (h) await h.close(); });
ddiff --git a/test/chat-apis.test.js b/test/chat-apis.test.js
iindex feb6d9c..469dee3 100644
---- a/test/chat-apis.test.js
++++ b/test/chat-apis.test.js
@@@ -4,7 +4,7 @@
* - Live's internal read API (/internal/moderation/*, capability chat.moderation.read)
* - the public emote API (/api/emotes, Live's paths and shapes; bytes on OpenVibe.Media)
* - channel moderators & moderation settings (/api/chat/channels/:id/…)
-- * - the bridge alert-sound op (Live says "play the alert", Chat resolves the sound)
++ * - the alert sound (playAlertSound: "play the alert", Chat resolves the sound)
* Chat is the only writer of the six tables (C-04 done), so the writes never refuse.
*/
const assert = require('assert');
@@@ -14,7 +14,7 @@ const { validate } = require('openvibe-contracts');
const { boot, suite } = require('./helpers');
const t = suite('chat-apis');
--let h, RO, BRIDGE, streamer, other, admin, mod, stranger, channelId, streamId;
++let h, RO, OTHER, streamer, other, admin, mod, stranger, channelId, streamId;
const deletedAssets = [];
let assets = { upload: null, delete: null };
@@@ -27,7 +27,7 @@ t('boot', async () => {
// 3 emotes per channel, so both the clash check and the cap can be exercised.
h = await boot({ env: { MAX_EMOTES_PER_CHANNEL: '3' } });
RO = h.serviceToken(['chat.moderation.read']);
-- BRIDGE = h.serviceToken(['chat.live_bridge.write']);
++ OTHER = h.serviceToken(['chat.live_bridge.write']);
streamer = h.addUser('streamer', { role: 'streamer' });
other = h.addUser('otherstreamer', { role: 'streamer' });
admin = h.addUser('adminny', { role: 'admin' });
@@@ -46,9 +46,9 @@ t('boot', async () => {
t('internal read API: loopback + service token (401 / 403 / forwarded)', async () => {
let r = await h.http('GET', `/internal/moderation/channels/${channelId}`);
assert.strictEqual(r.status, 401, r.text);
-- r = await h.http('GET', `/internal/moderation/channels/${channelId}`, { token: BRIDGE, headers: { 'x-forwarded-for': '203.0.113.9' } });
++ r = await h.http('GET', `/internal/moderation/channels/${channelId}`, { token: OTHER, headers: { 'x-forwarded-for': '203.0.113.9' } });
assert.strictEqual(r.status, 403, 'a forwarded request is refused (loopback only)');
-- r = await h.http('GET', `/internal/moderation/channels/${channelId}`, { token: BRIDGE });
++ r = await h.http('GET', `/internal/moderation/channels/${channelId}`, { token: OTHER });
assert.strictEqual(r.status, 403, 'wrong capability');
r = await h.http('GET', `/api/internal/moderation/channels/${channelId}`, { token: RO });
assert.strictEqual(r.status, 404, 'not a public path');
@@@ -235,7 +235,7 @@ t('moderation settings: owner writes, a mod cannot change policy keys', async ()
assert.ok(Array.isArray(search.body.messages));
});
--// ── Alert-sound op ───────────────────────────────────────────
++// ── Alert sounds ───────────────────────────────────────────
t('alert op: Chat resolves the sound from its own row and broadcasts it', async () => {
const snd = path.join(h.tmp, 'sounds', 'donation.mp3');
fs.writeFileSync(snd, 'ID3alert');
@@@ -245,22 +245,16 @@ t('alert op: Chat resolves the sound from its own row and broadcasts it', async
ws.sendJson({ type: 'join', streamId, channelUserId: streamer.id, anonId: null });
await ws.next((m) => m.type === 'auth' || m.type === 'connected');
-- const calls = (ops) => h.http('POST', '/internal/live/calls', { token: BRIDGE, body: { boot: 'live-boot-alerts', ops: ops.map((o, i) => ({ seq: i + 1, ...o })) } });
-- let r = await calls([{ op: 'playAlertSound', args: [streamer.id, streamId, 'donation'] }]);
-- assert.strictEqual(r.status, 200, r.text);
-- const res = r.body.results[0];
-- assert.strictEqual(res.ok, true, JSON.stringify(res));
-- assert.deepStrictEqual(res.result, { played: true, source: 'donation-alert' });
++ const { playAlertSound } = require('../server/chat/alert-sounds');
++ assert.deepStrictEqual(await playAlertSound(h.chatServer, streamer.id, streamId, 'donation'), { played: true, source: 'donation-alert' });
const frame = await ws.next((m) => m.type === 'soundboard-audio');
assert.strictEqual(frame.source, 'donation-alert');
assert.strictEqual(Buffer.from(frame.audio, 'base64').toString(), 'ID3alert');
ws.close();
// No goal sound set: goal falls back to the donation sound; a channel with none does not play.
-- r = await calls([{ op: 'playAlertSound', args: [streamer.id, streamId, 'goal'] }]);
-- assert.deepStrictEqual(r.body.results[0].result, { played: true, source: 'goal-alert' });
-- r = await calls([{ op: 'playAlertSound', args: [other.id, streamId, 'donation'] }]);
-- assert.deepStrictEqual(r.body.results[0].result, { played: false, source: 'donation-alert' });
++ assert.deepStrictEqual(await playAlertSound(h.chatServer, streamer.id, streamId, 'goal'), { played: true, source: 'goal-alert' });
++ assert.deepStrictEqual(await playAlertSound(h.chatServer, other.id, streamId, 'donation'), { played: false, source: 'donation-alert' });
});
t.run(async () => { if (h) await h.close(); });
ddiff --git a/test/chat-ingress.test.js b/test/chat-ingress.test.js
iindex 228e7bd..f28eb6c 100644
---- a/test/chat-ingress.test.js
++++ b/test/chat-ingress.test.js
@@@ -290,7 +290,7 @@ t('moderation handles bulk deletes, pending IP decisions, relay identities and v
assert.strictEqual(await h.db.getTtsVoiceOverride('viewer'), null);
});
--t('invalidate accepts typed hints; presence retains the bridge snapshot shape', async () => {
++t('invalidate accepts typed hints; presence is the connection snapshot; the retired bridge is gone', async () => {
const hint = { key: 'ingress:hint:1', user: viewer.id, approvals: channelId, bans: true };
assert.strictEqual((await post('invalidate', hint)).status, 200);
assert.strictEqual((await post('invalidate', hint)).status, 200);
@@@ -305,7 +305,8 @@ t('invalidate accepts typed hints; presence retains the bridge snapshot shape',
assert.ok(r.body.total >= 2);
assert.strictEqual(r.body.streams[streamId], 1);
assert.ok(r.body.users.some((u) => u.user_id === viewer.id));
-- assert.strictEqual((await h.http('GET', '/internal/live/presence', { token: h.serviceToken(['chat.presence.read']) })).status, 200);
++ assert.strictEqual((await h.http('GET', '/internal/live/presence', { token: h.serviceToken(['chat.presence.read']) })).status, 404);
++ assert.strictEqual((await h.http('POST', '/internal/live/calls', { token: h.serviceToken(['chat.live_bridge.write']), body: { boot: 'b', ops: [] } })).status, 404);
});
t('moderation disconnect closes only the selected user socket', async () => {
ddiff --git a/test/events-consumer.test.js b/test/events-consumer.test.js
iindex ca44713..8720ea3 100644
---- a/test/events-consumer.test.js
++++ b/test/events-consumer.test.js
@@@ -4,11 +4,10 @@
*
* - POST /internal/events takes signature v2 only: a bad signature, a timestamp outside ±300 s and a
* v1-only delivery are 401; no secret is 503; a forwarded request is 403
-- * - live.release.deployed makes the deploy card exactly as the bridge's deployNotice does; the inbox
-- * makes a redelivered event a no-op (one card)
-- * - a bridge notice and an event for the same head are one card, whichever comes first
++ * - live.release.deployed makes the deploy card; the inbox makes a redelivered event a no-op (one
++ * card), and a second event for a head already announced is one card too
* - folding as before: a later deploy folds into the last card, a message in a stream room in
-- * between starts a new card; the two paths fold into each other's cards
++ * between starts a new card
* - network.module.updated drops a cached chat.preferences copy when newer, ignores an older revision
* - boot creates the subscriptions once (idempotent), never re-enables a disabled one, and the
* secret it hands Events is the one the consumer verifies
@@@ -26,9 +25,8 @@ const t = suite('events-consumer');
const SECRET = 'e'.repeat(64);
const OTHER = 'f'.repeat(64);
const ANN = 'usr_01J9ANN0000000000000000AAA';
--let h, BRIDGE, streamer, viewer, streamId, prefs;
++let h, streamer, viewer, streamId, prefs;
let seq = 0;
--let n = 0;
// ── Stub OpenVibe.Events: subscriptions (token-checked, 409 on a duplicate) and publishing ──
const stubEvents = { subs: [], posts: 0 };
@@@ -104,7 +102,6 @@ function deliver(event, { secret = SECRET, now = Date.now(), headers = {}, strip
for (const k of strip) delete signed[k];
return h.http('POST', '/internal/events', { raw, headers: { 'Content-Type': 'application/json', 'X-OpenVibe-Event-Id': event.event_id, ...signed, ...headers } });
}
--const bridge = (commits) => h.http('POST', '/internal/live/calls', { token: BRIDGE, body: { boot: 'live-boot-1', ops: [{ seq: ++n, op: 'deployNotice', args: [commits] }] } });
const cards = async () => (await h.db.all("SELECT id, metadata FROM chat_messages WHERE message_type = 'system' AND metadata LIKE '%\"kind\":\"deploy\"%' ORDER BY id")).map((r) => ({ id: r.id, ...JSON.parse(r.metadata) }));
const card = async (id) => (await cards()).find((c) => c.id === id);
const speak = async (text) => await h.db.saveChatMessage({ stream_id: streamId, user_id: viewer.id, username: viewer.username, message: text, message_type: 'chat', is_global: false });
@@@ -113,7 +110,6 @@ const releaseRow = async (head) => await h.db.get('SELECT * FROM deploy_releases
t('boot with the consumer on and a stub Events', async () => {
const eventsPort = await new Promise((r) => eventsServer.listen(0, '127.0.0.1', () => r(eventsServer.address().port)));
h = await boot({ env: { CHAT_EVENTS_SECRET: `${SECRET}, short-is-ignored`, EVENTS_URL: `http://127.0.0.1:${eventsPort}`, CHAT_PREFS_TTL_MS: '60000' } });
-- BRIDGE = h.serviceToken(['chat.live_bridge.write', 'chat.message.send']);
streamer = h.addUser('streamer', { role: 'streamer' });
viewer = h.addUser('viewer');
streamId = h.addStream(streamer.id, h.addChannel(streamer.id));
@@@ -219,7 +215,7 @@ t('live.release.deployed makes the deploy card; the same event again is one card
assert.strictEqual(r.body.duplicate, true);
assert.strictEqual((await cards()).length, before, 'still one card');
assert.strictEqual((await card(id)).deploys, 1, 'not folded into itself');
-- // Late joiners get this announcement once, keyed by the row id (as with the bridge).
++ // Late joiners get this announcement once, keyed by the row id.
const late = await h.ws({ ip: '198.51.100.70' });
late.sendJson({ type: 'join' });
const upd = await late.next((m) => m.type === 'update' && m.kind === 'deploy', 4000);
@@@ -230,55 +226,20 @@ t('live.release.deployed makes the deploy card; the same event again is one card
assert.deepStrictEqual([rel.first_via, rel.message_id, rel.event_id, rel.bridge_at], ['events', id, ev.event_id, null]);
});
--t('bridge first, then the event for the same head: one card', async () => {
-- await speak('a stream message, so the next deploy is a new card');
-- const commits = [commit('c', 'bridge: head'), commit('d', 'bridge: parent')];
-- const b = await bridge(commits);
-- const id = b.body.results[0].result.id;
-- assert.strictEqual(b.body.results[0].result.announced, 2);
-- const before = (await cards()).length;
-- const r = await deliver(release(commits));
-- assert.deepStrictEqual([r.status, r.body.duplicate, r.body.outcome], [200, false, 'duplicate:release']);
-- assert.strictEqual(r.body.detail.message_id, id);
-- assert.strictEqual((await cards()).length, before, 'no second card');
-- assert.strictEqual((await card(id)).deploys, 1, 'the card was not folded again');
-- const rel = await releaseRow(commits[0].hash);
-- assert.strictEqual(rel.first_via, 'bridge');
-- assert.ok(rel.bridge_at && rel.event_at, 'both paths are recorded as having delivered it');
--});
--
--t('the event first, then the bridge for the same head: one card', async () => {
-- await speak('another stream message');
-- const commits = [commit('e', 'event: head')];
-- const r = await deliver(release(commits));
-- assert.strictEqual(r.body.outcome, 'announced');
-- const id = r.body.detail.message_id;
-- const before = (await cards()).length;
-- const b = await bridge(commits);
-- assert.deepStrictEqual(b.body.results[0].result, { announced: 0, id, duplicate: true });
-- // A retried bridge op (Live's durable queue after a lost response) is the same.
-- const again = await bridge(commits);
-- assert.strictEqual(again.body.results[0].result.duplicate, true);
-- assert.strictEqual((await cards()).length, before);
-- assert.strictEqual((await card(id)).deploys, 1);
-- const rel = await releaseRow(commits[0].hash);
-- assert.deepStrictEqual([rel.first_via, !!rel.event_at, !!rel.bridge_at], ['events', true, true]);
--});
--
t('folding as before: the next deploy folds into the last card; a stream message in between starts a new card', async () => {
await speak('new card next');
let r = await deliver(release([commit('1', 'fold: one')]));
const first = r.body.detail.message_id;
assert.strictEqual(r.body.outcome, 'announced');
-- // Nobody spoke: the next deploy (from either path) folds into that card.
-- const b = await bridge([commit('2', 'fold: two')]);
-- assert.strictEqual(b.body.results[0].result.id, first, 'the bridge folds into the card the event made');
++ // Nobody spoke: the next deploy folds into that card.
++ r = await deliver(release([commit('2', 'fold: two')]));
++ assert.deepStrictEqual([r.body.outcome, r.body.detail.message_id], ['folded', first]);
r = await deliver(release([commit('3', 'fold: three')]));
assert.deepStrictEqual([r.body.outcome, r.body.detail.message_id], ['folded', first]);
assert.strictEqual((await card(first)).deploys, 3);
assert.deepStrictEqual((await card(first)).commits.map((c) => c.subject), ['fold: three', 'fold: two', 'fold: one']);
r = await deliver(release([commit('2', 'fold: two')]));
-- assert.strictEqual(r.body.outcome, 'duplicate:release', 'the bridge already announced that head');
++ assert.strictEqual(r.body.outcome, 'duplicate:release', 'an earlier event already announced that head');
assert.strictEqual((await card(first)).deploys, 3);
// Someone speaks in a stream room (the global feed shows it): the next deploy is a new card.
await speak('hello brother');
ddiff --git a/test/message-deleted.test.js b/test/message-deleted.test.js
iindex 065baac..4e41c96 100644
---- a/test/message-deleted.test.js
++++ b/test/message-deleted.test.js
@@@ -1,7 +1,7 @@
'use strict';
/**
* chat.message.deleted: every way a public-room message is deleted (one message by a moderator or
-- * through Live's bridge, a user's / anon's / relay user's history, a time-range purge, the
++ * through Chat's moderation ingress, a user's / anon's / relay user's history, a time-range purge, the
* auto-delete sweep) adds one outbox envelope per 500 ids, in the same transaction, carrying only
* ids and payload.redacts. The relay publishes it after the message's chat.message.created.
*
@@@ -78,16 +78,16 @@ t('boot (a real OpenVibe.Events on a reserved port when one is checked out next
}
});
--t('one message deleted (a moderator, or Live’s bridge): one deletion event, ids only', async () => {
++t('one message deleted (a moderator, or Live’s /api/mod through the ingress): one deletion event, ids only', async () => {
const id = await say(SECRET);
const from = await outboxSeq();
await h.db.deleteChatMessage(id, admin.id);
const [d] = await deletions(from);
assertDeletion(d, [id]);
assert.ok(!JSON.stringify(d).includes('555-0199') && !JSON.stringify(d).includes('anon7'));
-- // Through the bridge, the way Live's /api/mod forwards it.
-- const id2 = await say('via the bridge');
-- const r = await h.http('POST', '/internal/live/calls', { token: h.serviceToken(['chat.live_bridge.write']), body: { boot: 'b1', ops: [{ seq: 1, op: 'db', args: ['deleteChatMessage', id2, admin.id] }] } });
++ // Through Chat's moderation ingress, the way Live's /api/mod forwards it.
++ const id2 = await say('via the ingress');
++ const r = await h.http('POST', '/internal/chat/moderation', { token: h.serviceToken(['chat.moderation.write']), body: { key: 'mod:delete:1', action: 'delete-message', id: id2, deleted_by: admin.id } });
assert.strictEqual(r.status, 200, r.text);
assertDeletion((await deletions(from)).at(-1), [id2]);
const before = await outboxSeq();
ddiff --git a/test/parity.test.js b/test/parity.test.js
iindex 0963d2a..f299cb1 100644
---- a/test/parity.test.js
++++ b/test/parity.test.js
@@@ -7,7 +7,7 @@
* (A) holds an active subscription to the channel (the stub Live's GET /subscriber); the driver sets
* network blocks itself (network.block.changed, as the Events consumer applies it).
*
-- * On top, what only a test can reach: Live's moderator paths through the bridge (the context-menu
++ * On top, what only a test can reach: Live's moderator paths on Chat's own calls (the context-menu
* ban that disconnects, the single-message delete) and the global feed's cursor read (Live's
* floating widget, openvibe.chat), sub-only failing closed when Live cannot be asked, slow mode as
* the saved setting (a restart of every cache, the dashboard's value), /clear's answer, blocks
@@@ -24,12 +24,7 @@ const t = suite('parity');
let h, p, streamer, mod, alice, bob, dave, channelId, streamId;
const IPS = { a: '198.51.100.101', b: '198.51.100.102', mod: '198.51.100.103', streamer: '198.51.100.104', anon: '198.51.100.105', dave: '198.51.100.106' };
const results = [];
--let bridgeSeq = 0;
let blockRev = 0;
--const bridge = (ops) => h.http('POST', '/internal/live/calls', {
-- token: h.serviceToken(['chat.live_bridge.write', 'chat.message.send']),
-- body: { boot: 'parity-live', ops: ops.map((o) => ({ seq: ++bridgeSeq, ...o })) },
--});
/** 8 kHz 8-bit mono WAV of `seconds`. */
function wav(seconds) {
@@@ -111,7 +106,7 @@ t('dm: delivery keeps the participant rule (chatServer.sendDm checks dm.isPartic
await p.closeAll();
});
--t('ban from Live’s context menu (/api/mod/stream-ban → the bridge): the socket is closed, the user list drops them, they cannot talk', async () => {
++t('ban from Live’s context menu (/api/mod/stream-ban → disconnectUser): the socket is closed, the user list drops them, they cannot talk', async () => {
const watcher = await p.join('a');
const daveTokens = { ...p.tokens, dave: dave.token };
const q = new parity.Parity({ driver, tokens: daveTokens, streamId });
@@@ -121,8 +116,8 @@ t('ban from Live’s context menu (/api/mod/stream-ban → the bridge): the sock
await watcher.next(listed);
// Live writes the bans row, then asks Chat to disconnect (Live server/admin/mod-routes.js).
h.live.addBan({ stream_id: streamId, user_id: dave.id });
-- const r = await bridge([{ op: 'disconnectUser', args: [{ userId: dave.id, ip: null, streamId }] }]);
-- assert.strictEqual(r.status, 200, r.text);
++ await h.ctx.invalidateBans();
++ h.chatServer.disconnectUser({ userId: dave.id, ip: null, streamId });
await daveWs.next((m) => m.type === 'system' && m.message === 'You have been banned.');
await new Promise((res) => (daveWs.readyState === 3 ? res() : daveWs.once('close', res)));
await watcher.next((m) => m.type === 'users-list' && !m.users.logged.some((u) => u.username === 'dave'));
@@@ -147,12 +142,9 @@ t('reconnect in global chat (Live’s floating widget, openvibe.chat): Live’s
const m1 = await p.saw(stayer, doomed); await p.saw(leaver, doomed);
viewB.frames(leaver.all.slice(fromB));
await p.left(leaver);
-- // Live's /api/mod/delete-message: the row through the bridge, then the frame to the global feed.
-- const r = await bridge([
-- { op: 'db', args: ['deleteChatMessage', m1.id, mod.id] },
-- { op: 'broadcastGlobal', args: [{ type: 'delete-messages', ids: [m1.id] }] },
-- ]);
-- assert.ok(r.body.results.every((x) => x.ok), r.text);
++ // Live's /api/mod/delete-message: the row, then the frame to the global feed.
++ await h.db.deleteChatMessage(m1.id, mod.id);
++ await h.chatServer.broadcastGlobal({ type: 'delete-messages', ids: [m1.id] });
await stayer.next((m) => m.type === 'delete-messages' && m.ids.includes(m1.id));
const later = p.text('global line while B was away');
await p.say(stayer, later);
ddiff --git a/test/rest.test.js b/test/rest.test.js
iindex 542d8f5..8f10d0d 100644
---- a/test/rest.test.js
++++ b/test/rest.test.js
@@@ -255,7 +255,7 @@ t('CORS: no *.openvibe.tools wildcard — only exact origins get credentialed CO
const apex = await probe('https://openvibe.tools');
assert.strictEqual(apex.headers.get('access-control-allow-origin'), 'https://openvibe.tools', 'the listed apex keeps working');
// The same list decides the exported check (and the WebSocket upgrade).
-- const stub = { chatServer: { getTotalConnections: () => 0 }, bridge: (q, s, n) => n() };
++ const stub = { chatServer: { getTotalConnections: () => 0 } };
const { isAllowedOrigin, allowedOrigins } = createApp(stub);
assert.strictEqual(isAllowedOrigin('https://pastes.openvibe.tools'), false);
assert.strictEqual(isAllowedOrigin('https://openvibe.live'), true);
ddiff --git a/test/restart-resume.test.js b/test/restart-resume.test.js
iindex 5136f63..97b148c 100644
---- a/test/restart-resume.test.js
++++ b/test/restart-resume.test.js
@@@ -8,9 +8,9 @@
* cursor Live's chat.js keeps (the highest message id they have shown, from history pages and live
* frames) and, after the restart, reconnect, join, and read `?after_id=<cursor>`. What they end up
* with must be exactly the room's rows in the database: nothing missing (messages written while
-- * the reader was away, before and after the restart, by people and through Live's bridge) and
-- * nothing twice (no row at or under the cursor comes back, and every live frame carries the row's
-- * real id — including a bridge broadcast whose insert was acknowledged before the restart).
++ * the reader was away, before and after the restart, by people and through Live's ingress) and
++ * nothing twice (no row at or under the cursor comes back, every live frame carries the row's real
++ * id, and an ingress message Live sends again after the restart with its key is one row).
*/
const assert = require('assert');
const path = require('path');
@@@ -18,12 +18,11 @@ const { boot, suite } = require('./helpers');
const t = suite('restart-resume');
let h, chat, port, streamer, posters, reader, channelId, streamId;
--const REF = -(2 ** 40) - 1;
--const BOOT = 'live-boot-restart';
--let BRIDGE;
++let SEND;
const sockets = [];
--const calls = (ops) => h.http('POST', '/internal/live/calls', { token: BRIDGE, body: { boot: BOOT, ops: ops.map((o, i) => ({ seq: i + 1, ...o })) } });
++// Live's AI viewer posting through Chat's ingress (POST /internal/chat/messages), keyed like Live's.
++const ingest = (key, message) => h.http('POST', '/internal/chat/messages', { token: SEND, body: { key, stream_id: streamId, username: 'ChatBot', message, message_type: 'chat', source_platform: 'ai', mirror: true } });
// What the database says the rooms hold (this process's own handle on the same database: Chat is another process).
async function roomRows(where, params) {
@@@ -82,7 +81,7 @@ const readers = {};
t('boot the stubs, then Chat as its own process', async () => {
h = await boot();
-- BRIDGE = h.serviceToken(['chat.live_bridge.write', 'chat.message.send']);
++ SEND = h.serviceToken(['chat.message.send']);
streamer = h.addUser('streamer', { role: 'streamer' });
reader = h.addUser('reader');
posters = ['ann', 'ben', 'cat', 'dan', 'eve', 'fay'].map((n) => h.addUser(n));
@@@ -106,16 +105,12 @@ t('readers load the rooms and follow live frames', async () => {
page.body.messages.forEach((m) => r.show(m));
assert.strictEqual(r.cursor, page.body.latest_id, `${r.name}: the page's latest_id is the cursor`);
}
-- // A live message and a bridge message (Live's AI viewer) reach the connected readers.
++ // A live message and an ingress message (Live's AI viewer) reach the connected readers.
const id2 = await post(posters[1], '198.51.100.11', 'before 2');
-- const b = await calls([
-- { op: 'db', ref: REF, key: 'live:r1', args: ['saveChatMessage', { stream_id: streamId, username: 'ChatBot', message: 'bridge before', message_type: 'chat', source_platform: 'ai' }] },
-- { op: 'broadcastToStream', args: [streamId, { type: 'chat', id: REF, username: 'ChatBot', message: 'bridge before' }] },
-- { op: 'forwardToGlobal', args: [streamId, { type: 'chat', id: REF, username: 'ChatBot', message: 'bridge before' }] },
-- ]);
-- assert.ok(b.body.results.every((x) => x.ok), b.text);
-- await readers.stream.ws.next((m) => m.type === 'chat' && m.message === 'bridge before');
-- await readers.global.ws.next((m) => m.type === 'chat' && m.message === 'bridge before');
++ const b = await ingest('live:r1', 'ingress before');
++ assert.strictEqual(b.status, 200, b.text);
++ assert.strictEqual((await readers.stream.ws.next((m) => m.type === 'chat' && m.message === 'ingress before')).id, b.body.id);
++ assert.strictEqual((await readers.global.ws.next((m) => m.type === 'chat' && m.message === 'ingress before')).id, b.body.id);
assert.ok(readers.stream.shown.has(id2) && readers.global.shown.has(id2));
// The channel reader is a page that polls (no socket): it read before these two.
});
@@@ -124,13 +119,10 @@ let beforeMax;
t('the stream reader drops off; messages keep coming; Chat restarts (SIGTERM, new process)', async () => {
readers.stream.ws.close();
await post(posters[2], '198.51.100.12', 'missed while away');
-- // Live's bridge: an insert acknowledged before the restart whose broadcast is sent after it
-- // (Live's queue split the batch). The broadcast must still carry the real id.
-- const ins = await calls([{ op: 'db', ref: REF - 1, key: 'live:r2', args: ['saveChatMessage', { stream_id: streamId, username: 'ChatBot', message: 'bridge across restart', message_type: 'chat', source_platform: 'ai' }] }]);
-- assert.ok(ins.body.results[0].ok, ins.text);
-- // And one whose response Live never got: it will send it again after the restart.
-- const lost = await calls([{ op: 'db', ref: REF - 3, key: 'live:r4', args: ['saveChatMessage', { stream_id: streamId, username: 'ChatBot', message: 'bridge retried', message_type: 'chat', source_platform: 'ai' }] }]);
-- assert.ok(lost.body.results[0].ok, lost.text);
++ // An ingress message whose response Live never got: it will send it again after the restart.
++ const lost = await ingest('live:r4', 'ingress retried');
++ assert.strictEqual(lost.status, 200, lost.text);
++ await readers.global.ws.next((m) => m.type === 'chat' && m.message === 'ingress retried');
beforeMax = (await roomRows('1 = 1', [])).reduce((m, x) => Math.max(m, x.id), 0);
const closed = new Promise((r) => readers.global.ws.on('close', r));
@@@ -142,35 +134,21 @@ t('the stream reader drops off; messages keep coming; Chat restarts (SIGTERM, ne
chat = await h.spawnChat({ port });
});
--t('after the restart: the split broadcast of an acknowledged insert, a retried insert, new messages', async () => {
++t('after the restart: a retried ingress message, new messages', async () => {
// The global reader reconnects at once (chat.js reconnects on close); the stream reader later.
await openReader(readers.global, { ip: '198.51.100.21' });
-- // A batch of a later flush: only the broadcast, carrying the placeholder of an insert the old
-- // process acknowledged (Live deleted it from its outbox; it is never sent again).
-- const late = await calls([{ op: 'forwardToGlobal', args: [streamId, { type: 'chat', id: REF - 1, username: 'ChatBot', message: 'bridge across restart' }] }]);
-- assert.ok(late.body.results[0].ok, late.text);
-- const frame = await readers.global.ws.next((m) => m.type === 'chat' && m.message === 'bridge across restart');
-- assert.strictEqual(frame.id, (await roomRows("message = 'bridge across restart'", []))[0].id, 'the placeholder maps to the real id after a restart');
--
-- // The insert whose answer was lost comes again with its key and its broadcast: applied once,
-- // and the broadcast carries the first insert's id.
-- const retry = await calls([
-- { op: 'db', ref: REF - 3, key: 'live:r4', args: ['saveChatMessage', { stream_id: streamId, username: 'ChatBot', message: 'bridge retried', message_type: 'chat', source_platform: 'ai' }] },
-- { op: 'forwardToGlobal', args: [streamId, { type: 'chat', id: REF - 3, username: 'ChatBot', message: 'bridge retried' }] },
-- ]);
-- assert.ok(retry.body.results.every((x) => x.ok), retry.text);
-- const once = await roomRows("message = 'bridge retried'", []);
-- assert.strictEqual(once.length, 1, 'a retried insert is applied once across the restart');
-- assert.strictEqual(retry.body.results[0].result.lastInsertRowid, once[0].id);
-- assert.strictEqual((await readers.global.ws.next((m) => m.type === 'chat' && m.message === 'bridge retried')).id, once[0].id);
++ // The message whose answer was lost comes again with its key: applied once, with the first id.
++ const retry = await ingest('live:r4', 'ingress retried');
++ assert.strictEqual(retry.status, 200, retry.text);
++ const once = await roomRows("message = 'ingress retried'", []);
++ assert.strictEqual(once.length, 1, 'a retried message is applied once across the restart');
++ assert.strictEqual(retry.body.id, once[0].id);
const after1 = await post(posters[3], '198.51.100.13', 'after restart 1');
assert.ok(after1 > beforeMax, 'ids keep increasing across the restart');
-- await calls([
-- { op: 'db', ref: REF - 2, key: 'live:r3', args: ['saveChatMessage', { stream_id: streamId, username: 'ChatBot', message: 'bridge after', message_type: 'chat', source_platform: 'ai' }] },
-- { op: 'forwardToGlobal', args: [streamId, { type: 'chat', id: REF - 2, username: 'ChatBot', message: 'bridge after' }] },
-- ]);
-- await readers.global.ws.next((m) => m.type === 'chat' && m.message === 'bridge after');
++ const after2 = await ingest('live:r3', 'ingress after');
++ assert.strictEqual(after2.status, 200, after2.text);
++ assert.strictEqual((await readers.global.ws.next((m) => m.type === 'chat' && m.message === 'ingress after')).id, after2.body.id);
});
for (const [name, room] of [['stream', streamRoom], ['global', globalRoom], ['channel', channelRoom]]) {
@@@ -197,7 +175,7 @@ for (const [name, room] of [['stream', streamRoom], ['global', globalRoom], ['ch
assert.deepStrictEqual([...r.shown.keys()].sort((a, b) => a - b), want.map((x) => x.id), `${name}: exactly the room's rows`);
const seen = new Set();
for (const text of r.texts) { assert.ok(!seen.has(text), `${name}: "${text}" shown twice`); seen.add(text); }
-- for (const text of ['before 1', 'before 2', 'bridge before', 'missed while away', 'bridge across restart', 'bridge retried', 'after restart 1', 'bridge after']) {
++ for (const text of ['before 1', 'before 2', 'ingress before', 'missed while away', 'ingress retried', 'after restart 1', 'ingress after']) {
assert.ok(seen.has(text), `${name}: "${text}" missing`);
}
// The next read from the new cursor is empty.
ddiff --git a/test/ws-protocol.test.js b/test/ws-protocol.test.js
iindex fa0a94b..286adc6 100644
---- a/test/ws-protocol.test.js
++++ b/test/ws-protocol.test.js
@@@ -5,7 +5,7 @@
* room (delivery, global cross-feed, persistence with the Network subject, outbox, Live's
* reactions and the coin reply), history, /w (DMs replaced whispers), the DM participant check on
* delivery, moderation (/ban /timeout /unban /slow /clear with moderator checks), /tts, /paste,
-- * !-commands routed to Live (arena reply through the bridge, media queue), and the upgrade guards
++ * !-commands routed to Live (arena reply to the sending socket, media queue), and the upgrade guards
* (origin allow-list, IP bans, the per-address cap).
*/
const assert = require('assert');
@@@ -292,15 +292,14 @@ t('/paste creates the paste through Live and announces it', async () => {
assert.strictEqual(fx.body.stream_id, streamId);
});
--t('!hype goes to Live; Live answers this socket through the bridge (sendToConn)', async () => {
++t('!hype goes to Live with the socket’s conn_id; sendToConn answers that socket', async () => {
await h.sleep(1100);
aliceWs.sendJson({ type: 'chat', message: '!hype' });
await h.sleep(200);
const fx = h.live.effects.find((e) => e.name === 'arena-command');
assert.strictEqual(fx.body.cmd, '!hype');
assert.strictEqual(fx.body.client.user.id, alice.id);
-- const res = await h.http('POST', '/internal/live/calls', { token: h.serviceToken(['chat.live_bridge.write']), body: { boot: 'b1', ops: [{ seq: 1, op: 'sendToConn', args: [fx.body.client.conn_id, { type: 'system', message: 'Hyped!' }] }] } });
-- assert.strictEqual(res.status, 200);
++ h.chatServer.sendToConn(fx.body.client.conn_id, { type: 'system', message: 'Hyped!' });
await aliceWs.next((m) => m.type === 'system' && m.message === 'Hyped!');
});