openvibers/openvibe.chat Commits

Calls can run in Chat (WS-I task 1): Live's call server moved to server/calls/ with the same /ws/call protocol and limits, Live's voice-channel and /api/streams/:id/call routes on the same paths and shapes, loopback /internal/calls/stream-channel for Live's stream hooks (bridge service token, chat.live_bridge.write), and a calls table with the pending/ringing/active/ended/missed/declined/failed lifecycle; off until CHAT_CALLS=1 so nothing changes before the cutover in docs/calls-cutover.md

OP OpenVibers committed 25 Sept 2026 · 603d0b482d · 18 files, +2010 −17


Changes

ddiff --git a/.env.example b/.env.example
iindex 97d6231..9bbba40 100644
---- a/.env.example
++++ b/.env.example
@@@ -84,5 +84,11 @@ SOUNDS_PATH=./data/sounds
#MAX_SOUNDS_PER_CHANNEL=150
#MAX_SOUNDS_PER_UPLOADER_PER_CHANNEL=10
++# Voice/video calls (server/calls/, Live's call server moved): off until the calls cutover
++# (docs/calls-cutover.md). Off = /api/streams/… 404, /ws/call refused, /internal/calls/* 409.
++#CHAT_CALLS=1
++# A direct call nobody answers is missed after this long (ms); the caller hears "no answer".
++#CALL_RING_TIMEOUT_MS=45000
++
# Optional diagnostics
#DEBUG_DM_DELIVERY=1
ddiff --git a/README.md b/README.md
iindex b8e5a58..8226737 100644
---- a/README.md
++++ b/README.md
@@@ -138,6 +138,31 @@ sudo node --env-file=/etc/openvibe/chat.env scripts/subscribe-events.js --enable
failed, the last type and outcome); `SELECT * FROM deploy_releases ORDER BY created_at DESC` shows
which path delivered each deploy first and when the other one arrived.
++### Calls
++
++Live's voice/video calls moved here (roadmap WS-I task 1, `server/calls/`), **off until the calls
++cutover** (`CHAT_CALLS=1`, `docs/calls-cutover.md`):
++
++- **`/ws/call`** — Live's call server with the same protocol: `?channelId=` (or a stream number),
++ signalling only (full-mesh WebRTC; `offer`/`answer`/`ice-candidate` relayed, `mute`, `camera-off`,
++ `speaking`, `auth-update`, `force-mute`, `force-camera-off`, `kick`, `ban`, `unban`, `end-call`), the
++ permanent `public` lobby, one temporary channel per person (private for a 1:1 call), stream-linked
++ channels, 8 participants, 3 sockets per address, a one-minute kick cooldown, per-channel bans, empty
++ channels deleted after an hour. Sign-in is Chat's (as on `/ws/chat`); accounts, streams and cosmetics
++ come through `live-context`; the voice-channel list and call invites go out through Chat's chat server.
++- **REST on Live's paths** (`server/calls/routes.js`): `GET/POST /api/streams/voice-channels`,
++ `GET/DELETE /api/streams/voice-channels/:channelId`, `POST …/call-user` and `…/call-user/respond`,
++ `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.
++- **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
++ channel's occupancy is a `channel` / `stream` session, `active` while anyone is in, `ended` when it
++ empties or closes. A restart closes what was open (reason `restart`).
++
### VIP member badges
A member's messages in a creator's room (stream or offline channel chat) carry that creator's VIP
@@@ -212,7 +237,7 @@ node scripts/migrate-chat-preferences.js --rollback /root/chat-prefs-<date>.json
```bash
npm install
cp .env.example .env # OV_LIVE_INTERNAL_URL, Network URLs, OV_OAUTH_CLIENT_SECRET, SOUNDS_PATH
--npm start # 127.0.0.1:4400 — /ws/chat, /api/{chat,dm,tts,sounds}, /health, /ready
++npm start # 127.0.0.1:4400 — /ws/chat, /api/{chat,dm,tts,sounds}, /health, /ready (CHAT_CALLS=1: /ws/call, /api/streams/…)
npm test # Node 22; stub Live and Network in-process
node scripts/import-from-live.js --live-db /tmp/live-snapshot.db --dry-run
node scripts/parity-check.js --live https://openvibe.live --chat http://127.0.0.1:4401 --before "…"
@@@ -243,9 +268,10 @@ server/events/consumer.js POST /internal/events: Chat's Events subscriptions (l
server/events/subscriptions.js creates them at boot when missing; list/disable/enable for scripts/subscribe-events.js
server/net/service-auth.js service tokens: client (Chat → others) and guard (others → Chat)
server/prefs/ chat preferences in the Network user module chat.preferences (routes, cache, migration from Live)
++server/calls/ moved from Live: the call server (/ws/call), its REST routes, Live's stream hooks (/internal/calls), the calls lifecycle
server/db/ schema.sql, database.js (Live's chat functions, same names and arguments)
scripts/ import-from-live, parity-check, mirror-flush, migrate-chat-preferences, subscribe-events
--docs/ cutover.md, live-patch.diff, capabilities-proposal/
++docs/ cutover.md, calls-cutover.md, live-patch.diff, capabilities-proposal/
```
## Owns
@@@ -257,8 +283,9 @@ docs/ cutover.md, live-patch.diff, capabilities-proposal/
- call signalling metadata and the `pending/ringing/active/ended/missed/declined/failed` lifecycle
(Wave 6 moved messages, DMs, TTS, sounds and the chat side of moderation, and gave TTS and sounds a
--persisted queue with skip/clear/failed states; bans, channel moderation settings, the media-request
--queue and calls are still Live's and reached through `live-context`.)
++persisted queue with skip/clear/failed states; bans, channel moderation settings and the media-request
++queue are still Live's and reached through `live-context`. Calls are ported with their lifecycle,
++`server/calls/`, and switch over with `docs/calls-cutover.md`.)
## Does not own
@@@ -296,7 +323,7 @@ Events: `chat.message.created`, `chat.message.deleted`, `chat.dm.created`, `chat
- 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`)*
--- a call row without a working signalling/media path is not parity — *calls are still Live's (`/ws/call`)*
++- 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*
## Launch rule
ddiff --git a/STATUS.json b/STATUS.json
iindex f933a90..325a232 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -16,6 +16,7 @@
"liveAuthorityFlag": "CHAT_AUTHORITY=chat is set in production live.env since the cutover (2026-09-23 02:03 UTC); Live's chat tables are a read mirror Chat keeps current",
"vip": "Member badges from OpenVibe.VIP (server/vip/badges.js: entitlements/check with product chat, cached 60 s, never blocks sending, fails closed). In code only, not deployed; needs the Network grant chat vip.entitlement.check openvibe.vip and Live's chat UI to render vip_badge / chat_vip_badge",
"preferences": "Chat preferences in the Network user module chat.preferences (server/prefs/: GET/PUT /api/chat/preferences with Chat's service token, cached 60 s, If-Match on every write; scripts/migrate-chat-preferences.js moves the choices in Live's user_preferences). In code only, not deployed; needs Network with the chat.preferences handoff and the grants chat network.modules.read/write on chat.preferences, then the migration, then Live's chat.js calling /api/chat/preferences",
++ "calls": "Live's call server, call routes and stream hooks ported to server/calls/ with a calls lifecycle table (WS-I task 1). In code only, off (CHAT_CALLS unset); the switch is CHAT_CALLS=1, Live CALLS_AUTHORITY=chat and the nginx include deploy/nginx/calls/openvibe.live-calls.locations.conf (docs/calls-cutover.md)",
"deployed": true,
"deployment": "openvibe-ovh /opt/openvibe.chat, unit openvibe-chat, 127.0.0.1:4400 (loopback); public only through openvibe.live's nginx include for the chat paths; release 45f2b1f; EVENTS_URL set (chat.message.created reaches OpenVibe.Events); restore drill passed 2026-09-23",
"updated": "2026-09-23"
ddiff --git a/deploy/nginx/calls/openvibe.live-calls.locations.conf b/deploy/nginx/calls/openvibe.live-calls.locations.conf
nnew file mode 100644
iindex 0000000..1d11881
---- /dev/null
++++ b/deploy/nginx/calls/openvibe.live-calls.locations.conf
@@@ -0,0 +1,53 @@
++# OpenVibe.Chat call locations (roadmap WS-I task 1) — include INSIDE the `server { listen 443 …
++# server_name openvibe.live … }` block of /etc/nginx/sites-enabled/openvibe.live.conf, next to the
++# chat include and above `location /api/`:
++#
++# include /opt/openvibe.chat/deploy/nginx/calls/openvibe.live-calls.locations.conf;
++#
++# This file sits in calls/ on purpose: Live's vhost already includes deploy/nginx/*.locations.conf,
++# so a file named like that one directory up would send calls to Chat at the first nginx reload after
++# a Chat deploy — before Chat runs with CHAT_CALLS=1 and Live with CALLS_AUTHORITY=chat. Adding the
++# include line (+ nginx -t + reload) is the flip; removing it is the rollback (docs/calls-cutover.md).
++#
++# Headers and limits are Live's for the same paths (the /api/ zone and burst, the client address
++# from the connection only, upgrade for the socket). `^~` keeps every regex location (the static-asset
++# one) from taking the prefix paths. Everything else under /api/streams/ stays on Live.
++
++location ^~ /ws/call {
++ proxy_pass http://127.0.0.1:4400;
++ proxy_http_version 1.1;
++ proxy_set_header Host $host;
++ proxy_set_header X-Real-IP $remote_addr;
++ proxy_set_header X-Forwarded-For $remote_addr;
++ proxy_set_header CF-Connecting-IP $remote_addr;
++ proxy_set_header X-Forwarded-Proto $scheme;
++ proxy_set_header Upgrade $http_upgrade;
++ proxy_set_header Connection "upgrade";
++}
++
++# GET/POST /api/streams/voice-channels, GET/DELETE …/:channelId, POST …/call-user and …/call-user/respond
++location ^~ /api/streams/voice-channels {
++ limit_req zone=api_limit burst=40 nodelay;
++ client_max_body_size 10m;
++ proxy_pass http://127.0.0.1:4400;
++ proxy_http_version 1.1;
++ proxy_set_header Host $host;
++ proxy_set_header X-Real-IP $remote_addr;
++ proxy_set_header X-Forwarded-For $remote_addr;
++ proxy_set_header CF-Connecting-IP $remote_addr;
++ proxy_set_header X-Forwarded-Proto $scheme;
++}
++
++# GET/PUT /api/streams/<id>/call. A regex location: it wins over Live's plain `location /api/` prefix,
++# and only this exact shape leaves Live (/api/streams/<id>, /<id>/end, … do not match).
++location ~ ^/api/streams/[0-9]+/call$ {
++ limit_req zone=api_limit burst=40 nodelay;
++ client_max_body_size 10m;
++ proxy_pass http://127.0.0.1:4400;
++ proxy_http_version 1.1;
++ proxy_set_header Host $host;
++ proxy_set_header X-Real-IP $remote_addr;
++ proxy_set_header X-Forwarded-For $remote_addr;
++ proxy_set_header CF-Connecting-IP $remote_addr;
++ proxy_set_header X-Forwarded-Proto $scheme;
++}
ddiff --git a/docs/calls-cutover.md b/docs/calls-cutover.md
nnew file mode 100644
iindex 0000000..85276e6
---- /dev/null
++++ b/docs/calls-cutover.md
@@@ -0,0 +1,175 @@
++# Calls cutover — Live → OpenVibe.Chat (roadmap WS-I task 1)
++
++What changes at the cutover: browsers keep calling `https://openvibe.live`, but nginx sends the call
++socket and the call REST routes to OpenVibe.Chat (127.0.0.1:4400), Chat runs with `CHAT_CALLS=1` and
++Live with `CALLS_AUTHORITY=chat`. Nothing in Live's browser code changes (`public/js/call.js`,
++`voice-channels.js`, the ringing overlays in `chat.js`); ICE servers still come from Live's
++`GET /api/auth/ice-servers`. Everything below is reversible.
++
++- [What is served where](#what-is-served-where)
++- [The two switches](#the-two-switches)
++- [Prerequisites](#prerequisites)
++- [Cutover](#cutover)
++- [Checks](#checks)
++- [Rollback](#rollback)
++- [Behaviour that is not byte-for-byte identical](#behaviour-that-is-not-byte-for-byte-identical)
++
++## What is served where
++
++nginx locations (`deploy/nginx/calls/openvibe.live-calls.locations.conf`, included in Live's
++`server { … server_name openvibe.live … }` block next to the chat include, above `location /api/`):
++
++| nginx location | Chat serves (`server/calls/`) |
++| --- | --- |
++| `^~ /ws/call` | the call WebSocket (`?channelId=` — or the legacy `?streamId=` / a bare stream number — and `?token=`; every message type unchanged) |
++| `^~ /api/streams/voice-channels` | `GET/POST /api/streams/voice-channels`, `GET/DELETE /api/streams/voice-channels/:channelId`, `POST /api/streams/voice-channels/call-user`, `POST /api/streams/voice-channels/call-user/respond` |
++| `~ ^/api/streams/[0-9]+/call$` | `GET/PUT /api/streams/:id/call` |
++
++Everything else under `/api/streams/` (go-live, stream end, VODs, channels, …) stays on Live.
++
++Loopback only, never routed: Chat `POST /internal/calls/stream-channel` and
++`DELETE /internal/calls/stream-channel/:streamId` (Live's stream hooks; they refuse anything that came
++through nginx).
++
++**The file is in `deploy/nginx/calls/` on purpose.** Live's vhost already has
++`include /opt/openvibe.chat/deploy/nginx/*.locations.conf;` (the chat cutover). A calls file in that
++directory would route calls to Chat at the first nginx reload after a Chat deploy, before either
++switch below. The flip is adding its own include line; the rollback is removing it.
++
++## The two switches
++
++| Where | Variable | Off (default) | On |
++| --- | --- | --- | --- |
++| Chat, `/etc/openvibe/chat.env` | `CHAT_CALLS=1` | `/api/streams/…` answers 404, `/ws/call` upgrades are refused, `/internal/calls/*` answers 409 `calls.off`; Chat never pushes a voice-channel list or a call invite | Chat serves the paths above and owns the lifecycle rows (`calls`) |
++| Live, `/etc/openvibe/live.env` | `CALLS_AUTHORITY=chat` | Live's call server makes and removes stream voice channels, exactly as before | go-live with a call mode, stream end, the WHIP teardown and the admin force-end call Chat's `/internal/calls/stream-channel` instead |
++
++Chat needs its own switch because its chat sockets and openvibe.chat's `/api/` are already public:
++without it, `POST https://openvibe.chat/api/streams/voice-channels…` would reach the new routes and
++push Chat's voice-channel list and call invites into openvibe.live's chat sockets while calls were
++still Live's.
++
++Live → Chat authentication is the chat bridge's: a Network service token for audience
++`openvibe.chat` carrying `chat.live_bridge.write` (Live already holds that grant since the chat
++cutover; no new grant, no new secret). Live retries a hook after a network error or a 5xx (1 s, 5 s,
++15 s), never after a 4xx, and drops a retry that a later hook for the same stream superseded.
++
++Optional: `CALL_RING_TIMEOUT_MS` (Chat, default 45000) — a direct call nobody answers is `missed`
++after this long and the caller is told (see below).
++
++## Prerequisites
++
++- Chat deployed with `server/calls/` (this release), `curl -s 127.0.0.1:4400/ready | jq .status` →
++ `"ready"`.
++- Live deployed with `server/streaming/calls-authority.js` and the `notify/call-invite` chat effect
++ (the cross-site "X is calling you" notification is Live's to push).
++- Live's bridge works: `journalctl -u openvibe-live --since -1h | grep ChatRemote` shows no refusals.
++- A quiet moment. Voice channels are memory, in Live as in Chat: calls in progress at the flip drop
++ (clients reconnect to Chat and rejoin), user channels are not carried over, and **a stream that is
++ live with calls on at the flip has no voice channel in Chat** until the streamer goes live again
++ (or `PUT /api/streams/:id/call`). Check there is none:
++ ```bash
++ sqlite3 /opt/openvibe.live/data/live.db "select id, user_id, call_mode from streams where is_live = 1 and call_mode is not null"
++ ```
++
++## Cutover
++
++1. **Chat serves calls.** Add `CHAT_CALLS=1` to `/etc/openvibe/chat.env`, restart Chat (chat clients
++ are told it restarts and reconnect):
++ ```bash
++ sudo systemctl restart openvibe-chat
++ journalctl -u openvibe-chat -n 50 | grep CallServer # "Voice channels initialized (serving)"
++ curl -s 127.0.0.1:4400/api/streams/voice-channels | jq '.channels[].id' # "public"
++ curl -s -o /dev/null -w '%{http_code}\n' -X DELETE 127.0.0.1:4400/internal/calls/stream-channel/1 # 401 (mounted, guarded)
++ ```
++2. **Live hands its stream hooks to Chat.** Add `CALLS_AUTHORITY=chat` to `/etc/openvibe/live.env`
++ and restart Live (socket activation: no 502s; boot about 6 s):
++ ```bash
++ sudo systemctl restart openvibe-live
++ until curl -sf https://openvibe.live/api/ready >/dev/null; do sleep 1; done
++ journalctl -u openvibe-live -n 200 | grep 'CALLS_AUTHORITY=chat'
++ ```
++3. **Route the paths.** In `/etc/nginx/sites-enabled/openvibe.live.conf`, next to the chat include:
++ ```nginx
++ include /opt/openvibe.chat/deploy/nginx/calls/openvibe.live-calls.locations.conf;
++ ```
++ On openvibe.chat (`/etc/nginx/sites-available/openvibe.chat.conf` and this repo's reference
++ `deploy/nginx/openvibe.chat.conf`) its `location /api/` already reaches Chat; add the socket:
++ ```nginx
++ location ^~ /ws/call {
++ proxy_pass http://127.0.0.1:4400;
++ proxy_http_version 1.1;
++ proxy_set_header Host $host;
++ proxy_set_header X-Real-IP $remote_addr;
++ proxy_set_header X-Forwarded-For $remote_addr;
++ proxy_set_header CF-Connecting-IP $remote_addr;
++ proxy_set_header X-Forwarded-Proto $scheme;
++ proxy_set_header Upgrade $http_upgrade;
++ proxy_set_header Connection "upgrade";
++ proxy_read_timeout 3600s;
++ proxy_send_timeout 3600s;
++ }
++ ```
++ (Every `proxy_set_header` is repeated inside the location on purpose: a location that sets any of
++ them inherits none from the server block.) Then:
++ ```bash
++ sudo nginx -t && sudo systemctl reload nginx
++ ```
++ Between steps 2 and 3, `/ws/call` still reaches Live's (now empty) call server; streams that go live
++ in that window get their channel in Chat, where browsers land after the reload.
++
++## Checks
++
++- `curl -s https://openvibe.live/api/streams/voice-channels | jq '.channels[].id'` → `"public"` and
++ whatever users created since the flip.
++- Two test accounts in the Chat tab: join the Public voice channel, hear and see each other, mute
++ shows on the other side, leave. `sqlite3 /var/lib/openvibe-chat/chat.db "select id, kind, state,
++ channel_id, end_reason from calls order by id desc limit 5"` shows the `channel` session `active`,
++ then `ended` / `empty`.
++- Call one test account from the other (user menu → Call): the callee rings (and gets the
++ notification), accept → connected; the row is `direct`, `ringing` → `active` → `ended`. Decline →
++ `declined`; busy → `declined` / `busy`; let it ring out → `missed` (`no-answer` from the callee's
++ browser at 30 s, else `timeout` at 45 s, and the caller sees "No answer").
++- A test stream with calls on: go live with a call mode, the stream page's call joins
++ `stream-<id>`, `GET /api/streams/<id>/call` shows the mode and participants; end the stream →
++ everyone gets `call-ended`, the row is `stream` / `ended` / `stream_ended`. An admin force-end does
++ the same. `journalctl -u openvibe-live | grep '\[Calls\]'` shows no "Chat refused".
++- Moderation: in a test channel, kick (the kicked account cannot rejoin for a minute) and ban/unban.
++
++After one release without a rollback: remove Live's call server and routes (a later change) and move
++the include into `deploy/nginx/` so the glob carries it.
++
++## Rollback
++
++1. Remove the include line from openvibe.live's nginx (and the `/ws/call` location from
++ openvibe.chat's); `sudo nginx -t && sudo systemctl reload nginx`. Browsers are back on Live's
++ `/ws/call` and routes; `call.js` reconnects with backoff.
++2. Remove `CALLS_AUTHORITY` from `/etc/openvibe/live.env`, restart Live. From the next go-live on,
++ Live's call server makes stream channels again (the ones Chat held are not copied back: a stream
++ live at that moment needs its streamer to go live again).
++3. Optional: remove `CHAT_CALLS` from `/etc/openvibe/chat.env` and restart Chat, so it stops
++ answering the call routes on openvibe.chat too. Chat keeps its `calls` rows.
++
++## Behaviour that is not byte-for-byte identical
++
++The wire protocol is the same: query parameters, every frame type and field, the REST paths, bodies,
++answers and error texts. What differs:
++
++- **The ring times out on the server.** A direct call nobody answers becomes `missed` after
++ `CALL_RING_TIMEOUT_MS` (45 s), and the caller gets the `vc-call-response` frame with
++ `status: 'no-answer'` — the same frame the callee's browser sends when it gives up at 30 s. With an
++ offline callee, Live left the caller ringing until they cancelled.
++- **Client address.** The per-address cap (3) and anonymous ids use Chat's rule (CF-Connecting-IP only
++ when nginx's peer is Cloudflare, as on `/ws/chat`); Live's call server believed CF-Connecting-IP and
++ X-Forwarded-For from anyone.
++- **Socket setup is asynchronous.** Sign-in is Chat's cached resolution through Live (and the stream
++ and cosmetics are warmed, at most 2.5 s on a cold cache); frames a client sends before its `welcome`
++ are handled after it, in order. `auth-update` resolves its token the same way.
++- **A reconnect as the only participant** keeps a reachable room (Live's replaced-socket path dropped
++ the room from its map, so the next person to join landed in a different one).
++- **Call mode** in `GET /api/streams/:id/call` is the mode of the stream's channel in Chat. Live still
++ writes `streams.call_mode` at go-live, but Chat's `PUT` does not write it back (nothing else reads it).
++- **Cosmetics** (`nameFX`, `particleFX`, `hatFX`) come from live-context's decor cache (60 s).
++- **Pushes** (the voice-channel list, invites, responses) go to Chat's chat sockets directly; before
++ they went Live → bridge → Chat.
++- **Every call is a row** of `calls` (lifecycle in `server/calls/lifecycle.js`); a Chat restart closes
++ what was open (`pending` → `failed`, `ringing` → `missed`, `active` → `ended`, reason `restart`).
ddiff --git a/docs/cutover.md b/docs/cutover.md
iindex 68b3624..d7cec29 100644
---- a/docs/cutover.md
++++ b/docs/cutover.md
@@@ -29,7 +29,8 @@ nginx locations (`deploy/nginx/openvibe.live-chat.locations.conf`, included in L
Stay on Live (do not route): `/api/chat-ai/*` (the trailing slash in `/api/chat/` keeps it out),
`/api/emotes/*`, `/api/channels/*` (channel moderation dashboard), `/api/mod/*`, `/api/media/*`,
--`/ws/call`, `/ws/broadcast`, `/ws/control`, and everything else.
++`/ws/call` (until the calls cutover, `docs/calls-cutover.md`), `/ws/broadcast`, `/ws/control`, and
++everything else.
Loopback only, never routed: Chat `/internal/live/*`, `/health`, `/ready`; Live
`/internal/chat-context/*`, `/internal/chat-effects/*` (both refuse anything that came through
ddiff --git a/server/app.js b/server/app.js
iindex e998f83..b271267 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -2,7 +2,8 @@
* OpenVibe.Chat — HTTP app and WebSocket upgrade handling.
*
* Browsers reach Chat on Live's origin: nginx sends /ws/chat and the chat REST prefixes
-- * (/api/chat/, /api/dm/, /api/tts/, /api/sounds) here (docs/cutover.md), so the guards Live ran
++ * (/api/chat/, /api/dm/, /api/tts/, /api/sounds) here (docs/cutover.md), and at the calls cutover
++ * /ws/call and the call routes under /api/streams (docs/calls-cutover.md), so the guards Live ran
* in front of those routes run here too, with Live's values: credentialed CORS for an explicit
* list of origins (never a wildcard), the /api rate limit, IP/network bans (admins exempt) and the ov_banned cookie, the
* WebSocket origin allow-list and IP-ban check at upgrade.
@@@ -59,7 +60,7 @@ function getAllowedOrigins() {
return allowed;
}
--function createApp({ chatServer, bridge, mirror, relay, events = null }) {
++function createApp({ chatServer, bridge, mirror, relay, events = null, callServer = null }) {
const allowedOrigins = getAllowedOrigins();
const app = express();
app.disable('x-powered-by');
@@@ -95,6 +96,8 @@ function createApp({ chatServer, bridge, mirror, relay, events = null }) {
// OpenVibe.Events deliveries (live.release.deployed, network.module.updated): signature v2 over
// the raw body, so this router reads the body itself (server/events/consumer.js).
if (events) app.use('/internal/events', events.router);
++ // Live's stream hooks for calls (CALLS_AUTHORITY=chat): stream voice channels (server/calls/internal.js).
++ app.use('/internal/calls', express.json({ limit: '64kb' }), require('./calls/internal').createInternalRouter());
app.get('/health', (req, res) => res.json({ ok: true, service: 'chat' }));
// Readiness in the openvibe-shared/ready shape (status ready/degraded/not_ready, named checks):
// 503 only when the required check fails. `db` is Chat's own database, which it cannot serve
@@@ -208,6 +211,10 @@ function createApp({ chatServer, bridge, mirror, relay, events = null }) {
app.use('/api/dm', require('./chat/dm-routes'));
app.use('/api/tts', require('./chat/tts-routes'));
app.use('/api/sounds', require('./chat/sounds-routes'));
++ // Calls on Live's paths (server/calls/routes.js): /api/streams/voice-channels… and /api/streams/:id/call.
++ // Nothing answers here until CHAT_CALLS is on (docs/calls-cutover.md).
++ const callRoutes = require('./calls/routes');
++ app.use('/api/streams', (req, res, next) => (config.calls.enabled ? callRoutes(req, res, next) : next()));
// openvibe.chat, the site: pages, sign-in, the Frame's /shared/ files (server/web/).
app.use(require('./web/pages').createWebRoutes({ config }));
@@@ -221,7 +228,7 @@ function createApp({ chatServer, bridge, mirror, relay, events = null }) {
if (!res.headersSent) res.status(500).json({ error: 'Internal server error' });
});
-- /** Live's upgrade guard for /ws/chat: exact origin allow-list, then IP / network bans. */
++ /** Live's upgrade guard for /ws/chat and /ws/call: exact origin allow-list, then IP / network bans. */
async function handleUpgrade(req, socket, head) {
const url = req.url || '';
const origin = normalizeOrigin(req.headers.origin);
@@@ -230,7 +237,8 @@ function createApp({ chatServer, bridge, mirror, relay, events = null }) {
socket.destroy();
return;
}
-- if (!url.startsWith('/ws/chat')) { socket.destroy(); return; }
++ const isCall = url.startsWith('/ws/call') && !!callServer && config.calls.enabled;
++ if (!url.startsWith('/ws/chat') && !isCall) { socket.destroy(); return; }
try {
const wsIp = chatServer.getClientIp(req);
if (ctx.isIpBanned(wsIp, null)) {
@@@ -240,7 +248,8 @@ function createApp({ chatServer, bridge, mirror, relay, events = null }) {
if (!exempt) { socket.destroy(); return; }
}
} catch { /* non-critical — allow through on a policy error */ }
-- chatServer.handleUpgrade(req, socket, head);
++ if (isCall) callServer.handleUpgrade(req, socket, head);
++ else chatServer.handleUpgrade(req, socket, head);
}
return { app, handleUpgrade, allowedOrigins, isAllowedOrigin };
ddiff --git a/server/auth/auth.js b/server/auth/auth.js
iindex 434a59c..1e5f602 100644
---- a/server/auth/auth.js
++++ b/server/auth/auth.js
@@@ -21,6 +21,7 @@ const session = require('./network-session');
const TOKEN_DENIED_PREFIXES = ['/api/admin', '/api/mod', '/api/funds', '/api/payments', '/api/auth/stream-key', '/api/auth/tokens', '/api/cosmetics', '/api/analytics'];
const TOKEN_WRITE_SCOPES = [
['/api/chat', ['chat']], ['/api/dm', ['chat']], ['/api/emotes', ['chat']], ['/api/sounds', ['chat']], ['/api/tts', ['chat']],
++ ['/api/streams', ['stream']], // the call routes (server/calls/routes.js), as on Live
];
function apiTokenAllows(req, scopes) {
const path = String(req.originalUrl || req.url || '').split('?')[0];
ddiff --git a/server/auth/permissions.js b/server/auth/permissions.js
iindex 6162e2e..bc18b92 100644
---- a/server/auth/permissions.js
++++ b/server/auth/permissions.js
@@@ -118,6 +118,14 @@ function canModerateStream(user, streamId) {
return false;
}
++/**
++ * Can this user moderate a call on a specific stream?
++ * Same rules as chat moderation.
++ */
++function canModerateCall(user, streamId) {
++ return canModerateStream(user, streamId);
++}
++
/**
* Can this user view chat logs?
*/
@@@ -163,6 +171,7 @@ module.exports = {
getChannelIdForStream,
canModerateChannel,
canModerateStream,
++ canModerateCall,
canViewChatLogs,
canViewOtherUserLogs,
};
ddiff --git a/server/calls/call-server.js b/server/calls/call-server.js
nnew file mode 100644
iindex 0000000..3ff299a
---- /dev/null
++++ b/server/calls/call-server.js
@@@ -0,0 +1,629 @@
++/**
++ * OpenVibe.Chat — Voice Channel WebSocket Server (/ws/call)
++ *
++ * Moved from OpenVibe.Live server/streaming/call-server.js (roadmap WS-I task 1) with the same
++ * protocol: query parameters, every message type and payload, the limits (8 participants, 3 sockets
++ * per address, the kick cooldown, per-channel bans), the permanent 'public' lobby, stream-linked and
++ * temporary channels and the inactivity cleanup are Live's. What changed is only where Live-owned
++ * data comes from — accounts, streams and cosmetics are read through ../live-context.js, the chat
++ * pushes (the voice-channel list, call invites) go through Chat's own chat server — that sign-in is
++ * Chat's cached token resolution (so a socket's setup is async; messages that arrive meanwhile wait,
++ * in order), and that every call is a row of `calls` (./lifecycle.js).
++ *
++ * Discord-style voice/video channels accessible from the Chat tab.
++ * Uses full-mesh WebRTC peer connections (each participant ↔ every other).
++ * Signaling only — no media passes through the server.
++ *
++ * A permanent "Public" lobby channel always exists. Users can create
++ * temporary channels. Stream-linked channels are auto-created when a
++ * streamer enables voice on their stream (Live asks: POST /internal/calls/stream-channel).
++ *
++ * WebSocket path: /ws/call?channelId=<id>&token=X
++ *
++ * Channel modes:
++ * 'mic' — Microphone only
++ * 'mic+cam' — Mic + optional camera
++ * 'cam+mic' — Both mic and camera
++ */
++'use strict';
++
++const WebSocket = require('ws');
++const config = require('../config');
++const ctx = require('../live-context');
++const { extractWsToken, authenticateWs } = require('../auth/auth');
++const permissions = require('../auth/permissions');
++const chatServer = require('../chat/chat-server');
++const lifecycle = require('./lifecycle');
++// Cosmetics are Live's (monetization/cosmetics); read through live-context, warmed when a socket joins.
++const cosmetics = { getCosmeticProfile: (userId) => ctx.getCosmeticProfile(userId) };
++
++const WS_HEARTBEAT_MS = 30000;
++const MAX_PARTICIPANTS = 8;
++const MAX_SOCKETS_PER_IP = 3; // one household in one channel; never one host filling it
++const KICK_COOLDOWN_MS = 60 * 1000;
++const PUBLIC_CHANNEL_ID = 'public';
++const MODES = ['mic', 'mic+cam', 'cam+mic'];
++
++class CallServer {
++ constructor() {
++ this.wss = null;
++ /** channelId → Map<peerId, clientInfo> */
++ this.rooms = new Map();
++ /** ws → { channelId, peerId } */
++ this.clients = new Map();
++ /** channelId → Set<userId> */
++ this.callBans = new Map();
++ /** channelId → channel metadata { id, name, mode, createdBy, streamId?, permanent, createdAt, maxParticipants } */
++ this.channels = new Map();
++ this._nextPeerId = 1;
++ this._heartbeatInterval = null;
++ this._inactivityTimer = null;
++ /** `${channelId}:${identity}` → until (ms): a kicked participant waits before rejoining */
++ this.kickCooldown = new Map();
++ this._notifyTimer = null;
++
++ // Seed the permanent Public channel
++ this.channels.set(PUBLIC_CHANNEL_ID, {
++ id: PUBLIC_CHANNEL_ID, name: 'Public', mode: 'mic+cam', createdBy: null,
++ streamId: null, permanent: true, createdAt: Date.now(), maxParticipants: MAX_PARTICIPANTS,
++ });
++ }
++
++ init(server) {
++ this.wss = new WebSocket.Server({ noServer: true, maxPayload: 256 * 1024, perMessageDeflate: false });
++ if (this._heartbeatInterval) clearInterval(this._heartbeatInterval);
++ this._heartbeatInterval = setInterval(() => {
++ if (!this.wss) return;
++ this.wss.clients.forEach((ws) => {
++ if (ws.isAlive === false) { try { ws.terminate(); } catch {} return; }
++ ws.isAlive = false;
++ try { ws.ping(); } catch {}
++ });
++ }, WS_HEARTBEAT_MS);
++ if (this._heartbeatInterval.unref) this._heartbeatInterval.unref();
++ this.wss.on('connection', (ws, req) => this._handleConnection(ws, req));
++
++ // Cleanup user-created channels that have been empty for over an hour
++ if (this._inactivityTimer) clearInterval(this._inactivityTimer);
++ this._inactivityTimer = setInterval(() => this._cleanupInactiveChannels(), 5 * 60 * 1000);
++ if (this._inactivityTimer.unref) this._inactivityTimer.unref();
++
++ // Rooms are memory: calls the previous process left open are over.
++ try { lifecycle.recover(); } catch (err) { console.warn('[Calls] recover failed:', err.message); }
++
++ console.log(`[CallServer] Voice channels initialized (${config.calls.enabled ? 'serving' : 'off: CHAT_CALLS is not set'})`);
++ return this.wss;
++ }
++
++ _cleanupInactiveChannels() {
++ const ONE_HOUR = 60 * 60 * 1000;
++ const now = Date.now();
++ for (const [k, until] of this.kickCooldown) if (until <= now) this.kickCooldown.delete(k);
++ for (const [channelId, ch] of this.channels) {
++ if (ch.permanent || ch.streamId) continue;
++ const room = this.rooms.get(channelId);
++ const isEmpty = !room || room.size === 0;
++ if (isEmpty && ch.lastEmptiedAt && (now - ch.lastEmptiedAt) >= ONE_HOUR) {
++ console.log(`[CallServer] Auto-deleting inactive channel "${ch.name}" (${channelId}) — empty for ${Math.round((now - ch.lastEmptiedAt) / 60000)}m`);
++ this.channels.delete(channelId);
++ this.callBans.delete(channelId);
++ lifecycle.channelClosed(channelId, 'inactive');
++ }
++ }
++ }
++
++ handleUpgrade(req, socket, head) {
++ if (!req.url.startsWith('/ws/call')) return false;
++ this.wss.handleUpgrade(req, socket, head, (ws) => { this.wss.emit('connection', ws, req); });
++ return true;
++ }
++
++ /* ── Channel management ──────────────────────────────────── */
++
++ /** Channels a viewer may see: a private call only for its creator, its invitees and staff. */
++ listChannels(viewer = null) {
++ const result = [];
++ for (const [id, ch] of this.channels) {
++ if (ch.private && !this._canSeePrivate(ch, viewer)) continue;
++ const room = this.rooms.get(id);
++ const participants = [];
++ if (room) { for (const [pid, info] of room) participants.push(this._buildParticipantInfo(pid, info)); }
++ result.push({ ...this._publicChannel(ch), participantCount: room ? room.size : 0, participants });
++ }
++ return result;
++ }
++ _publicChannel(ch) { const { invited, ...rest } = ch; return rest; }
++ _canSeePrivate(ch, viewer) {
++ if (!ch.private) return true;
++ if (!viewer) return false;
++ if (permissions.can(viewer, 'staff.moderation.calls')) return true;
++ return ch.createdBy === viewer.id || !!(ch.invited && ch.invited.has(viewer.id));
++ }
++ /** A caller invited this user to their call (POST /voice-channels/call-user). */
++ invite(channelId, userId) {
++ const ch = this.channels.get(channelId);
++ if (!ch) return false;
++ if (!ch.invited) ch.invited = new Set();
++ ch.invited.add(userId);
++ return true;
++ }
++ hasInvite(channelId, userId) {
++ const ch = this.channels.get(channelId);
++ return !!(ch && ch.invited && ch.invited.has(userId));
++ }
++ /** Debounced "the list changed" for every chat socket (the sidebar used to poll every 4 s). */
++ _notifyChannelsChanged() {
++ if (this._notifyTimer) return;
++ this._notifyTimer = setTimeout(() => {
++ this._notifyTimer = null;
++ try { chatServer.broadcastAll({ type: 'voice-channels', channels: this.listChannels(null) }); } catch { /* */ }
++ }, 400);
++ if (this._notifyTimer.unref) this._notifyTimer.unref();
++ }
++
++ getChannel(channelId, viewer = null) {
++ const ch = this.channels.get(channelId);
++ if (!ch || !this._canSeePrivate(ch, viewer)) return null;
++ const room = this.rooms.get(channelId);
++ const participants = [];
++ if (room) { for (const [pid, info] of room) participants.push(this._buildParticipantInfo(pid, info)); }
++ return { ...this._publicChannel(ch), participantCount: room ? room.size : 0, participants };
++ }
++
++ createChannel({ name, mode, createdBy, maxParticipants, isPrivate = false }) {
++ // One user-created channel per user
++ for (const [, ch] of this.channels) {
++ if (!ch.permanent && !ch.streamId && ch.createdBy === createdBy) {
++ throw Object.assign(new Error('You already have a voice channel. Delete it first.'), { code: 'CHANNEL_LIMIT' });
++ }
++ }
++ const id = `user-${createdBy}-${Date.now().toString(36)}`;
++ const ch = {
++ id, name: String(name || 'Voice Channel').slice(0, 40),
++ mode: MODES.includes(mode) ? mode : 'mic+cam',
++ createdBy, streamId: null, permanent: false, createdAt: Date.now(),
++ maxParticipants: Math.min(Math.max(Number(maxParticipants) || MAX_PARTICIPANTS, 2), MAX_PARTICIPANTS),
++ lastEmptiedAt: Date.now(), // starts empty; inactivity timer tracks from creation
++ private: !!isPrivate,
++ };
++ this.channels.set(id, ch);
++ this._notifyChannelsChanged();
++ return this._publicChannel(ch);
++ }
++
++ /** Live's go-live / call-mode hook. `stream` is the stream row when the caller has it (its title). */
++ createStreamChannel(streamId, mode, streamerId, stream = null) {
++ const id = `stream-${streamId}`;
++ const existing = this.channels.get(id);
++ if (existing) { const old = existing.mode; existing.mode = mode; if (old !== mode) this.endCall(id); return existing; }
++ if (!stream) stream = ctx.getStreamById(streamId);
++ const ch = {
++ id, name: stream ? (stream.title || `Stream ${streamId}`) : `Stream ${streamId}`,
++ mode: MODES.includes(mode) ? mode : 'mic',
++ createdBy: streamerId, streamId, permanent: false, createdAt: Date.now(), maxParticipants: MAX_PARTICIPANTS,
++ };
++ this.channels.set(id, ch);
++ return ch;
++ }
++
++ removeStreamChannel(streamId) {
++ const id = `stream-${streamId}`;
++ const had = this.channels.has(id);
++ this.endCall(id, 'stream_ended');
++ this.channels.delete(id);
++ this.callBans.delete(id);
++ lifecycle.channelClosed(id, 'stream_ended');
++ return had;
++ }
++
++ /** `user` is the requester (Live passed an id and looked the account up; Chat's routes have it). */
++ deleteChannel(channelId, user) {
++ const ch = this.channels.get(channelId);
++ if (!ch || ch.permanent || !user) return false;
++ if (ch.createdBy !== user.id) {
++ if (!permissions.can(user, 'staff.moderation.calls')) return false;
++ }
++ this.endCall(channelId, 'closed');
++ this.channels.delete(channelId);
++ this.callBans.delete(channelId);
++ lifecycle.channelClosed(channelId, 'closed');
++ this._notifyChannelsChanged();
++ return true;
++ }
++
++ /* ── WebRTC signaling ──────────────────────────────────────── */
++
++ _generatePeerId() { return `call-peer-${this._nextPeerId++}-${Date.now().toString(36)}`; }
++
++ _buildParticipantInfo(peerId, info) {
++ let cosmeticProfile = {};
++ // From live-context's decor cache (warmed at join, refreshed in the background).
++ if (info.user?.id) { try { cosmeticProfile = cosmetics.getCosmeticProfile(info.user.id) || {}; } catch {} }
++ return {
++ peerId, username: info.user ? info.user.username : null,
++ anonId: info.anonId || null,
++ displayName: info.user ? (info.user.display_name || info.user.username) : info.anonId,
++ userId: info.user ? info.user.id : null,
++ avatarUrl: info.user ? info.user.avatar_url : null,
++ profileColor: info.user ? info.user.profile_color : null,
++ isChannelCreator: info.isChannelCreator || false,
++ isStreamer: info.isStreamer || false,
++ muted: info.muted, cameraOff: info.cameraOff,
++ forceMuted: info.forceMuted || false, forceCameraOff: info.forceCameraOff || false,
++ speaking: info.speaking || false,
++ nameFX: cosmeticProfile.nameFX || null, particleFX: cosmeticProfile.particleFX || null, hatFX: cosmeticProfile.hatFX || null,
++ };
++ }
++
++ _handleConnection(ws, req) {
++ const url = new URL(req.url, 'http://localhost');
++ const channelId = url.searchParams.get('channelId') || url.searchParams.get('streamId');
++ const token = extractWsToken(req);
++ // CF-Connecting-IP / X-Forwarded-For only when nginx's peer is Cloudflare (net/client-ip.js), as on /ws/chat.
++ const ip = chatServer.getClientIp(req);
++
++ ws.isAlive = true;
++ ws.on('pong', () => { ws.isAlive = true; });
++ try { ws._socket?.setNoDelay(true); } catch {}
++
++ if (!channelId) { ws.send(JSON.stringify({ type: 'error', message: 'Missing channelId' })); ws.close(); return; }
++
++ // Legacy: bare number → stream-<N>
++ const resolvedId = /^\d+$/.test(channelId) ? `stream-${channelId}` : channelId;
++
++ if (!this.channels.get(resolvedId)) { ws.send(JSON.stringify({ type: 'error', message: 'Voice channel not found' })); ws.close(); return; }
++
++ // Sign-in, the anon number and the stream are (cached) calls to Live now. Messages that arrive
++ // meanwhile wait, in order, and are handled once the socket is in the room.
++ const early = [];
++ let closedEarly = false;
++ const onEarlyMessage = (data) => early.push(data);
++ const onEarlyClose = () => { closedEarly = true; };
++ ws.on('message', onEarlyMessage);
++ ws.on('close', onEarlyClose);
++ ws.on('error', onEarlyClose);
++ (async () => {
++ const first = this.channels.get(resolvedId);
++ let stream = null;
++ if (first && first.streamId) stream = ctx.getStreamById(first.streamId) || await ctx.ensureStream(first.streamId).catch(() => null);
++ const user = await authenticateWs(token).catch(() => null);
++ if (!user) await chatServer._resolveUnifiedAnonNum(chatServer.normalizeIp(ip)).catch(() => {});
++ await ctx.warm({ user, streamId: first && first.streamId ? first.streamId : null, ip }).catch(() => {});
++ ws.off('message', onEarlyMessage);
++ ws.off('close', onEarlyClose);
++ ws.off('error', onEarlyClose);
++ if (closedEarly || ws.readyState !== WebSocket.OPEN) return;
++ this._admit(ws, { resolvedId, ip, user, stream, early });
++ })().catch((err) => {
++ console.warn('[Call] connection setup failed:', err.message);
++ try { ws.close(1011, 'setup failed'); } catch {}
++ });
++ }
++
++ /** The rest of Live's _handleConnection, once the socket's identity is known. */
++ _admit(ws, { resolvedId, ip, user, stream, early }) {
++ // The channel may have gone while the socket signed in.
++ const channel = this.channels.get(resolvedId);
++ if (!channel) { ws.send(JSON.stringify({ type: 'error', message: 'Voice channel not found' })); ws.close(); return; }
++
++ // If stream-linked, verify stream is live
++ if (channel.streamId) {
++ if (!stream || Number(stream.id) !== Number(channel.streamId)) stream = ctx.getStreamById(channel.streamId);
++ if (!stream || !stream.is_live) { ws.send(JSON.stringify({ type: 'error', message: 'Stream not live' })); ws.close(); return; }
++ }
++
++ if (channel.private && !this._canSeePrivate(channel, user)) { ws.send(JSON.stringify({ type: 'error', message: 'This is a private call' })); ws.close(); return; }
++
++ const peerId = this._generatePeerId();
++ const anonId = user ? null : chatServer.getAnonIdForConnection(ip, resolvedId);
++ const identity = user ? `u:${user.id}` : (anonId ? `a:${anonId}` : `ip:${ip}`);
++ const cooled = this.kickCooldown.get(`${resolvedId}:${identity}`);
++ if (cooled && cooled > Date.now()) { ws.send(JSON.stringify({ type: 'error', message: 'You were removed from this channel; try again in a minute' })); ws.close(); return; }
++ const isChannelCreator = user && channel.createdBy === user.id;
++ const isStreamer = channel.streamId ? (user && ctx.getStreamById(channel.streamId)?.user_id === user.id) : false;
++ const canModerate = this._canModerate(user, resolvedId);
++
++ const bans = this.callBans.get(resolvedId);
++ if (bans) {
++ const banned = user
++ ? bans.has(user.id) || bans.has(`u:${user.id}`)
++ : bans.has(`a:${anonId}`) || bans.has(`ip:${ip}`);
++ if (banned) {
++ ws.send(JSON.stringify({ type: 'error', message: 'You are banned from this voice channel' })); ws.close(); return;
++ }
++ }
++
++ if (!this.rooms.has(resolvedId)) this.rooms.set(resolvedId, new Map());
++ const room = this.rooms.get(resolvedId);
++ // The same account again (a second tab, or a reconnect before the old socket timed out)
++ // replaces its older socket instead of standing next to it as a ghost for a minute.
++ if (user) {
++ for (const [pid, info] of room) {
++ if (info.user && info.user.id === user.id) {
++ try { info.ws.send(JSON.stringify({ type: 'replaced' })); } catch { /* */ }
++ this._handleDisconnect(info.ws, resolvedId, pid);
++ try { info.ws.close(); } catch { /* */ }
++ }
++ }
++ }
++ // Replacing the only socket emptied the room and dropped it from this.rooms: put it back, or
++ // this socket would sit in a room nobody else can reach.
++ if (!this.rooms.has(resolvedId)) this.rooms.set(resolvedId, room);
++ let fromIp = 0;
++ for (const [, info] of room) if (info.ip === ip) fromIp++;
++ if (fromIp >= MAX_SOCKETS_PER_IP && !permissions.can(user, 'staff.limits.exempt')) { ws.send(JSON.stringify({ type: 'error', message: 'Too many connections from your network' })); ws.close(); return; }
++ const maxP = channel.maxParticipants || MAX_PARTICIPANTS;
++ if (room.size >= maxP) { ws.send(JSON.stringify({ type: 'error', message: `Channel full (max ${maxP})` })); ws.close(); return; }
++
++ const clientInfo = { ws, user, anonId, ip, peerId, identity, muted: false, cameraOff: true, forceMuted: false, forceCameraOff: false, speaking: false, isChannelCreator, isStreamer, _msgCount: 0, _msgResetTime: Date.now() };
++ this.clients.set(ws, { channelId: resolvedId, peerId });
++ room.set(peerId, clientInfo);
++
++ // Mark channel as active (reset inactivity timer)
++ if (channel && !channel.permanent && !channel.streamId) {
++ channel.lastEmptiedAt = null;
++ }
++ // The lifecycle: the channel's session is active; an invitee joining answers the ring.
++ try {
++ if (room.size === 1) lifecycle.sessionStarted({ channelId: resolvedId, streamId: channel.streamId || null, createdBy: channel.createdBy || null });
++ if (user) lifecycle.answeredByJoin(resolvedId, user.id);
++ } catch (err) { console.warn('[Calls] lifecycle:', err.message); }
++
++ const participants = [];
++ for (const [pid, info] of room) participants.push(this._buildParticipantInfo(pid, info));
++
++ ws.send(JSON.stringify({
++ type: 'welcome',
++ peerId,
++ channelId: resolvedId,
++ channelName: channel.name,
++ callMode: channel.mode,
++ participants,
++ isStreamer: isStreamer || isChannelCreator,
++ canModerate,
++ }));
++
++ const joinMsg = JSON.stringify({ type: 'peer-joined', ...this._buildParticipantInfo(peerId, clientInfo) });
++ for (const [pid, info] of room) { if (pid !== peerId && info.ws.readyState === WebSocket.OPEN) info.ws.send(joinMsg); }
++ this._broadcastParticipantCount(resolvedId);
++ this._notifyChannelsChanged();
++
++ // One message at a time per socket, in arrival order (auth-update resolves a token first).
++ let queue = Promise.resolve();
++ const onMessage = (data) => {
++ // Rate limit: 200/s. A newcomer to a full room sends seven offers and their candidates in
++ // a burst; the old 50/s silently dropped some of them and the join stalled.
++ const now = Date.now();
++ if (now - clientInfo._msgResetTime > 1000) { clientInfo._msgCount = 0; clientInfo._msgResetTime = now; }
++ if (++clientInfo._msgCount > 200) return;
++ queue = queue.then(() => this._handleMessage(ws, JSON.parse(data), resolvedId, peerId))
++ .catch((err) => { console.warn('[Call] Message error for peer', peerId, ':', err.message); });
++ };
++ ws.on('message', onMessage);
++ ws.on('close', () => this._handleDisconnect(ws, resolvedId, peerId));
++ ws.on('error', () => this._handleDisconnect(ws, resolvedId, peerId));
++ for (const data of early) onMessage(data);
++ }
++
++ _canModerate(user, channelId) {
++ if (!user) return false;
++ if (permissions.can(user, 'staff.moderation.calls')) return true;
++ const ch = this.channels.get(channelId);
++ if (ch?.createdBy === user.id) return true;
++ if (ch?.streamId) return permissions.canModerateCall(user, ch.streamId);
++ return false;
++ }
++
++ async _handleMessage(ws, msg, channelId, peerId) {
++ const room = this.rooms.get(channelId);
++ if (!room) return;
++
++ switch (msg.type) {
++ case 'offer': case 'answer': case 'ice-candidate': {
++ // Validate SDP/candidate size to prevent abuse
++ if (msg.sdp && (typeof msg.sdp.sdp !== 'string' || msg.sdp.sdp.length > 16384)) break;
++ if (msg.candidate && (typeof msg.candidate.candidate !== 'string' || msg.candidate.candidate.length > 2048)) break;
++ const tp = room.get(msg.targetPeerId);
++ if (tp && tp.ws.readyState === WebSocket.OPEN) tp.ws.send(JSON.stringify({ type: msg.type, fromPeerId: peerId, sdp: msg.sdp, candidate: msg.candidate }));
++ break;
++ }
++ case 'mute': {
++ const c = room.get(peerId); if (!c) break;
++ c.muted = !!msg.muted;
++ const m = JSON.stringify({ type: 'peer-muted', peerId, muted: c.muted });
++ for (const [pid, info] of room) { if (pid !== peerId && info.ws.readyState === WebSocket.OPEN) info.ws.send(m); }
++ this._notifyChannelsChanged();
++ break;
++ }
++ case 'camera-off': {
++ const c = room.get(peerId); if (!c) break;
++ c.cameraOff = !!msg.cameraOff;
++ const m = JSON.stringify({ type: 'peer-camera', peerId, cameraOff: c.cameraOff });
++ for (const [pid, info] of room) { if (pid !== peerId && info.ws.readyState === WebSocket.OPEN) info.ws.send(m); }
++ break;
++ }
++ case 'speaking': {
++ const c = room.get(peerId); if (!c) break;
++ const s = !!msg.speaking; if (c.speaking === s) break; c.speaking = s;
++ const m = JSON.stringify({ type: 'peer-speaking', peerId, speaking: s });
++ for (const [pid, info] of room) { if (pid !== peerId && info.ws.readyState === WebSocket.OPEN) info.ws.send(m); }
++ break;
++ }
++ case 'auth-update': {
++ const c = room.get(peerId); if (!c) break;
++ let user = c.user || null;
++ if (typeof msg.token === 'string' && msg.token.trim()) {
++ const nextUser = await authenticateWs(msg.token).catch(() => null);
++ if (!nextUser) {
++ console.warn('[Call] auth-update rejected for peer', peerId, '(invalid or expired token)');
++ } else if (user && user.id !== nextUser.id) {
++ console.warn(`[Call] auth-update identity mismatch for peer ${peerId}: keeping existing user ${user.username}, ignoring ${nextUser.username}`);
++ } else {
++ user = nextUser;
++ }
++ if (user) await ctx.ensureDecor([user.id]).catch(() => {});
++ // The socket may have left while the token was resolved.
++ if (this.rooms.get(channelId) !== room || room.get(peerId) !== c) break;
++ }
++ if (user && this.callBans.has(channelId) && this.callBans.get(channelId).has(user.id)) {
++ if (c.ws.readyState === WebSocket.OPEN) { c.ws.send(JSON.stringify({ type: 'error', message: 'Banned' })); c.ws.close(); } break;
++ }
++ const ch = this.channels.get(channelId);
++ c.user = user; c.anonId = user ? null : chatServer.getAnonIdForConnection(c.ip, channelId);
++ c.isChannelCreator = !!(user && ch?.createdBy === user.id);
++ c.isStreamer = ch?.streamId ? !!(user && ctx.getStreamById(ch.streamId)?.user_id === user.id) : false;
++ if (user) { try { lifecycle.answeredByJoin(channelId, user.id); } catch { /* */ } }
++ const pInfo = this._buildParticipantInfo(peerId, c);
++ if (c.ws.readyState === WebSocket.OPEN) c.ws.send(JSON.stringify({
++ type: 'self-updated',
++ isStreamer: c.isStreamer || c.isChannelCreator,
++ canModerate: this._canModerate(user, channelId),
++ participant: pInfo,
++ }));
++ const u = JSON.stringify({ type: 'peer-updated', ...pInfo });
++ for (const [pid, info] of room) { if (pid !== peerId && info.ws.readyState === WebSocket.OPEN) info.ws.send(u); }
++ break;
++ }
++ case 'force-mute': {
++ const sender = room.get(peerId); if (!sender || !this._canModerate(sender.user, channelId)) break;
++ const target = room.get(msg.targetPeerId); if (!target || target.isChannelCreator || target.isStreamer) break;
++ target.forceMuted = !!msg.forceMuted;
++ if (target.ws.readyState === WebSocket.OPEN) target.ws.send(JSON.stringify({ type: 'force-muted', forceMuted: target.forceMuted }));
++ const m = JSON.stringify({ type: 'peer-force-muted', peerId: msg.targetPeerId, forceMuted: target.forceMuted });
++ for (const [pid, info] of room) { if (info.ws.readyState === WebSocket.OPEN) info.ws.send(m); }
++ break;
++ }
++ case 'force-camera-off': {
++ const sender = room.get(peerId); if (!sender || !this._canModerate(sender.user, channelId)) break;
++ const target = room.get(msg.targetPeerId); if (!target || target.isChannelCreator || target.isStreamer) break;
++ target.forceCameraOff = !!msg.forceCameraOff;
++ if (target.ws.readyState === WebSocket.OPEN) target.ws.send(JSON.stringify({ type: 'force-camera-off', forceCameraOff: target.forceCameraOff }));
++ const m = JSON.stringify({ type: 'peer-force-camera-off', peerId: msg.targetPeerId, forceCameraOff: target.forceCameraOff });
++ for (const [pid, info] of room) { if (info.ws.readyState === WebSocket.OPEN) info.ws.send(m); }
++ break;
++ }
++ case 'kick': {
++ const sender = room.get(peerId); if (!sender || !this._canModerate(sender.user, channelId)) break;
++ const target = room.get(msg.targetPeerId); if (!target || target.isChannelCreator || target.isStreamer) break;
++ if (target.identity) this.kickCooldown.set(`${channelId}:${target.identity}`, Date.now() + KICK_COOLDOWN_MS);
++ if (target.ws.readyState === WebSocket.OPEN) { target.ws.send(JSON.stringify({ type: 'kicked' })); target.ws.close(); }
++ room.delete(msg.targetPeerId); this.clients.delete(target.ws);
++ const m = JSON.stringify({ type: 'peer-left', peerId: msg.targetPeerId, reason: 'kicked' });
++ for (const [pid, info] of room) { if (info.ws.readyState === WebSocket.OPEN) info.ws.send(m); }
++ this._broadcastParticipantCount(channelId);
++ break;
++ }
++ case 'ban': {
++ const sender = room.get(peerId); if (!sender || !this._canModerate(sender.user, channelId)) break;
++ const target = room.get(msg.targetPeerId); if (!target || target.isChannelCreator || target.isStreamer) break;
++ if (!this.callBans.has(channelId)) this.callBans.set(channelId, new Set());
++ const banSet = this.callBans.get(channelId);
++ if (target.user?.id) {
++ banSet.add(target.user.id); // backward compatibility with existing entries
++ banSet.add(`u:${target.user.id}`);
++ } else {
++ if (target.anonId) banSet.add(`a:${target.anonId}`);
++ if (target.ip) banSet.add(`ip:${target.ip}`);
++ }
++ if (target.ws.readyState === WebSocket.OPEN) { target.ws.send(JSON.stringify({ type: 'banned' })); target.ws.close(); }
++ room.delete(msg.targetPeerId); this.clients.delete(target.ws);
++ const m = JSON.stringify({ type: 'peer-left', peerId: msg.targetPeerId, reason: 'banned' });
++ for (const [pid, info] of room) { if (info.ws.readyState === WebSocket.OPEN) info.ws.send(m); }
++ this._broadcastParticipantCount(channelId);
++ break;
++ }
++ case 'unban': {
++ const sender = room.get(peerId); if (!sender || !this._canModerate(sender.user, channelId)) break;
++ const uid = parseInt(msg.userId);
++ if (uid && this.callBans.has(channelId)) {
++ this.callBans.get(channelId).delete(uid);
++ this.callBans.get(channelId).delete(`u:${uid}`);
++ }
++ break;
++ }
++ case 'end-call': case 'end-channel': {
++ const c = room.get(peerId); if (!c) break;
++ const ch = this.channels.get(channelId);
++ const canEnd = (ch?.createdBy && c.user?.id === ch.createdBy) || c.isStreamer || permissions.can(c.user, 'staff.moderation.calls');
++ if (canEnd) {
++ this.endCall(channelId, 'ended');
++ if (ch && !ch.permanent) { this.channels.delete(channelId); lifecycle.channelClosed(channelId, 'ended'); }
++ }
++ break;
++ }
++ }
++ }
++
++ _handleDisconnect(ws, channelId, peerId) {
++ if (!this.clients.has(ws)) return; // already handled (error+close fire back-to-back)
++ this.clients.delete(ws);
++ const room = this.rooms.get(channelId);
++ if (!room) return;
++ const leaving = room.get(peerId);
++ const leftInfo = leaving ? this._buildParticipantInfo(peerId, leaving) : { peerId };
++ room.delete(peerId);
++ const m = JSON.stringify({ type: 'peer-left', ...leftInfo, reason: 'disconnect' });
++ for (const [pid, info] of room) { if (info.ws.readyState === WebSocket.OPEN) info.ws.send(m); }
++ this._broadcastParticipantCount(channelId);
++ this._notifyChannelsChanged();
++ if (room.size === 0) {
++ this.rooms.delete(channelId);
++ const ch = this.channels.get(channelId);
++ // For user-created channels: don't delete immediately; track when they
++ // became empty so the inactivity timer can clean up after 1 hour.
++ if (ch && !ch.permanent && !ch.streamId) {
++ ch.lastEmptiedAt = Date.now();
++ }
++ try { lifecycle.channelEmptied(channelId, 'empty'); } catch (err) { console.warn('[Calls] lifecycle:', err.message); }
++ }
++ }
++
++ _broadcastParticipantCount(channelId) {
++ const room = this.rooms.get(channelId);
++ const count = room ? room.size : 0;
++ const m = JSON.stringify({ type: 'participant-count', count, channelId });
++ if (room) { for (const [pid, info] of room) { if (info.ws.readyState === WebSocket.OPEN) info.ws.send(m); } }
++ }
++
++ /* ── Public helpers ────────────────────────────────────────── */
++
++ getParticipantCount(channelId) { const r = this.rooms.get(channelId); return r ? r.size : 0; }
++
++ getParticipants(channelId) {
++ const r = this.rooms.get(channelId); if (!r) return [];
++ const l = []; for (const [pid, info] of r) l.push(this._buildParticipantInfo(pid, info)); return l;
++ }
++
++ getCallBans(channelId) { const b = this.callBans.get(channelId); return b ? [...b] : []; }
++
++ endCall(channelId, reason = 'ended') {
++ const r = this.rooms.get(channelId); if (!r) return;
++ const entries = [...r.values()];
++ this.rooms.delete(channelId);
++ const m = JSON.stringify({ type: 'call-ended' });
++ for (const info of entries) {
++ this.clients.delete(info.ws);
++ if (info.ws.readyState === WebSocket.OPEN) { info.ws.send(m); info.ws.close(); }
++ }
++ try { lifecycle.channelEmptied(channelId, reason); } catch (err) { console.warn('[Calls] lifecycle:', err.message); }
++ }
++
++ /** Legacy compat: end call by stream ID */
++ endStreamCall(streamId) { this.endCall(`stream-${streamId}`); }
++
++ close() {
++ for (const [cid] of this.rooms) this.endCall(cid, 'restart');
++ if (this._heartbeatInterval) { clearInterval(this._heartbeatInterval); this._heartbeatInterval = null; }
++ if (this._inactivityTimer) { clearInterval(this._inactivityTimer); this._inactivityTimer = null; }
++ if (this._notifyTimer) { clearTimeout(this._notifyTimer); this._notifyTimer = null; }
++ lifecycle.stop();
++ if (this.wss) this.wss.close();
++ }
++}
++
++module.exports = new CallServer();
++module.exports.CallServer = CallServer;
++module.exports.MAX_PARTICIPANTS = MAX_PARTICIPANTS;
++module.exports.MAX_SOCKETS_PER_IP = MAX_SOCKETS_PER_IP;
++module.exports.KICK_COOLDOWN_MS = KICK_COOLDOWN_MS;
++module.exports.PUBLIC_CHANNEL_ID = PUBLIC_CHANNEL_ID;
ddiff --git a/server/calls/internal.js b/server/calls/internal.js
nnew file mode 100644
iindex 0000000..7efa7f1
---- /dev/null
++++ b/server/calls/internal.js
@@@ -0,0 +1,70 @@
++/**
++ * 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):
++ *
++ * 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
++ * (a new mode ends the call in progress, as in Live). The stream is read from Live first, so
++ * its title and live state are current when viewers join.
++ * DELETE /internal/calls/stream-channel/:streamId capability chat.live_bridge.write
++ * → { ok, removed } the stream ended (or was force-ended): its call ends, the channel goes.
++ *
++ * While CHAT_CALLS is off both answer 409 and change nothing, so a Live switched too early learns
++ * why (Live logs it) instead of creating channels nobody can reach.
++ */
++'use strict';
++
++const express = require('express');
++const config = require('../config');
++const ctx = require('../live-context');
++const serviceAuth = require('../net/service-auth');
++const callServer = require('./call-server');
++const { MODES } = require('./routes');
++
++const CAPABILITY = 'chat.live_bridge.write';
++const STREAM_LOOKUP_MS = 2500;
++
++function createInternalRouter() {
++ const router = express.Router();
++ router.use(serviceAuth.guard(CAPABILITY));
++ router.use((req, res, next) => {
++ if (!config.calls.enabled) return res.status(409).json({ error: 'Calls are not served by Chat (CHAT_CALLS is not set)', code: 'calls.off' });
++ next();
++ });
++
++ router.post('/stream-channel', async (req, res) => {
++ const b = req.body || {};
++ const streamId = Number(b.stream_id);
++ const userId = Number(b.user_id);
++ const mode = b.mode;
++ if (!Number.isInteger(streamId) || streamId <= 0) return res.status(400).json({ error: 'stream_id required' });
++ if (!Number.isInteger(userId) || userId <= 0) return res.status(400).json({ error: 'user_id required' });
++ if (!MODES.includes(mode)) return res.status(400).json({ error: `mode must be one of ${MODES.join(', ')}` });
++ try {
++ // Bounded: a slow Live must not hold Live's own go-live; the projection answers meanwhile.
++ const stream = await Promise.race([ctx.refreshStream(streamId), new Promise((r) => setTimeout(() => r(null), STREAM_LOOKUP_MS).unref?.())])
++ || ctx.getStreamById(streamId);
++ const ch = callServer.createStreamChannel(streamId, mode, userId, stream);
++ res.json({ ok: true, channel: callServer._publicChannel(ch) });
++ } catch (err) {
++ console.error('[Calls] stream-channel:', err.message);
++ res.status(500).json({ error: 'Failed to create the stream voice channel' });
++ }
++ });
++
++ router.delete('/stream-channel/:streamId', (req, res) => {
++ const streamId = Number(req.params.streamId);
++ if (!Number.isInteger(streamId) || streamId <= 0) return res.status(400).json({ error: 'stream id required' });
++ try {
++ res.json({ ok: true, removed: callServer.removeStreamChannel(streamId) });
++ } catch (err) {
++ console.error('[Calls] stream-channel:', err.message);
++ res.status(500).json({ error: 'Failed to remove the stream voice channel' });
++ }
++ });
++
++ return router;
++}
++
++module.exports = { createInternalRouter, CAPABILITY };
ddiff --git a/server/calls/lifecycle.js b/server/calls/lifecycle.js
nnew file mode 100644
iindex 0000000..c37688f
---- /dev/null
++++ b/server/calls/lifecycle.js
@@@ -0,0 +1,199 @@
++/**
++ * OpenVibe.Chat — the call lifecycle (roadmap WS-I task 1): one row of `calls` per call.
++ *
++ * direct — a ring from one person to another (POST /api/streams/voice-channels/call-user):
++ *
++ * pending ──▶ ringing ──▶ active ──▶ ended (the channel emptied or closed)
++ * │ ├──────▶ missed (no answer within the ring timeout, or the callee's "no-answer")
++ * │ ├──────▶ declined (declined, or busy in another call)
++ * │ └──────▶ ended (the caller's channel closed before an answer: canceled)
++ * └───────────┴──────▶ failed (the invite could not be delivered)
++ *
++ * channel / stream — a session of a voice channel ('stream' for a stream-linked one): created
++ * active when the first person joins, ended when the channel empties or is closed.
++ *
++ * pending the row exists, the invite is not out yet.
++ * ringing the invite went to the callee's chat sockets (and Live was asked for the notification).
++ * active answered: the callee accepted, or joined the channel; a session: somebody is in.
++ * ended, missed, declined, failed final; `end_reason` says why (timeout, no-answer, busy,
++ * declined, canceled, empty, ended, closed, stream_ended, restart, invite_failed: …).
++ *
++ * Signalling state is memory (as it was in Live), so a Chat restart forgets every room: recover()
++ * at boot closes what was left open — pending → failed, ringing → missed, active → ended — with
++ * reason `restart`. Times are ms.
++ */
++'use strict';
++
++const db = require('../db/database');
++
++const STATES = ['pending', 'ringing', 'active', 'ended', 'missed', 'declined', 'failed'];
++const OPEN = ['pending', 'ringing', 'active'];
++const TRANSITIONS = {
++ pending: ['ringing', 'failed', 'ended'],
++ ringing: ['active', 'missed', 'declined', 'failed', 'ended'],
++ active: ['ended', 'failed'],
++ ended: [],
++ missed: [],
++ declined: [],
++ failed: [],
++};
++const KINDS = ['channel', 'direct', 'stream'];
++// A callee's answer (POST …/call-user/respond `status`) → [state, end_reason].
++const RESPONSES = {
++ accepted: ['active', null],
++ declined: ['declined', 'declined'],
++ busy: ['declined', 'busy'],
++ 'no-answer': ['missed', 'no-answer'],
++ canceled: ['ended', 'canceled'],
++};
++
++const _timers = new Map(); // call id → ring timeout
++
++function canTransition(from, to) {
++ return !!(TRANSITIONS[from] && TRANSITIONS[from].includes(to));
++}
++
++function get(id) { return db.get('SELECT * FROM calls WHERE id = ?', [id]) || null; }
++
++function clearTimer(id) {
++ const t = _timers.get(id);
++ if (t) { clearTimeout(t); _timers.delete(id); }
++}
++
++/** Move a call to `to` when its state allows it. Returns the updated row, or null (nothing written). */
++function transition(id, to, { reason = null, now = Date.now() } = {}) {
++ const row = get(id);
++ if (!row || !canTransition(row.state, to)) return null;
++ const sets = ['state = ?'];
++ const params = [to];
++ if (to === 'ringing') { sets.push('started_at = COALESCE(started_at, ?)'); params.push(now); }
++ if (to === 'active') {
++ sets.push('started_at = COALESCE(started_at, ?)'); params.push(now);
++ if (row.kind === 'direct') { sets.push('answered_at = COALESCE(answered_at, ?)'); params.push(now); }
++ }
++ if (!OPEN.includes(to)) { sets.push('ended_at = ?', 'end_reason = ?'); params.push(now, reason); }
++ const r = db.run(`UPDATE calls SET ${sets.join(', ')} WHERE id = ? AND state = ?`, [...params, id, row.state]);
++ if (r.changes !== 1) return null;
++ if (to !== 'ringing') clearTimer(id);
++ return get(id);
++}
++
++function insert({ kind, state, channelId, streamId = null, createdBy = null, targetId = null, now = Date.now() }) {
++ if (!KINDS.includes(kind) || !STATES.includes(state)) throw new Error(`bad call ${kind}/${state}`);
++ const r = db.run(`INSERT INTO calls (kind, state, channel_id, stream_id, created_by, created_by_subject, target_user_id, target_subject, created_at, started_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
++ [kind, state, String(channelId), streamId || null, createdBy || null, db.subjectFor(createdBy), targetId || null, db.subjectFor(targetId), now, state === 'active' ? now : null]);
++ return get(r.lastInsertRowid);
++}
++
++// ── Direct calls ────────────────────────────────────────────────────────────────────────────────
++
++/** The caller's open ring to this person on this channel, if any (calling again re-rings it). */
++function openDirectFor(callerId, targetId, channelId) {
++ return db.get(`SELECT * FROM calls WHERE kind = 'direct' AND created_by = ? AND target_user_id = ? AND channel_id = ?
++ AND state IN ('pending', 'ringing') ORDER BY id DESC LIMIT 1`, [callerId, targetId, String(channelId)]) || null;
++}
++
++/** A ring from callerId to targetId in channelId: the open one again, or a new pending row. */
++function openDirect({ callerId, targetId, channelId }) {
++ return openDirectFor(callerId, targetId, channelId) || insert({ kind: 'direct', state: 'pending', channelId, createdBy: callerId, targetId });
++}
++
++/**
++ * The invite went out: ringing, and missed after `timeoutMs` unless answered first. `onTimeout(row)`
++ * runs once the row is missed (the call server tells the caller). Ringing again restarts the timer.
++ */
++function ring(id, { timeoutMs, onTimeout = null } = {}) {
++ let row = get(id);
++ if (!row) return null;
++ if (row.state === 'pending') row = transition(id, 'ringing');
++ if (!row || row.state !== 'ringing') return row;
++ clearTimer(id);
++ const t = setTimeout(() => {
++ _timers.delete(id);
++ const missed = transition(id, 'missed', { reason: 'timeout' });
++ if (missed && onTimeout) { try { onTimeout(missed); } catch (err) { console.warn('[Calls] ring timeout:', err.message); } }
++ }, Math.max(1, Number(timeoutMs) || 0));
++ if (t.unref) t.unref();
++ _timers.set(id, t);
++ return row;
++}
++
++/** The invite could not be delivered. */
++function fail(id, reason) {
++ return transition(id, 'failed', { reason: String(reason || 'failed').slice(0, 200) });
++}
++
++/**
++ * The callee answered (a respond status). Returns the row after the answer, or null when there is
++ * no ring to answer (never rang, already over). `accepted` on an active call changes nothing.
++ */
++function respond({ callerId, targetId, channelId, status }) {
++ const rule = RESPONSES[status];
++ if (!rule) return null;
++ const row = db.get(`SELECT * FROM calls WHERE kind = 'direct' AND created_by = ? AND target_user_id = ? AND channel_id = ?
++ AND state IN ('pending', 'ringing', 'active') ORDER BY id DESC LIMIT 1`, [callerId, targetId, String(channelId)]);
++ if (!row) return null;
++ if (row.state === 'active') return status === 'accepted' ? row : null;
++ return transition(row.id, rule[0], { reason: rule[1] });
++}
++
++/** The callee joined the caller's channel while it rang: that is an answer too. */
++function answeredByJoin(channelId, userId) {
++ if (!userId) return [];
++ const rows = db.all(`SELECT id FROM calls WHERE kind = 'direct' AND channel_id = ? AND target_user_id = ? AND state = 'ringing'`, [String(channelId), userId]);
++ return rows.map((r) => transition(r.id, 'active')).filter(Boolean);
++}
++
++// ── Channel sessions ────────────────────────────────────────────────────────────────────────────
++
++/** Somebody is in the channel: its session is active (one per channel at a time). */
++function sessionStarted({ channelId, streamId = null, createdBy = null }) {
++ const open = db.get(`SELECT * FROM calls WHERE kind IN ('channel', 'stream') AND channel_id = ? AND state = 'active' ORDER BY id DESC LIMIT 1`, [String(channelId)]);
++ if (open) return open;
++ return insert({ kind: streamId ? 'stream' : 'channel', state: 'active', channelId, streamId, createdBy });
++}
++
++/** Nobody is left (or everyone was sent away): the session and every answered ring on it end. */
++function channelEmptied(channelId, reason = 'empty') {
++ const rows = db.all(`SELECT id FROM calls WHERE channel_id = ? AND state = 'active'`, [String(channelId)]);
++ return rows.map((r) => transition(r.id, 'ended', { reason })).filter(Boolean);
++}
++
++/** The channel is gone (deleted, ended, stream over): rings still waiting on it end as canceled. */
++function channelClosed(channelId, reason = 'closed') {
++ const ended = channelEmptied(channelId, reason);
++ const rows = db.all(`SELECT id FROM calls WHERE kind = 'direct' AND channel_id = ? AND state IN ('pending', 'ringing')`, [String(channelId)]);
++ return ended.concat(rows.map((r) => transition(r.id, 'ended', { reason: 'canceled' })).filter(Boolean));
++}
++
++// ── Boot ────────────────────────────────────────────────────────────────────────────────────────
++
++/** After a restart no room survives: close whatever the previous process left open. */
++function recover() {
++ const out = { failed: 0, missed: 0, ended: 0 };
++ for (const r of db.all(`SELECT id, state FROM calls WHERE state IN ('pending', 'ringing', 'active')`)) {
++ const to = r.state === 'pending' ? 'failed' : r.state === 'ringing' ? 'missed' : 'ended';
++ if (transition(r.id, to, { reason: 'restart' })) out[to]++;
++ }
++ if (out.failed + out.missed + out.ended) console.log(`[Calls] closed calls left open by the last run: ${JSON.stringify(out)}`);
++ return out;
++}
++
++function list({ channelId = null, kind = null, limit = 50 } = {}) {
++ const where = [];
++ const params = [];
++ if (channelId) { where.push('channel_id = ?'); params.push(String(channelId)); }
++ if (kind) { where.push('kind = ?'); params.push(kind); }
++ return db.all(`SELECT * FROM calls ${where.length ? `WHERE ${where.join(' AND ')}` : ''} ORDER BY id DESC LIMIT ?`, [...params, Math.min(Math.max(Number(limit) || 50, 1), 500)]);
++}
++
++function stop() { for (const id of [..._timers.keys()]) clearTimer(id); }
++
++module.exports = {
++ STATES, OPEN, TRANSITIONS, KINDS, RESPONSES,
++ canTransition, transition, get, list,
++ openDirect, ring, fail, respond, answeredByJoin,
++ sessionStarted, channelEmptied, channelClosed,
++ recover, stop,
++};
ddiff --git a/server/calls/routes.js b/server/calls/routes.js
nnew file mode 100644
iindex 0000000..1cf4d4e
---- /dev/null
++++ b/server/calls/routes.js
@@@ -0,0 +1,266 @@
++/**
++ * OpenVibe.Chat — the call REST routes, on Live's paths (mounted at /api/streams; nginx sends
++ * exactly these there at the calls cutover, docs/calls-cutover.md). Moved from OpenVibe.Live
++ * server/streaming/routes.js with the same requests, answers and errors:
++ *
++ * GET /voice-channels { channels } (private calls: creator, invitees, staff)
++ * GET /voice-channels/:channelId { channel } | 404
++ * POST /voice-channels 201 { channel } | 400 (one channel per person)
++ * DELETE /voice-channels/:channelId { deleted: true } | 403
++ * POST /voice-channels/call-user { invited, reusedChannel, channel } — rings a person
++ * POST /voice-channels/call-user/respond { ok: true } — the callee's answer, relayed to the caller
++ * PUT /:id/call { call_mode, channelId, participants, participant_count } (the streamer)
++ * GET /:id/call the same, for anyone
++ *
++ * What changed: accounts and streams come from live-context; the stream's call mode is the mode of
++ * its channel here (Chat owns it; Live's streams.call_mode is no longer read); the ring is a row of
++ * `calls` (./lifecycle.js) — a callee who never answers is missed after CALL_RING_TIMEOUT_MS and the
++ * caller gets the same `vc-call-response` (status no-answer) the callee's browser sends when it gives
++ * up; Live's cross-site VC_CALL_INVITE notification is asked of Live (notify/call-invite).
++ */
++'use strict';
++
++const express = require('express');
++const config = require('../config');
++const ctx = require('../live-context');
++const dm = require('../chat/dm');
++const chatServer = require('../chat/chat-server');
++const { requireAuth, optionalAuth } = require('../auth/auth');
++const callServer = require('./call-server');
++const lifecycle = require('./lifecycle');
++
++const router = express.Router();
++const MODES = ['mic', 'mic+cam', 'cam+mic'];
++
++/* ── Voice Channels (global, non-stream) ───────────────────── */
++
++router.get('/voice-channels', optionalAuth, (req, res) => {
++ try {
++ res.set('Cache-Control', 'no-store'); // private calls differ per viewer; the list is pushed anyway
++ res.json({ channels: callServer.listChannels(req.user || null) });
++ } catch (err) {
++ console.error('[Calls]', err.message);
++ res.status(500).json({ error: 'Failed to list voice channels' });
++ }
++});
++
++router.get('/voice-channels/:channelId', optionalAuth, (req, res) => {
++ try {
++ const ch = callServer.getChannel(req.params.channelId, req.user || null);
++ if (!ch) return res.status(404).json({ error: 'Channel not found' });
++ res.json({ channel: ch });
++ } catch (err) {
++ console.error('[Calls]', err.message);
++ res.status(500).json({ error: 'Failed to get voice channel' });
++ }
++});
++
++router.post('/voice-channels', requireAuth, (req, res) => {
++ try {
++ const { name, mode, maxParticipants } = req.body || {};
++ const ch = callServer.createChannel({ name, mode, createdBy: req.user.id, maxParticipants });
++ res.status(201).json({ channel: ch });
++ } catch (err) {
++ if (err.code === 'CHANNEL_LIMIT') return res.status(400).json({ error: err.message });
++ console.error('[Calls]', err.message);
++ res.status(500).json({ error: 'Failed to create voice channel' });
++ }
++});
++
++router.delete('/voice-channels/:channelId', requireAuth, (req, res) => {
++ try {
++ const ok = callServer.deleteChannel(req.params.channelId, req.user);
++ if (!ok) return res.status(403).json({ error: 'Cannot delete this channel' });
++ res.json({ deleted: true });
++ } catch (err) {
++ console.error('[Calls]', err.message);
++ res.status(500).json({ error: 'Failed to delete voice channel' });
++ }
++});
++
++/** A person by id or login: the projection, else Live. */
++async function findUser(id, username) {
++ let user = null;
++ if (id > 0) {
++ user = ctx.getUserById(id);
++ if (!user) { await ctx.ensureUsers([id]).catch(() => {}); user = ctx.getUserById(id); }
++ }
++ if (!user && username) user = await ctx.ensureUserByUsername(username);
++ return user || null;
++}
++
++/** The caller hears "no answer" when their ring times out (the frame the callee's browser sends). */
++function tellCallerNoAnswer(row, channelName, target) {
++ chatServer.sendDm(row.created_by, {
++ type: 'vc-call-response',
++ status: 'no-answer',
++ channelId: row.channel_id,
++ channelName,
++ fromUserId: target.id,
++ fromUsername: target.username,
++ fromDisplayName: target.display_name || target.username || 'Someone',
++ fromAvatarUrl: target.avatar_url || null,
++ createdAt: Date.now(),
++ });
++}
++
++const _callUserRate = new Map(); // userId → [timestamps]
++router.post('/voice-channels/call-user', requireAuth, async (req, res) => {
++ let call = null;
++ try {
++ // Six calls a minute per account: a ring is a notification on someone else's screen.
++ const now = Date.now();
++ const recent = (_callUserRate.get(req.user.id) || []).filter((t) => now - t < 60000);
++ if (recent.length >= 6) return res.status(429).json({ error: 'Slow down — try again in a minute' });
++ recent.push(now); _callUserRate.set(req.user.id, recent);
++ const targetUserId = Number(req.body?.user_id || 0);
++ const targetUsername = String(req.body?.username || '').trim();
++
++ const targetUser = await findUser(targetUserId, targetUsername);
++ if (!targetUser) return res.status(404).json({ error: 'User not found' });
++ if (targetUser.id === req.user.id) return res.status(400).json({ error: 'You cannot call yourself' });
++ try { if (dm.isBlockedEither && dm.isBlockedEither(req.user.id, targetUser.id)) return res.status(403).json({ error: 'You cannot call this user' }); } catch { /* */ }
++
++ // Reuse caller's existing temp channel if present; otherwise create a private one — a
++ // 1:1 call is not something the whole site should see listed and be able to walk into.
++ const existing = (callServer.listChannels(req.user) || []).find(ch => !ch.permanent && !ch.streamId && ch.createdBy === req.user.id) || null;
++ const channel = existing || callServer.createChannel({
++ name: `${req.user.display_name || req.user.username}'s call`,
++ mode: 'mic+cam',
++ createdBy: req.user.id,
++ maxParticipants: 8,
++ isPrivate: true,
++ });
++ call = lifecycle.openDirect({ callerId: req.user.id, targetId: targetUser.id, channelId: channel.id });
++ if (!callServer.invite(channel.id, targetUser.id)) throw new Error('the call channel is gone');
++
++ const callerName = req.user.display_name || req.user.username || 'Someone';
++ const payload = {
++ type: 'vc-call-invite',
++ channelId: channel.id,
++ channelName: channel.name,
++ fromUserId: req.user.id,
++ fromUsername: req.user.username,
++ fromDisplayName: callerName,
++ fromAvatarUrl: req.user.avatar_url || null,
++ createdAt: Date.now(),
++ };
++
++ // Real-time invite for online users via existing chat WS connections.
++ chatServer.sendDm(targetUser.id, payload);
++
++ // Persistent cross-site notification for offline users / later join (Live pushes it).
++ ctx.effects.notifyCallInvite({
++ caller_id: req.user.id,
++ target_id: targetUser.id,
++ channel_id: channel.id,
++ channel_name: channel.name,
++ });
++
++ lifecycle.ring(call.id, { timeoutMs: config.calls.ringTimeoutMs, onTimeout: (row) => tellCallerNoAnswer(row, channel.name, targetUser) });
++
++ return res.json({ invited: true, reusedChannel: !!existing, channel });
++ } catch (err) {
++ if (err.code === 'CHANNEL_LIMIT') return res.status(400).json({ error: err.message });
++ if (call) { try { lifecycle.fail(call.id, `invite_failed: ${err.message}`); } catch { /* */ } }
++ console.error('[Calls]', err.message);
++ return res.status(500).json({ error: 'Failed to call user' });
++ }
++});
++
++router.post('/voice-channels/call-user/respond', requireAuth, (req, res) => {
++ try {
++ const callerUserId = Number(req.body?.caller_user_id || 0);
++ const channelId = String(req.body?.channel_id || '').trim();
++ const channelName = String(req.body?.channel_name || 'Voice Channel').trim() || 'Voice Channel';
++ const status = String(req.body?.status || '').trim().toLowerCase();
++
++ const allowed = new Set(['accepted', 'declined', 'busy', 'no-answer', 'canceled']);
++ if (!callerUserId) return res.status(400).json({ error: 'caller_user_id is required' });
++ if (!channelId) return res.status(400).json({ error: 'channel_id is required' });
++ if (!allowed.has(status)) return res.status(400).json({ error: 'Invalid response status' });
++ if (callerUserId === req.user.id) return res.status(400).json({ error: 'Invalid caller target' });
++ // Only an invited user can answer, and only to the caller who owns that channel.
++ const ch = callServer.getChannel(channelId, req.user);
++ if (!ch || ch.createdBy !== callerUserId || !callServer.hasInvite(channelId, req.user.id)) return res.status(403).json({ error: 'No such invite' });
++
++ try { lifecycle.respond({ callerId: callerUserId, targetId: req.user.id, channelId, status }); } catch (err) { console.warn('[Calls] lifecycle:', err.message); }
++
++ const fromDisplayName = req.user.display_name || req.user.username || 'Someone';
++ chatServer.sendDm(callerUserId, {
++ type: 'vc-call-response',
++ status,
++ channelId,
++ channelName,
++ fromUserId: req.user.id,
++ fromUsername: req.user.username,
++ fromDisplayName,
++ fromAvatarUrl: req.user.avatar_url || null,
++ createdAt: Date.now(),
++ });
++
++ return res.json({ ok: true });
++ } catch (err) {
++ console.error('[Calls]', err.message);
++ return res.status(500).json({ error: 'Failed to send call response' });
++ }
++});
++
++// ── Group Call: Enable / Disable / Get Status ────────────────
++
++function callStatus(streamId, callMode) {
++ const channelId = `stream-${streamId}`;
++ return {
++ call_mode: callMode,
++ channelId: callMode ? channelId : null,
++ participants: callServer.getParticipants(channelId),
++ participant_count: callServer.getParticipantCount(channelId),
++ };
++}
++
++router.put('/:id/call', requireAuth, async (req, res) => {
++ try {
++ // Ownership and live state from Live now, not a cached copy.
++ const stream = await ctx.refreshStream(req.params.id);
++ if (!stream) return res.status(404).json({ error: 'Stream not found' });
++ if (stream.user_id !== req.user.id) {
++ return res.status(403).json({ error: 'Not your stream' });
++ }
++ if (!stream.is_live) {
++ return res.status(400).json({ error: 'Stream is not live' });
++ }
++
++ const { call_mode } = req.body || {};
++ const validModes = [...MODES, null];
++ if (!validModes.includes(call_mode)) {
++ return res.status(400).json({ error: 'Invalid call mode. Use: mic, mic+cam, cam+mic, or null to disable' });
++ }
++
++ // Create or remove stream voice channel
++ if (call_mode) {
++ callServer.createStreamChannel(stream.id, call_mode, stream.user_id, stream);
++ } else {
++ callServer.removeStreamChannel(stream.id);
++ }
++
++ res.json(callStatus(stream.id, call_mode));
++ } catch (err) {
++ console.error('[Calls] Call mode error:', err.message);
++ res.status(500).json({ error: 'Failed to update call mode' });
++ }
++});
++
++router.get('/:id/call', optionalAuth, async (req, res) => {
++ try {
++ const stream = ctx.getStreamById(req.params.id) || await ctx.ensureStream(req.params.id);
++ if (!stream) return res.status(404).json({ error: 'Stream not found' });
++ const ch = callServer.channels.get(`stream-${stream.id}`);
++ res.json(callStatus(stream.id, ch ? ch.mode : null));
++ } catch (err) {
++ console.error('[Calls]', err.message);
++ res.status(500).json({ error: 'Failed to get call status' });
++ }
++});
++
++module.exports = router;
++module.exports.MODES = MODES;
ddiff --git a/server/config.js b/server/config.js
iindex 41f5ae8..1d0ea9b 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -119,6 +119,17 @@ module.exports = {
endpoint: strip(process.env.CHAT_EVENTS_ENDPOINT || ''),
},
++ // Voice/video calls (server/calls/): Live's call server, moved. Off until the calls cutover
++ // (docs/calls-cutover.md): Chat's chat sockets and openvibe.chat's /api/ are already public,
++ // so while calls are Live's this keeps /api/streams/voice-channels…, /api/streams/:id/call and
++ // /ws/call unanswered here (404 / refused) and /internal/calls/* answering 409, and Chat never
++ // pushes a voice-channel list or a call invite of its own. A direct call nobody answers is
++ // missed after ringTimeoutMs (the callee's browser gives up after 30 s by itself).
++ calls: {
++ enabled: bool(process.env.CHAT_CALLS, false),
++ ringTimeoutMs: int(process.env.CALL_RING_TIMEOUT_MS, 45_000),
++ },
++
// Channel sound clips live on disk. At cutover this is Live's sounds directory: stored rows
// hold absolute paths into it, and Live's donation alerts read the alert sounds from there.
sounds: {
ddiff --git a/server/db/schema.sql b/server/db/schema.sql
iindex 73087a2..2dd09f5 100644
---- a/server/db/schema.sql
++++ b/server/db/schema.sql
@@@ -490,3 +490,28 @@ CREATE TABLE IF NOT EXISTS chat_event_inbox (
processed_at INTEGER NOT NULL,
PRIMARY KEY (consumer, event_id)
);
++
++-- Calls (server/calls/lifecycle.js): one row per call. kind 'direct' is a ring from one person to
++-- another (POST /api/streams/voice-channels/call-user): pending → ringing → active → ended, or
++-- missed (no answer within the ring timeout, or the callee's no-answer), declined (declined or busy)
++-- or failed (the invite could not be delivered; `end_reason` says why). kind 'channel' / 'stream' is
++-- a session of a voice channel (a stream-linked one for 'stream'): active from the first person in
++-- until the channel empties or is closed → ended. Times are ms.
++CREATE TABLE IF NOT EXISTS calls (
++ id INTEGER PRIMARY KEY AUTOINCREMENT,
++ kind TEXT NOT NULL CHECK(kind IN ('channel', 'direct', 'stream')),
++ state TEXT NOT NULL DEFAULT 'pending' CHECK(state IN ('pending', 'ringing', 'active', 'ended', 'missed', 'declined', 'failed')),
++ channel_id TEXT NOT NULL,
++ stream_id INTEGER,
++ created_by INTEGER,
++ created_by_subject TEXT,
++ target_user_id INTEGER,
++ target_subject TEXT,
++ end_reason TEXT,
++ created_at INTEGER NOT NULL,
++ started_at INTEGER,
++ answered_at INTEGER,
++ ended_at INTEGER
++);
++CREATE INDEX IF NOT EXISTS idx_calls_channel_state ON calls(channel_id, state);
++CREATE INDEX IF NOT EXISTS idx_calls_open ON calls(state) WHERE state IN ('pending', 'ringing', 'active');
ddiff --git a/server/index.js b/server/index.js
iindex 8edfd36..a776b26 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -4,7 +4,7 @@
* node server/index.js (PORT 4400; systemd unit deploy/systemd/openvibe-chat.service)
*
* Boot: database → Live projections (first full sync; Chat serves with whatever it has if Live is
-- * slow) → chat WebSocket server → HTTP app (with the Events consumer, POST /internal/events) →
++ * slow) → chat and call WebSocket servers → HTTP app (with the Events consumer, POST /internal/events) →
* background loops (Live sync, events outbox relay, Live mirror relay) → once listening, Chat's
* Events subscriptions are checked and created if missing (in the background, retried). SIGTERM
* tells clients the chat is restarting, closes sockets and exits.
@@@ -16,6 +16,7 @@ const config = require('./config');
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');
@@@ -34,11 +35,12 @@ async function start() {
const server = http.createServer();
chatServer.init(server);
++ callServer.init(server);
const bridge = createBridge({ chatServer });
const mirror = createMirror({ config });
const relay = createRelay({ config });
const events = createEventsConsumer({ chatServer, secrets: config.events.secrets });
-- const { app, handleUpgrade } = createApp({ chatServer, bridge, mirror, relay, events });
++ const { app, handleUpgrade } = createApp({ chatServer, bridge, 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();
@@@ -46,7 +48,7 @@ async function start() {
events.start();
await new Promise((resolve) => server.listen(config.port, config.host, resolve));
-- console.log(`[Chat] listening on http://${config.host}:${server.address().port} (ws /ws/chat)`);
++ console.log(`[Chat] listening on http://${config.host}:${server.address().port} (ws /ws/chat${config.calls.enabled ? ', /ws/call' : ''})`);
const subs = subscriptions.startAtBoot({ config, port: server.address().port });
let stopping = false;
@@@ -68,12 +70,13 @@ async function start() {
subs.stop();
try { await Promise.race([mirror.flush(), new Promise((r) => setTimeout(r, 3000))]); } catch { /* */ }
try { chatServer.close(); } catch { /* */ }
++ try { callServer.close(); } catch { /* */ }
server.close(() => { db.close(); process.exit(0); });
setTimeout(() => process.exit(0), 5000).unref();
};
process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);
-- return { server, mirror, relay, bridge, events, subscriptions: subs, shutdown };
++ return { server, mirror, relay, bridge, events, callServer, subscriptions: subs, shutdown };
}
if (require.main === module) {
ddiff --git a/server/live-context.js b/server/live-context.js
iindex 74c1484..b1e60a6 100644
---- a/server/live-context.js
++++ b/server/live-context.js
@@@ -29,7 +29,8 @@
* users getUserById, getUserByUsername, getUserByDisplayName (sync, projection)
* ensureUsers(ids), ensureUserByUsername(name) (async, fetch misses)
* streams getStreamById, latestStreamIdForUser, getLiveStreamsByUserId,
-- * getStreamsByUserId, getManagedStreamsByUserId (sync) · ensureStream(id)
++ * getStreamsByUserId, getManagedStreamsByUserId (sync) · ensureStream(id),
++ * refreshStream(id) (a fresh read past the cache)
* channels getChannelById, getChannelByUserId (sync) · ensureChannelForUser(uid),
* createChannel(uid) (effect)
* policy getChannelModerationSettings(channelId), isChannelModerator(uid, channelId),
@@@ -276,6 +277,16 @@ async function ensureStream(id) {
if (!db.get('SELECT 1 FROM ctx_streams WHERE id = ?', [sid])) await _streamFetch.ensure(sid);
return db.get(`${STREAM_SELECT} WHERE s.id = ?`, [sid]) || null;
}
++/**
++ * Read one stream from Live now, past the cache (a call hook or an ownership check wants its live
++ * state this moment, not up to 10 s old). Falls back to the projection when Live does not answer.
++ */
++async function refreshStream(id) {
++ const sid = parseInt(id, 10);
++ if (!sid) return null;
++ try { await _streamFetch.refresh(sid); } catch (err) { console.warn(`[LiveContext] stream ${sid}: ${err.message}`); }
++ return db.get(`${STREAM_SELECT} WHERE s.id = ?`, [sid]) || null;
++}
function latestStreamIdForUser(userId) {
if (!userId) return null;
return db.get('SELECT id FROM ctx_streams WHERE user_id = ? ORDER BY id DESC LIMIT 1', [userId])?.id || null;
@@@ -779,6 +790,8 @@ const effects = {
translate: (text, channelUserId) => effect('translate', { text, channel_user_id: channelUserId }).then((r) => r.translation || null),
notifyDm: (body) => fire('notify/dm', body),
markDmRead: (userId, conversationId) => fire('notify/dm-read', { user_id: userId, conversation_id: conversationId }),
++ // A call-user ring: Live's cross-site VC_CALL_INVITE notification (server/calls/routes.js).
++ notifyCallInvite: (body) => fire('notify/call-invite', body),
// Awaitable (a sound's Media copy is removed from Live's row before the row goes), never throws.
assetSync: (op, assetId) => effect('asset-sync', { op, asset_id: assetId || null }).catch((err) => console.warn(`[LiveContext] asset-sync: ${err.message}`)),
userProfile: (username, viewerId) => read(`/users/profile?username=${encodeURIComponent(username)}${viewerId ? `&viewer_id=${viewerId}` : ''}`),
@@@ -794,7 +807,7 @@ module.exports = {
authenticate, authFailureReason, upsertUser, invalidateUser, subjectFor,
getUserById, getUserByUsername, getUserByDisplayName, ensureUsers, ensureUserByUsername,
// streams + channels
-- getStreamById, ensureStream, latestStreamIdForUser, getLiveStreamsByUserId, getStreamsByUserId, getManagedStreamsByUserId,
++ getStreamById, ensureStream, refreshStream, latestStreamIdForUser, getLiveStreamsByUserId, getStreamsByUserId, getManagedStreamsByUserId,
getChannelById, getChannelByUserId, ensureChannelForUser, createChannel,
// policy
getChannelModerationSettings, isChannelModerator, channelLanguage, getChannelAlertSoundsByUser, ensurePolicy, invalidateChannel,
ddiff --git a/test/calls.test.js b/test/calls.test.js
nnew file mode 100644
iindex 0000000..aa3a6e9
---- /dev/null
++++ b/test/calls.test.js
@@@ -0,0 +1,495 @@
++'use strict';
++/**
++ * Calls (WS-I task 1, server/calls/): Live's /ws/call protocol, voice-channel REST routes and stream
++ * hooks, served by Chat. Signalling between two signed-in sockets, the limits (8 per channel, 3 per
++ * address), kick with its cooldown, bans, private calls, the same-account replacement, auth-update;
++ * the REST routes with Live's shapes (list/create/delete, call-user ringing → accept → active,
++ * decline, busy, missed after the ring timeout, a failed invite); stream channels through
++ * /internal/calls (service token, capability, loopback only) and PUT/GET /:id/call; the `calls`
++ * rows of each; and CHAT_CALLS off = nothing answers.
++ */
++const assert = require('assert');
++const WebSocket = require('ws');
++const { ids } = require('openvibe-contracts');
++const { boot, suite } = require('./helpers');
++
++const t = suite('calls');
++let h, config, callServer, lifecycle;
++let ann, bob, cat, dan, eve, kay, staff, streamer, liveStream, offStream;
++
++/** Open /ws/call as a browser would (through nginx: the address in CF-Connecting-IP). */
++function callWs({ channelId, token = null, ip = '203.0.113.20', origin = 'https://openvibe.live', onOpen = null } = {}) {
++ return new Promise((resolve, reject) => {
++ const qs = new URLSearchParams();
++ if (channelId != null) qs.set('channelId', channelId);
++ if (token) qs.set('token', token);
++ const ws = new WebSocket(`ws://127.0.0.1:${h.port}/ws/call?${qs}`, { headers: { 'cf-connecting-ip': ip, origin } });
++ const all = [];
++ const waiters = [];
++ ws.on('message', (d) => {
++ let m; try { m = JSON.parse(d.toString()); } catch { return; }
++ all.push(m);
++ for (const w of [...waiters]) if (w.pred(m)) { waiters.splice(waiters.indexOf(w), 1); clearTimeout(w.timer); w.resolve(m); }
++ });
++ ws.all = all;
++ ws.next = (pred, ms = 3000) => {
++ const hit = all.find((m) => pred(m) && !m.__taken);
++ if (hit) { hit.__taken = true; return Promise.resolve(hit); }
++ return new Promise((res, rej) => {
++ const w = { pred: (m) => { if (pred(m)) { m.__taken = true; return true; } return false; }, resolve: res };
++ w.timer = setTimeout(() => { waiters.splice(waiters.indexOf(w), 1); rej(new Error(`timed out waiting (got: ${JSON.stringify(all.map((x) => [x.type, x.message]).slice(-12))})`)); }, ms);
++ waiters.push(w);
++ });
++ };
++ ws.none = async (pred, ms = 300) => { await new Promise((r) => setTimeout(r, ms)); return !all.some(pred); };
++ ws.sendJson = (o) => ws.send(JSON.stringify(o));
++ ws.closed = new Promise((r) => ws.on('close', () => r(true)));
++ ws.on('open', () => { if (onOpen) onOpen(ws); resolve(ws); });
++ ws.on('error', (e) => reject(e));
++ ws.on('unexpected-response', (req, res) => reject(new Error(`upgrade refused ${res.statusCode}`)));
++ });
++}
++/** Join a channel: resolves with the socket once it has its welcome (ws.welcome) or error (ws.error). */
++async function join(opts) {
++ const ws = await callWs(opts);
++ const m = await ws.next((x) => x.type === 'welcome' || x.type === 'error');
++ ws.welcome = m.type === 'welcome' ? m : null;
++ ws.error = m.type === 'error' ? m.message : null;
++ return ws;
++}
++const bye = async (...sockets) => { for (const ws of sockets) { if (ws && ws.readyState === WebSocket.OPEN) { ws.close(); await ws.closed; } } await h.sleep(30); };
++const api = (method, path, u, body) => h.http(method, `/api/streams${path}`, { token: u && u.token, body });
++const rows = (where = {}) => lifecycle.list(where);
++async function until(fn, ms = 2000) {
++ const end = Date.now() + ms;
++ for (;;) {
++ const v = fn();
++ if (v) return v;
++ if (Date.now() > end) throw new Error('condition not met in time');
++ await h.sleep(20);
++ }
++}
++
++t('boot (CHAT_CALLS=1, a short ring timeout)', async () => {
++ h = await boot({ env: { CHAT_CALLS: '1', CALL_RING_TIMEOUT_MS: '1200' } });
++ config = require('../server/config');
++ callServer = require('../server/calls/call-server');
++ lifecycle = require('../server/calls/lifecycle');
++ ann = h.addUser('ann', { subject: ids.newId('user') });
++ bob = h.addUser('bob', { subject: ids.newId('user') });
++ cat = h.addUser('cat', { subject: ids.newId('user') });
++ dan = h.addUser('dan', { subject: ids.newId('user') });
++ eve = h.addUser('eve', { subject: ids.newId('user') });
++ kay = h.addUser('kay', { subject: ids.newId('user') });
++ staff = h.addUser('staffer', { subject: ids.newId('user'), role: 'global_mod' });
++ streamer = h.addUser('streamy', { subject: ids.newId('user'), role: 'streamer' });
++ const ch = h.addChannel(streamer.id);
++ liveStream = h.addStream(streamer.id, ch, { title: 'Late show' });
++ offStream = h.addStream(streamer.id, ch, { title: 'Yesterday', is_live: 0 });
++ for (const u of [ann, bob, cat, dan, eve, kay, staff, streamer]) h.ctx.upsertUser(h.live.users.get(u.id));
++ await h.ctx.sync();
++ assert.strictEqual(config.calls.enabled, true);
++});
++
++t('CHAT_CALLS off: no route, no socket, internal hooks answer 409', async () => {
++ config.calls.enabled = false;
++ try {
++ assert.strictEqual((await api('GET', '/voice-channels', ann)).status, 404);
++ assert.strictEqual((await api('GET', `/${liveStream}/call`, ann)).status, 404);
++ await assert.rejects(callWs({ channelId: 'public', token: ann.token }).then((ws) => new Promise((res, rej) => { ws.on('close', () => rej(new Error('closed'))); setTimeout(res, 300); })));
++ const r = await h.http('POST', '/internal/calls/stream-channel', { token: h.serviceToken(['chat.live_bridge.write']), body: { stream_id: liveStream, mode: 'mic', user_id: streamer.id } });
++ assert.deepStrictEqual([r.status, r.body.code], [409, 'calls.off']);
++ assert.ok(!callServer.channels.has(`stream-${liveStream}`));
++ } finally { config.calls.enabled = true; }
++});
++
++t('upgrade guard: an origin outside the allow-list is refused; a missing or unknown channel is an error', async () => {
++ await assert.rejects(callWs({ channelId: 'public', token: ann.token, origin: 'https://evil.example' }));
++ const none = await join({ token: ann.token });
++ assert.strictEqual(none.error, 'Missing channelId');
++ const unknown = await join({ channelId: 'user-0-nope', token: ann.token });
++ assert.strictEqual(unknown.error, 'Voice channel not found');
++ await Promise.all([none.closed, unknown.closed]);
++});
++
++t('public lobby: join, peer-joined, participant info, the list pushed to chat sockets, leave', async () => {
++ const chat = await h.ws({ token: dan.token });
++ const a = await join({ channelId: 'public', token: ann.token, ip: '198.51.100.1' });
++ assert.ok(a.welcome, a.error);
++ assert.deepStrictEqual([a.welcome.channelId, a.welcome.channelName, a.welcome.callMode, a.welcome.isStreamer, a.welcome.canModerate], ['public', 'Public', 'mic+cam', false, false]);
++ assert.deepStrictEqual(a.welcome.participants.map((p) => p.username), ['ann']);
++ const p = a.welcome.participants[0];
++ for (const k of ['peerId', 'username', 'anonId', 'displayName', 'userId', 'avatarUrl', 'profileColor', 'isChannelCreator', 'isStreamer', 'muted', 'cameraOff', 'forceMuted', 'forceCameraOff', 'speaking', 'nameFX', 'particleFX', 'hatFX']) assert.ok(k in p, `participant has ${k}`);
++ assert.deepStrictEqual([p.userId, p.displayName, p.cameraOff, p.muted, p.nameFX], [ann.id, 'Ann', true, false, null]);
++
++ // Cosmetics are Live's, read through live-context (warmed at join).
++ h.live.decor.set(bob.id, { cosmetic: { nameFX: 'glow', hatFX: 'crown' }, tag: null });
++ const b = await join({ channelId: 'public', token: bob.token, ip: '198.51.100.2' });
++ const joined = await a.next((m) => m.type === 'peer-joined');
++ assert.deepStrictEqual([joined.username, joined.peerId, joined.nameFX, joined.hatFX, joined.particleFX], ['bob', b.welcome.peerId, 'glow', 'crown', null]);
++ assert.strictEqual((await a.next((m) => m.type === 'participant-count' && m.count === 2)).channelId, 'public');
++ const list = await chat.next((m) => m.type === 'voice-channels' && m.channels.some((c) => c.id === 'public' && c.participantCount === 2));
++ assert.ok(list.channels.find((c) => c.id === 'public').participants.some((x) => x.username === 'bob'), 'Chat\'s chat sockets get the voice-channel list');
++
++ const anon = await join({ channelId: 'public', ip: '198.51.100.3' });
++ assert.ok(anon.welcome, anon.error);
++ const me = anon.welcome.participants.find((x) => x.peerId === anon.welcome.peerId);
++ assert.ok(/^anon\d+$/.test(me.anonId) && me.userId === null && me.displayName === me.anonId, 'signed out = anonymous, as on Live');
++
++ let session = rows({ channelId: 'public' })[0];
++ assert.deepStrictEqual([session.kind, session.state, session.end_reason], ['channel', 'active', null]);
++ await bye(b);
++ const left = await a.next((m) => m.type === 'peer-left');
++ assert.deepStrictEqual([left.peerId, left.reason, left.username], [b.welcome.peerId, 'disconnect', 'bob']);
++ await bye(a, anon, chat);
++ session = lifecycle.get(session.id);
++ assert.deepStrictEqual([session.state, session.end_reason], ['ended', 'empty'], 'the session ends when the channel empties');
++ assert.ok(session.started_at && session.ended_at >= session.started_at);
++ assert.strictEqual(callServer.getParticipantCount('public'), 0);
++});
++
++t('signalling: offer / answer / ice-candidate relayed between two signed-in sockets; state frames', async () => {
++ const a = await join({ channelId: 'public', token: ann.token, ip: '198.51.100.1' });
++ const b = await join({ channelId: 'public', token: bob.token, ip: '198.51.100.2' });
++ const A = a.welcome.peerId;
++ const B = b.welcome.peerId;
++ const sdp = { type: 'offer', sdp: 'v=0\r\no=- 1 2 IN IP4 127.0.0.1\r\n' };
++ a.sendJson({ type: 'offer', targetPeerId: B, sdp });
++ assert.deepStrictEqual(await b.next((m) => m.type === 'offer'), { type: 'offer', fromPeerId: A, sdp, __taken: true });
++ b.sendJson({ type: 'answer', targetPeerId: A, sdp: { type: 'answer', sdp: 'v=0\r\n' } });
++ assert.strictEqual((await a.next((m) => m.type === 'answer')).fromPeerId, B);
++ const candidate = { candidate: 'candidate:1 1 udp 2122260223 10.0.0.1 54321 typ host', sdpMid: '0', sdpMLineIndex: 0 };
++ a.sendJson({ type: 'ice-candidate', targetPeerId: B, candidate });
++ const ice = await b.next((m) => m.type === 'ice-candidate');
++ assert.deepStrictEqual([ice.fromPeerId, ice.candidate], [A, candidate]);
++ a.sendJson({ type: 'offer', targetPeerId: B, sdp: { type: 'offer', sdp: 'x'.repeat(16385) } });
++ a.sendJson({ type: 'ice-candidate', targetPeerId: B, candidate: { candidate: 7 } });
++ assert.ok(await b.none((m) => (m.type === 'offer' || m.type === 'ice-candidate') && !m.__taken), 'oversized or malformed signalling is dropped');
++
++ a.sendJson({ type: 'mute', muted: true });
++ assert.deepStrictEqual(await b.next((m) => m.type === 'peer-muted'), { type: 'peer-muted', peerId: A, muted: true, __taken: true });
++ a.sendJson({ type: 'camera-off', cameraOff: false });
++ assert.strictEqual((await b.next((m) => m.type === 'peer-camera')).cameraOff, false);
++ a.sendJson({ type: 'speaking', speaking: true });
++ assert.strictEqual((await b.next((m) => m.type === 'peer-speaking')).speaking, true);
++ assert.deepStrictEqual(callServer.getParticipants('public').find((x) => x.peerId === A).muted, true);
++
++ // The same account again replaces its older socket.
++ const a2 = await join({ channelId: 'public', token: ann.token, ip: '198.51.100.9' });
++ assert.ok(a2.welcome);
++ await a.next((m) => m.type === 'replaced');
++ await a.closed;
++ assert.deepStrictEqual(callServer.getParticipants('public').map((x) => x.username).sort(), ['ann', 'bob']);
++ await bye(a2, b);
++});
++
++t('auth-update: an anonymous socket signs in (sent before its welcome, handled after, in order)', async () => {
++ const other = await join({ channelId: 'public', token: bob.token, ip: '198.51.100.2' });
++ const s = await callWs({ channelId: 'public', ip: '198.51.100.4', onOpen: (ws) => ws.sendJson({ type: 'auth-update', token: cat.token }) });
++ const w = await s.next((m) => m.type === 'welcome');
++ assert.ok(w.participants.find((x) => x.peerId === w.peerId).anonId, 'joined as anon');
++ const self = await s.next((m) => m.type === 'self-updated');
++ assert.deepStrictEqual([self.participant.username, self.participant.userId, self.participant.anonId, self.canModerate], ['cat', cat.id, null, false]);
++ const upd = await other.next((m) => m.type === 'peer-updated');
++ assert.deepStrictEqual([upd.peerId, upd.username], [w.peerId, 'cat']);
++ s.sendJson({ type: 'auth-update', token: dan.token });
++ const again = await s.next((m) => m.type === 'self-updated');
++ assert.strictEqual(again.participant.username, 'cat', 'a different account cannot take over the socket');
++ await bye(s, other);
++});
++
++t('limits: 3 sockets per address, 8 per channel', async () => {
++ const sockets = [];
++ for (let i = 0; i < 3; i++) sockets.push(await join({ channelId: 'public', ip: '192.0.2.10' }));
++ assert.ok(sockets.every((s) => s.welcome));
++ const fourth = await join({ channelId: 'public', ip: '192.0.2.10' });
++ assert.strictEqual(fourth.error, 'Too many connections from your network');
++ await fourth.closed;
++ for (let i = 0; i < 3; i++) sockets.push(await join({ channelId: 'public', ip: '192.0.2.11' }));
++ for (let i = 0; i < 2; i++) sockets.push(await join({ channelId: 'public', ip: '192.0.2.12' }));
++ assert.strictEqual(callServer.getParticipantCount('public'), 8);
++ const ninth = await join({ channelId: 'public', token: dan.token, ip: '192.0.2.13' });
++ assert.strictEqual(ninth.error, 'Channel full (max 8)');
++ await bye(...sockets);
++
++ // A channel's own maximum (2..8).
++ const r = await api('POST', '/voice-channels', eve, { name: 'Tiny', mode: 'mic', maxParticipants: 2 });
++ assert.deepStrictEqual([r.status, r.body.channel.maxParticipants, r.body.channel.mode], [201, 2, 'mic']);
++ const e1 = await join({ channelId: r.body.channel.id, token: eve.token, ip: '192.0.2.20' });
++ const e2 = await join({ channelId: r.body.channel.id, token: bob.token, ip: '192.0.2.21' });
++ const e3 = await join({ channelId: r.body.channel.id, token: cat.token, ip: '192.0.2.22' });
++ assert.ok(e1.welcome && e2.welcome);
++ assert.strictEqual(e3.error, 'Channel full (max 2)');
++ await bye(e1, e2);
++ assert.strictEqual((await api('DELETE', `/voice-channels/${r.body.channel.id}`, eve)).status, 200);
++});
++
++let kayChannel;
++t('kick: the target leaves, may not come back for a minute, then may', async () => {
++ const r = await api('POST', '/voice-channels', kay, { name: 'Kay\'s room' });
++ kayChannel = r.body.channel.id;
++ const k = await join({ channelId: kayChannel, token: kay.token, ip: '198.51.100.30' });
++ assert.deepStrictEqual([k.welcome.isStreamer, k.welcome.canModerate], [true, true], 'the creator runs the room');
++ const b = await join({ channelId: kayChannel, token: bob.token, ip: '198.51.100.31' });
++ assert.strictEqual(b.welcome.canModerate, false);
++ b.sendJson({ type: 'kick', targetPeerId: k.welcome.peerId });
++ assert.ok(await k.none((m) => m.type === 'kicked'), 'only moderators kick');
++ k.sendJson({ type: 'kick', targetPeerId: b.welcome.peerId });
++ await b.next((m) => m.type === 'kicked');
++ await b.closed;
++ const left = await k.next((m) => m.type === 'peer-left');
++ assert.deepStrictEqual([left.peerId, left.reason], [b.welcome.peerId, 'kicked']);
++ const back = await join({ channelId: kayChannel, token: bob.token, ip: '198.51.100.31' });
++ assert.strictEqual(back.error, 'You were removed from this channel; try again in a minute');
++ // The cooldown runs out.
++ callServer.kickCooldown.set(`${kayChannel}:u:${bob.id}`, Date.now() - 1);
++ const later = await join({ channelId: kayChannel, token: bob.token, ip: '198.51.100.31' });
++ assert.ok(later.welcome, later.error);
++ k.sendJson({ type: 'force-mute', targetPeerId: later.welcome.peerId, forceMuted: true });
++ assert.strictEqual((await later.next((m) => m.type === 'force-muted')).forceMuted, true);
++ assert.strictEqual((await k.next((m) => m.type === 'peer-force-muted')).peerId, later.welcome.peerId);
++ await bye(later, k);
++});
++
++t('ban and unban', async () => {
++ const k = await join({ channelId: kayChannel, token: kay.token, ip: '198.51.100.30' });
++ const c = await join({ channelId: kayChannel, token: cat.token, ip: '198.51.100.32' });
++ k.sendJson({ type: 'ban', targetPeerId: c.welcome.peerId });
++ await c.next((m) => m.type === 'banned');
++ assert.strictEqual((await k.next((m) => m.type === 'peer-left')).reason, 'banned');
++ assert.ok(callServer.getCallBans(kayChannel).includes(`u:${cat.id}`));
++ const again = await join({ channelId: kayChannel, token: cat.token, ip: '198.51.100.33' });
++ assert.strictEqual(again.error, 'You are banned from this voice channel');
++ k.sendJson({ type: 'unban', userId: cat.id });
++ await h.sleep(50);
++ const back = await join({ channelId: kayChannel, token: cat.token, ip: '198.51.100.33' });
++ assert.ok(back.welcome, back.error);
++ // The creator ends the call: everyone gets call-ended, the channel is gone.
++ k.sendJson({ type: 'end-call' });
++ await back.next((m) => m.type === 'call-ended');
++ await k.next((m) => m.type === 'call-ended');
++ assert.ok(!callServer.channels.has(kayChannel));
++ const session = rows({ channelId: kayChannel })[0];
++ assert.deepStrictEqual([session.kind, session.state, session.end_reason], ['channel', 'ended', 'ended']);
++ await bye(back, k);
++});
++
++t('REST: list, get, create (one each), delete (creator or staff)', async () => {
++ let r = await api('GET', '/voice-channels', null);
++ assert.strictEqual(r.status, 200);
++ assert.strictEqual(r.headers.get('cache-control'), 'no-store');
++ const pub = r.body.channels.find((c) => c.id === 'public');
++ assert.deepStrictEqual([pub.permanent, pub.participantCount, pub.maxParticipants], [true, 0, 8]);
++ assert.strictEqual((await api('POST', '/voice-channels', null, { name: 'x' })).status, 401);
++ r = await api('POST', '/voice-channels', dan, { name: 'Dan\'s den', mode: 'weird', maxParticipants: 99 });
++ assert.strictEqual(r.status, 201);
++ const den = r.body.channel;
++ assert.deepStrictEqual([den.mode, den.maxParticipants, den.createdBy, den.private, den.permanent, 'invited' in den], ['mic+cam', 8, dan.id, false, false, false]);
++ assert.ok(den.id.startsWith(`user-${dan.id}-`));
++ r = await api('POST', '/voice-channels', dan, { name: 'Second' });
++ assert.deepStrictEqual([r.status, r.body.error], [400, 'You already have a voice channel. Delete it first.']);
++ assert.strictEqual((await api('GET', `/voice-channels/${den.id}`, null)).body.channel.name, 'Dan\'s den');
++ assert.strictEqual((await api('GET', '/voice-channels/nope', null)).status, 404);
++ assert.deepStrictEqual([(await api('DELETE', `/voice-channels/${den.id}`, bob)).status], [403]);
++ assert.strictEqual((await api('DELETE', '/voice-channels/public', staff)).status, 403, 'the lobby stays');
++ assert.deepStrictEqual((await api('DELETE', `/voice-channels/${den.id}`, staff)).body, { deleted: true }, 'staff may delete any');
++ assert.strictEqual((await api('GET', `/voice-channels/${den.id}`, null)).status, 404);
++});
++
++let annCall;
++t('call-user: ringing → accepted → active → ended; private call; Live asked for the notification', async () => {
++ const annChat = await h.ws({ token: ann.token });
++ const catChat = await h.ws({ token: cat.token });
++ await h.sleep(100);
++ let r = await api('POST', '/voice-channels/call-user', ann, { username: 'cat' });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.deepStrictEqual([r.body.invited, r.body.reusedChannel, r.body.channel.private, r.body.channel.name], [true, false, true, 'Ann\'s call']);
++ annCall = r.body.channel.id;
++ const invite = await catChat.next((m) => m.type === 'vc-call-invite');
++ assert.deepStrictEqual([invite.channelId, invite.channelName, invite.fromUserId, invite.fromUsername, invite.fromDisplayName], [annCall, 'Ann\'s call', ann.id, 'ann', 'Ann']);
++ let ring = rows({ kind: 'direct' })[0];
++ assert.deepStrictEqual([ring.state, ring.created_by, ring.target_user_id, ring.channel_id, ring.created_by_subject, ring.target_subject], ['ringing', ann.id, cat.id, annCall, ann.subject_id, cat.subject_id]);
++ const note = await until(() => h.live.effects.find((e) => e.name === 'notify/call-invite'));
++ assert.deepStrictEqual(note.body, { caller_id: ann.id, target_id: cat.id, channel_id: annCall, channel_name: 'Ann\'s call' });
++
++ // Private: listed and joinable for the caller, the invitee and staff only.
++ assert.ok(!(await api('GET', '/voice-channels', dan)).body.channels.some((c) => c.id === annCall));
++ assert.ok((await api('GET', '/voice-channels', cat)).body.channels.some((c) => c.id === annCall));
++ assert.strictEqual((await api('GET', `/voice-channels/${annCall}`, dan)).status, 404);
++ const anon = await join({ channelId: annCall, ip: '198.51.100.40' });
++ assert.strictEqual(anon.error, 'This is a private call', 'an unauthenticated socket is refused from a private call');
++ const outsider = await join({ channelId: annCall, token: dan.token, ip: '198.51.100.41' });
++ assert.strictEqual(outsider.error, 'This is a private call');
++
++ // Only the invitee answers, and only to that caller.
++ assert.strictEqual((await api('POST', '/voice-channels/call-user/respond', dan, { caller_user_id: ann.id, channel_id: annCall, status: 'accepted' })).status, 403);
++ assert.strictEqual((await api('POST', '/voice-channels/call-user/respond', cat, { caller_user_id: ann.id, channel_id: annCall, status: 'maybe' })).status, 400);
++
++ const a = await join({ channelId: annCall, token: ann.token, ip: '198.51.100.42' });
++ r = await api('POST', '/voice-channels/call-user/respond', cat, { caller_user_id: ann.id, channel_id: annCall, channel_name: 'Ann\'s call', status: 'accepted' });
++ assert.deepStrictEqual(r.body, { ok: true });
++ const answer = await annChat.next((m) => m.type === 'vc-call-response');
++ assert.deepStrictEqual([answer.status, answer.channelId, answer.fromUserId, answer.fromUsername], ['accepted', annCall, cat.id, 'cat']);
++ ring = lifecycle.get(ring.id);
++ assert.deepStrictEqual([ring.state, ring.end_reason], ['active', null]);
++ assert.ok(ring.answered_at >= ring.started_at);
++ const c = await join({ channelId: annCall, token: cat.token, ip: '198.51.100.43' });
++ assert.ok(c.welcome, c.error);
++ await h.sleep(1400);
++ assert.strictEqual(lifecycle.get(ring.id).state, 'active', 'an answered call is not missed');
++ await bye(a, c, annChat, catChat, anon, outsider);
++ ring = lifecycle.get(ring.id);
++ assert.deepStrictEqual([ring.state, ring.end_reason], ['ended', 'empty']);
++});
++
++t('call-user: declined, busy, missed after the ring timeout (the caller hears "no-answer"), rate limit', async () => {
++ const annChat = await h.ws({ token: ann.token });
++ await h.sleep(100);
++ // declined
++ let r = await api('POST', '/voice-channels/call-user', ann, { user_id: dan.id });
++ assert.deepStrictEqual([r.status, r.body.reusedChannel, r.body.channel.id], [200, true, annCall], 'the caller\'s channel is reused');
++ await api('POST', '/voice-channels/call-user/respond', dan, { caller_user_id: ann.id, channel_id: annCall, status: 'declined' });
++ assert.strictEqual((await annChat.next((m) => m.type === 'vc-call-response')).status, 'declined');
++ const declined = rows({ kind: 'direct' })[0];
++ assert.deepStrictEqual([declined.target_user_id, declined.state, declined.end_reason], [dan.id, 'declined', 'declined']);
++ assert.ok(declined.ended_at);
++ // busy
++ await api('POST', '/voice-channels/call-user', ann, { username: 'eve' });
++ await api('POST', '/voice-channels/call-user/respond', eve, { caller_user_id: ann.id, channel_id: annCall, status: 'busy' });
++ assert.deepStrictEqual([rows({ kind: 'direct' })[0].state, rows({ kind: 'direct' })[0].end_reason], ['declined', 'busy']);
++ // missed: bob never answers
++ r = await api('POST', '/voice-channels/call-user', ann, { username: 'bob' });
++ const ringing = rows({ kind: 'direct' })[0];
++ assert.strictEqual(ringing.state, 'ringing');
++ // Ringing again re-rings the same call.
++ await api('POST', '/voice-channels/call-user', ann, { username: 'bob' });
++ assert.strictEqual(rows({ kind: 'direct' })[0].id, ringing.id);
++ const noAnswer = await annChat.next((m) => m.type === 'vc-call-response' && m.status === 'no-answer', 4000);
++ assert.deepStrictEqual([noAnswer.fromUserId, noAnswer.fromUsername, noAnswer.channelId], [bob.id, 'bob', annCall]);
++ const missed = lifecycle.get(ringing.id);
++ assert.deepStrictEqual([missed.state, missed.end_reason], ['missed', 'timeout']);
++ // bob's late "accepted" finds no ring to answer (the relay still goes, as on Live)
++ assert.strictEqual((await api('POST', '/voice-channels/call-user/respond', bob, { caller_user_id: ann.id, channel_id: annCall, status: 'accepted' })).status, 200);
++ assert.strictEqual(lifecycle.get(ringing.id).state, 'missed');
++ // Six rings a minute (ann has used five).
++ await api('POST', '/voice-channels/call-user', ann, { username: 'kay' });
++ r = await api('POST', '/voice-channels/call-user', ann, { username: 'kay' });
++ assert.deepStrictEqual([r.status, r.body.error], [429, 'Slow down — try again in a minute']);
++ // Errors of the request itself.
++ assert.strictEqual((await api('POST', '/voice-channels/call-user', eve, { username: 'nobody-here' })).status, 404);
++ assert.strictEqual((await api('POST', '/voice-channels/call-user', eve, { username: 'eve' })).status, 400);
++ await bye(annChat);
++});
++
++t('call-user: an invite that cannot be delivered is failed, with the reason', async () => {
++ const orig = h.chatServer.sendDm;
++ h.chatServer.sendDm = () => { throw new Error('socket layer down'); };
++ let r;
++ try { r = await api('POST', '/voice-channels/call-user', eve, { username: 'dan' }); } finally { h.chatServer.sendDm = orig; }
++ assert.deepStrictEqual([r.status, r.body.error], [500, 'Failed to call user']);
++ const row = rows({ kind: 'direct' })[0];
++ assert.deepStrictEqual([row.created_by, row.state, row.end_reason], [eve.id, 'failed', 'invite_failed: socket layer down']);
++ // Deleting a channel ends the rings still waiting on it (canceled).
++ r = await api('POST', '/voice-channels/call-user', eve, { username: 'dan' });
++ assert.strictEqual(rows({ kind: 'direct' })[0].state, 'ringing');
++ await api('DELETE', `/voice-channels/${r.body.channel.id}`, eve);
++ assert.deepStrictEqual([rows({ kind: 'direct' })[0].state, rows({ kind: 'direct' })[0].end_reason], ['ended', 'canceled']);
++});
++
++t('stream channel through /internal/calls: guard, create, join (legacy id), mode change, removal', async () => {
++ const token = h.serviceToken(['chat.live_bridge.write']);
++ const body = { stream_id: liveStream, mode: 'mic', user_id: streamer.id };
++ const post = (b, opts = {}) => h.http('POST', '/internal/calls/stream-channel', { token, body: b, ...opts });
++ assert.strictEqual((await h.http('POST', '/internal/calls/stream-channel', { body })).status, 401);
++ assert.strictEqual((await h.http('POST', '/internal/calls/stream-channel', { token: h.serviceToken(['chat.presence.read']), body })).status, 403);
++ assert.strictEqual((await h.http('POST', '/internal/calls/stream-channel', { token: h.serviceToken(['chat.live_bridge.write'], { aud: 'openvibe.live' }), body })).status, 401);
++ assert.strictEqual((await post(body, { headers: { 'X-Forwarded-For': '1.2.3.4' } })).status, 403, 'loopback only');
++ assert.strictEqual((await post({ ...body, mode: 'loud' })).status, 400);
++ assert.strictEqual((await post({ ...body, stream_id: 'x' })).status, 400);
++
++ let r = await post(body);
++ assert.strictEqual(r.status, 200, r.text);
++ assert.deepStrictEqual([r.body.channel.id, r.body.channel.name, r.body.channel.mode, r.body.channel.streamId, r.body.channel.createdBy], [`stream-${liveStream}`, 'Late show', 'mic', liveStream, streamer.id]);
++ r = await api('GET', `/${liveStream}/call`, null);
++ assert.deepStrictEqual(r.body, { call_mode: 'mic', channelId: `stream-${liveStream}`, participants: [], participant_count: 0 });
++
++ const viewer = await join({ channelId: String(liveStream), token: bob.token, ip: '198.51.100.50' });
++ assert.deepStrictEqual([viewer.welcome.channelId, viewer.welcome.callMode, viewer.welcome.isStreamer, viewer.welcome.canModerate], [`stream-${liveStream}`, 'mic', false, false]);
++ const host = await join({ channelId: `stream-${liveStream}`, token: streamer.token, ip: '198.51.100.51' });
++ assert.deepStrictEqual([host.welcome.isStreamer, host.welcome.canModerate], [true, true]);
++ assert.strictEqual(host.welcome.participants.find((p) => p.username === 'streamy').isStreamer, true);
++ r = await api('GET', `/${liveStream}/call`, null);
++ assert.strictEqual(r.body.participant_count, 2);
++ let session = rows({ channelId: `stream-${liveStream}` })[0];
++ assert.deepStrictEqual([session.kind, session.state, session.stream_id, session.created_by], ['stream', 'active', liveStream, streamer.id]);
++
++ // A new mode ends the call in progress (Live's createStreamChannel).
++ r = await post({ ...body, mode: 'cam+mic' });
++ assert.strictEqual(r.body.channel.mode, 'cam+mic');
++ await viewer.next((m) => m.type === 'call-ended');
++ await host.next((m) => m.type === 'call-ended');
++ assert.deepStrictEqual([lifecycle.get(session.id).state, lifecycle.get(session.id).end_reason], ['ended', 'ended']);
++ await bye(viewer, host);
++
++ const v2 = await join({ channelId: `stream-${liveStream}`, token: bob.token, ip: '198.51.100.50' });
++ assert.strictEqual(v2.welcome.callMode, 'cam+mic');
++ session = rows({ channelId: `stream-${liveStream}` })[0];
++ r = await h.http('DELETE', `/internal/calls/stream-channel/${liveStream}`, { token });
++ assert.deepStrictEqual(r.body, { ok: true, removed: true });
++ await v2.next((m) => m.type === 'call-ended');
++ assert.deepStrictEqual([lifecycle.get(session.id).state, lifecycle.get(session.id).end_reason], ['ended', 'stream_ended']);
++ assert.deepStrictEqual((await api('GET', `/${liveStream}/call`, null)).body, { call_mode: null, channelId: null, participants: [], participant_count: 0 });
++ assert.deepStrictEqual((await h.http('DELETE', `/internal/calls/stream-channel/${liveStream}`, { token })).body, { ok: true, removed: false });
++ const gone = await join({ channelId: `stream-${liveStream}`, token: bob.token, ip: '198.51.100.50' });
++ assert.strictEqual(gone.error, 'Voice channel not found');
++ await bye(v2);
++
++ // A stream that is not live has a channel nobody can join.
++ await post({ stream_id: offStream, mode: 'mic', user_id: streamer.id });
++ const off = await join({ channelId: `stream-${offStream}`, token: bob.token, ip: '198.51.100.52' });
++ assert.strictEqual(off.error, 'Stream not live');
++ await h.http('DELETE', `/internal/calls/stream-channel/${offStream}`, { token });
++});
++
++t('PUT /api/streams/:id/call: the streamer only, a live stream only, a known mode', async () => {
++ assert.strictEqual((await api('PUT', `/${liveStream}/call`, null, { call_mode: 'mic' })).status, 401);
++ assert.deepStrictEqual([(await api('PUT', `/${liveStream}/call`, bob, { call_mode: 'mic' })).status], [403]);
++ assert.strictEqual((await api('PUT', '/999999/call', streamer, { call_mode: 'mic' })).status, 404);
++ assert.strictEqual((await api('PUT', `/${offStream}/call`, streamer, { call_mode: 'mic' })).status, 400);
++ assert.strictEqual((await api('PUT', `/${liveStream}/call`, streamer, { call_mode: 'loud' })).status, 400);
++ let r = await api('PUT', `/${liveStream}/call`, streamer, { call_mode: 'mic+cam' });
++ assert.deepStrictEqual(r.body, { call_mode: 'mic+cam', channelId: `stream-${liveStream}`, participants: [], participant_count: 0 });
++ assert.strictEqual((await api('GET', `/${liveStream}/call`, null)).body.call_mode, 'mic+cam');
++ r = await api('PUT', `/${liveStream}/call`, streamer, { call_mode: null });
++ assert.deepStrictEqual(r.body, { call_mode: null, channelId: null, participants: [], participant_count: 0 });
++ assert.ok(!callServer.channels.has(`stream-${liveStream}`));
++ assert.strictEqual((await api('GET', '/999999/call', null)).status, 404);
++});
++
++t('lifecycle: transitions are one-way; a restart closes what was left open', async () => {
++ const d = lifecycle.openDirect({ callerId: ann.id, targetId: kay.id, channelId: 'user-x' });
++ assert.strictEqual(d.state, 'pending');
++ assert.strictEqual(lifecycle.transition(d.id, 'active'), null, 'pending cannot become active without ringing');
++ assert.strictEqual(lifecycle.transition(d.id, 'declined'), null);
++ lifecycle.ring(d.id, { timeoutMs: 60000 });
++ const declined = lifecycle.respond({ callerId: ann.id, targetId: kay.id, channelId: 'user-x', status: 'declined' });
++ assert.strictEqual(declined.state, 'declined');
++ assert.strictEqual(lifecycle.transition(d.id, 'active'), null, 'a final state stays final');
++ assert.strictEqual(lifecycle.respond({ callerId: ann.id, targetId: kay.id, channelId: 'user-x', status: 'accepted' }), null);
++
++ const pending = lifecycle.openDirect({ callerId: bob.id, targetId: kay.id, channelId: 'user-y' });
++ const ringing = lifecycle.ring(lifecycle.openDirect({ callerId: cat.id, targetId: kay.id, channelId: 'user-z' }).id, { timeoutMs: 60000 });
++ const active = lifecycle.sessionStarted({ channelId: 'user-z' });
++ assert.strictEqual(lifecycle.sessionStarted({ channelId: 'user-z' }).id, active.id, 'one session per channel');
++ const out = lifecycle.recover();
++ assert.ok(out.failed >= 1 && out.missed >= 1 && out.ended >= 1, JSON.stringify(out));
++ assert.deepStrictEqual([lifecycle.get(pending.id).state, lifecycle.get(pending.id).end_reason], ['failed', 'restart']);
++ assert.deepStrictEqual([lifecycle.get(ringing.id).state, lifecycle.get(ringing.id).end_reason], ['missed', 'restart']);
++ assert.deepStrictEqual([lifecycle.get(active.id).state, lifecycle.get(active.id).end_reason], ['ended', 'restart']);
++ assert.strictEqual(h.db.get("SELECT COUNT(*) AS n FROM calls WHERE state IN ('pending', 'ringing', 'active')").n, 0);
++});
++
++t.run(async () => {
++ try { if (callServer) callServer.close(); } catch { /* */ }
++ if (h && h.close) await h.close();
++});