openvibers/openrestream Commits

RTMP cutover runbook and tooling. docs/cutover.md covers the whole cutover with owner and agent steps, a check and an undo for each: deploy the newest release, OPENRE_RTMP_BIND=0.0.0.0 through a new worker generation, 1936/tcp at the provider edge (owner), Live's OPENRE_URL/OPENRE_PUBLIC_URL/OPENRE_EVENTS_SECRET and the Events subscription, a rehearsal with a real destination, and the per-slot migration with the switch, key regeneration and personal-key rotation (H4), plus rollback per slot and for every slot at once.

OP OpenVibers committed 23 Sept 2026 · e0dc1b46fd · 12 files, +1799 −22

New scripts:
- scripts/cutover-preflight.js checks, read-only: the release (and that it contains 6dc78a5), the service, the env, the bind, DNS, the port from outside (RTMP handshake, or an external probe URL), the database integrity, Live's env file and running process, the Events subscription (secret compared by hash), and one slot.
- scripts/subscribe-live-events.js creates, lists, disables or enables the openre.session.* -> Live subscription, as Live.
- scripts/set-definition-state.js disables or re-enables one slot's definition, as Live, for the per-slot rollback.

test/bind-switch.test.js proves the loopback-to-public generation switch with real workers. Fix: the migration's snapshot path is /opt/openvibe.live/data/live.db.

Changes

ddiff --git a/.env.example b/.env.example
iindex 15a464c..16afca7 100644
---- a/.env.example
++++ b/.env.example
@@@ -44,6 +44,9 @@ OPENRE_SECRETS_KEY=
OPENRE_RTMP_PORT=1936
# More public ports every worker also serves, e.g. 1935 after Live's RTMP is retired.
#OPENRE_RTMP_EXTRA_PORTS=1935
++# Public listener address (default 0.0.0.0). 127.0.0.1 keeps ingest host-local (production until the
++# cutover, docs/cutover.md step 5). A new value takes effect with the next worker generation
++# (deploy.sh workers), never by restarting a worker.
#OPENRE_RTMP_BIND=0.0.0.0
# What encoders are told: rtmp://<host>[:<port>]/live
#OPENRE_RTMP_PUBLIC_HOST=ingest.openre.stream
@@@ -98,5 +101,12 @@ MEDIA_API_KEY=
#OPENRE_RECORDING=on
#OPENRE_RECORDING_START_DELAY_MS=2000
++# ── Restore drills ────────────────────────────────────────────
++# OPENRE_DRILL=1 only in an `ovhost drill openre` instance (OpenVibe.Host sets it in the drill's
++# override file): openre-api serves reads from the restored copy and refuses writes, /play/ and
++# sign-in; the coordinator and every worker refuse to start; no events, no Media calls.
++# NEVER set it here in production (scripts/cutover-preflight.js --only env fails if it is).
++#OPENRE_DRILL=
++
# ── Links ─────────────────────────────────────────────────────
#OV_LIVE_URL=https://openvibe.live
ddiff --git a/README.md b/README.md
iindex 612e143..d7cb487 100644
---- a/README.md
++++ b/README.md
@@@ -157,6 +157,8 @@ sudo deploy/scripts/deploy.sh status
- An **API deploy** never restarts a worker unit (no unit depends on another). Viewers of `openre.stream/play/…` reconnect; encoders, restreams and recordings do not notice.
- A **worker deploy** starts `openre-rtmp-ingest@<sha>` and `openre-restream-worker@<sha>`; older instances are disabled (not stopped) and exit on their own when drained. Never `systemctl restart` a worker instance during a broadcast; `systemctl stop` starts a drain and waits up to 30 min (`TimeoutStopSec`), then kills.
++- Worker instances are named after the release. `deploy.sh workers <sha>` for a release that already runs a generation does nothing. An env change that workers read, such as `OPENRE_RTMP_BIND`, takes effect with the next release's generation (docs/cutover.md A3).
++- **Restore drills** (`ovhost drill openre`, OpenVibe.Host) start only `openre-api`, with `OPENRE_DRILL=1`, on a restored copy of the database. In drill mode the API serves reads and refuses writes, `/play/` and sign-in with 503 `openre.drill_read_only`. The coordinator and every worker exit before they open the database. The event relay and Media calls are off (`test/drill-mode.test.js`). Never set `OPENRE_DRILL` in `/etc/openvibe/openre.env`; the preflight's `env` check fails if it is set.
- First install: create the `openre` OAuth client and grants in Network, `/etc/openvibe/openre.env` (0600), the certificate for `openre.stream`, DNS for `openre.stream` (proxied) and `ingest.openre.stream` (DNS only), firewall 1936/tcp, then `release`, `api`, `workers`, and `systemctl enable --now openre-api openre-session-coordinator`.
## Live integration (patch)
@@@ -177,7 +179,7 @@ Not in the patch (follow-ups): live thumbnails, AI audio/vision taps and the Rob
## Migration
--`node scripts/migrate-from-live.js --live-db <snapshot> [--apply] [--slots 12,31] [--checklist out.md]` reads a **read-only snapshot** of Live's database (`sqlite3 …/live.db ".backup /tmp/live-snapshot.db"`), imports `managed_streams` as stream definitions (owner = `linked_accounts.subject_id`, recording mode/visibility from the slot and channel, refs `live:managed_stream` + `live:user`, `mirror_to_live` on) and their `restream_destinations` (keys sealed), and prints a per-channel cutover checklist. **No old key is imported**: each imported slot gets a new key nobody has seen; the broadcaster rotates to get one. Every source row is recorded in `migration_map` as imported, held (no subject yet; a private-host URL — kept disabled with the key sealed; unbound destination of a multi-slot user) or excluded (banned account), with the reason. Dry run by default; idempotent (`test/migration.test.js`).
++`node scripts/migrate-from-live.js --live-db <snapshot> [--apply] [--slots 12,31] [--checklist out.md]` reads a **read-only snapshot** of Live's database (production: `/opt/openvibe.live/data/live.db`; the snapshot commands are in [docs/cutover.md](docs/cutover.md) B2), imports `managed_streams` as stream definitions (owner = `linked_accounts.subject_id`, recording mode/visibility from the slot and channel, refs `live:managed_stream` + `live:user`, `mirror_to_live` on) and their `restream_destinations` (keys sealed), and prints a per-channel cutover checklist. **No old key is imported**: each imported slot gets a new key nobody has seen; the broadcaster rotates to get one. Every source row is recorded in `migration_map` as imported, held (no subject yet; a private-host URL — kept disabled with the key sealed; unbound destination of a multi-slot user) or excluded (banned account), with the reason. Dry run by default; idempotent (`test/migration.test.js`).
## Cutover runbook
@@@ -185,24 +187,24 @@ ADR-009: one protocol at a time (RTMP → WHIP → JSMPEG → SFU), per slot, be
### RTMP (ready)
--Once, before the first slot (status 2026-09-23: steps 1 and 2 are done, the Live patch of step 4 is deployed with the switch off; step 3, the `OPENRE_*` settings of step 4, and a rehearsal through a public 1936/tcp are not done — 1936/tcp has to be opened at the host firewall and provider edge first, then `OPENRE_RTMP_BIND` set to a public address):
--
--1. Network: OAuth client `openre` + the grants listed under "Auth". Contracts: release the proposals.
--2. Host: deploy OpenRe (see "Deploying"), `curl 127.0.0.1:4500/api/ready` shows `ready`, one `rtmp-ingest` and one `restream` worker `ready`, the coordinator lease valid.
--3. Events: subscription `openre.session.*` → `http://127.0.0.1:3000/internal/openre-events`; put its secret in Live as `OPENRE_EVENTS_SECRET`.
--4. Live: apply `docs/live-patch.diff`, set `OPENRE_URL=http://127.0.0.1:4500`, deploy Live (`deploy.sh --wait-idle`). With no slot switched this changes nothing.
--5. Rehearse with a test account: a slot, `migrate-from-live.js --apply --slots <id>`, switch, regenerate, stream from OBS to `rtmp://ingest.openre.stream:1936/live`: the channel page shows the stream, restreams go live, a VOD appears; deploy Live and `deploy.sh api` during the broadcast — nothing drops; `deploy.sh workers` — the session stays on the old generation, a new publish lands on the new one.
--
--Per channel (the checklist printed by the migration script, per slot):
--
--1. Maintenance window agreed with the broadcaster; the slot is offline.
--2. `migrate-from-live.js --live-db <fresh snapshot> --apply --slots <id>`; review destinations on openre.stream (held ones need the owner).
--3. `PUT /api/admin/openre/managed/<id>/ingest-authority {"authority":"openre"}` — Live refuses the old key from now on and has rotated its own copy.
--4. The broadcaster presses **Regenerate stream key** on the Go Live page (OpenRe issues the key, shown once) and pastes server `rtmp://ingest.openre.stream:1936/live` and the key into OBS.
--5. Rotate the broadcaster's personal Live key too (`users.stream_key`; not migrated).
--6. Test broadcast; watch `deploy.sh status`, `/api/admin/openre/status` on Live, the output health on openre.stream.
--
--Rollback (per slot): `{"authority":"live"}` — the broadcaster regenerates the key on Live and points OBS back at the Live RTMP URL shown on the Go Live page. All slots at once: unset `OPENRE_URL` in Live and restart it (waiting for idle).
++**[docs/cutover.md](docs/cutover.md) is the runbook.** It gives the exact commands, who does each step,
++the checks and the rollback for every step. In short:
++
++- **Once:** push, deploy the newest release (`deploy.sh release` and `api`), then set
++ `OPENRE_RTMP_BIND=0.0.0.0` and start the worker generation (`deploy.sh workers`). The **owner**
++ opens 1936/tcp at the provider edge. Set `OPENRE_URL`, `OPENRE_PUBLIC_URL` and
++ `OPENRE_EVENTS_SECRET` in Live, run `scripts/subscribe-live-events.js`, and restart Live. Then
++ rehearse with a test slot and a real destination.
++- **Per slot:** the broadcaster's window, a Live snapshot, `migrate-from-live.js --apply --slots <id>`,
++ `PUT /api/admin/openre/managed/<id>/ingest-authority` (owner, admin session), and Regenerate in
++ OBS. The broadcaster also rotates their personal key (hazard H4).
++- **Rollback:** per slot, `{"authority":"live"}` plus `scripts/set-definition-state.js --state disabled`.
++ For every slot at once, unset `OPENRE_URL` in Live and restart it.
++
++`scripts/cutover-preflight.js` checks, read-only, what each step needs: the release, the service,
++the env, the bind, DNS, the port from outside, the database, Live's env, the Events subscription and
++one slot. Status on 2026-09-23: OpenRe is deployed (`655b98a10aaa`) and DNS is done. The release with
++`6dc78a5`, the public bind, the provider port, Live's settings and the subscription are not.
### WHIP, JSMPEG, SFU (not ready)
ddiff --git a/docs/cutover.md b/docs/cutover.md
nnew file mode 100644
iindex 0000000..2293fbf
---- /dev/null
++++ b/docs/cutover.md
@@@ -0,0 +1,365 @@
++# RTMP cutover runbook (OpenRe.Stream)
++
++The order of work that moves RTMP ingest for Live's stream slots from Live to OpenRe.Stream, one slot
++at a time (ADR-009). Every step says who does it, how to check it, and how to undo it.
++
++- **OWNER** steps need the owner: the provider firewall, a Live admin session in the browser, the
++ broadcasters, and a real restream destination.
++- **AGENT** steps are commands on the host (`ssh openvibe-ovh`, then `sudo`).
++- Every command, path and env name below was checked against the code at the commit that added this
++ file: OpenRe `deploy/scripts/deploy.sh`, `server/config.js`, `scripts/*`; Live `server/openre/*`,
++ `deploy/scripts/deploy.sh`; Events `server/api/subscriptions.js`. The production facts (paths,
++ grants, subscriptions, listeners) were read on the host on 2026-09-23, without changing anything.
++
++## Where things stand (2026-09-23)
++
++This is the read-only preflight, run on the host against the deployed release `655b98a10aaa`:
++
++```
++FAIL release 6dc78a5 is not in /opt/openre.stream/repo yet (deploy.sh release fetches it), so 655b98a10aaa cannot contain it
++PASS service ready: rtmp-ingest#1, restream#1, coordinator lease valid, event relay on
++PASS env OPENRE_DRILL unset, RTMP port 1936, public host ingest.openre.stream, secrets and EVENTS_URL set
++FAIL bind OPENRE_RTMP_BIND=127.0.0.1; listening on 127.0.0.1:1936: set OPENRE_RTMP_BIND=0.0.0.0 …
++PASS dns ingest.openre.stream → 15.204.79.215 (this host)
++MANUAL port ingest.openre.stream:1936 (15.204.79.215) refused (nothing listens there) from this host, which never crosses the provider edge …
++PASS db integrity_check ok, no foreign key violations (stream_definitions 1, ingest_keys 1, ingest_sessions 1, destinations 0, migration_map 0)
++FAIL live-env OPENRE_URL not set (the integration is off); OPENRE_EVENTS_SECRET not set
++FAIL events no subscription live openre.session.* → http://127.0.0.1:3000/internal/openre-events …
++```
++
++(The release has no `scripts/cutover-preflight.js` yet, so this was run by piping the script to
++`node -` from `/opt/openre.stream/current`, which writes nothing.)
++
++From outside the host (the workstation), `node scripts/cutover-preflight.js --only port` answers:
++`ingest.openre.stream:1936 (15.204.79.215) ETIMEDOUT: filtered`. Port 1935, which Live uses, answers
++the RTMP handshake from outside, so the provider edge filters 1936 and lets 1935 through. The host
++firewall (ufw) is inactive. The Network grants are already in place:
++`[live, openre.stream.read|stream.write|key.rotate|session.read, openvibe.openre]`,
++`[live, events.subscription.manage, openvibe.events]` and `[openre, events.event.publish, openvibe.events]`.
++All 102 Live slots have `ingest_authority = 'live'`.
++
++## Tools
++
++Everything below runs on the host. The scripts ship with the release (`scripts/`):
++
++| Script | What it does | Writes |
++|---|---|---|
++| `cutover-preflight.js` | read-only checks: `release service env bind dns port db live-env events [slot]` (`--only`, `--skip`, `--slot <id>`, `--json`, `--strict`, `--expect-release <sha>`, `--probe-url`, `--host-ip`) | nothing |
++| `subscribe-live-events.js` | Live's Events subscription `openre.session.*` → Live, **as Live** (reads `/etc/openvibe/live.env`); `--dry-run`, `--disable`, `--enable` | one subscription in Events |
++| `migrate-from-live.js` | imports slots and destinations from a Live **snapshot** (dry run unless `--apply`) | `openre.db` |
++| `set-definition-state.js` | disables or re-enables one slot's OpenRe definition, **as Live** (per-slot rollback) | `openre.db` via the API |
++
++Shell helpers for the session (paste once per SSH session):
++
++```bash
++SHA=<the 12-character release id printed by step A2>
++# read-only preflight from that release
++pf() { sudo node /opt/openre.stream/releases/$SHA/scripts/cutover-preflight.js --expect-release "$SHA" "$@"; }
++# run a command in the current release with OpenRe's production env, AS THE SERVICE USER (ubuntu).
++# Never run a script that opens openre.db as root: root-owned -wal/-shm files lock the service out.
++openre_as_service() { sudo bash -c "set -a; . /etc/openvibe/openre.env; set +a; export NODE_ENV=production OPENRE_DB_PATH=/var/lib/openre/openre.db OPENRE_ENV_FILE=/nonexistent; cd /opt/openre.stream/current && sudo -E -u ubuntu $1"; }
++DEPLOY=/opt/openre.stream/releases/$SHA/deploy/scripts/deploy.sh
++```
++
++Live admin calls are made in the browser console on openvibe.live, **signed in as an admin**. Live
++refuses `hbt_` API tokens on `/api/admin/*`, so only an admin session in the browser can make them:
++`await api('/admin/openre/status')`.
++
++## Phase A: once, before the first slot
++
++### A1. Push (OWNER, or the lead)
++
++Push this repository's `main` to GitHub. `deploy.sh release` checks out `origin/main`, and nothing
++from this commit onward is on GitHub yet. Undo: nothing to undo.
++
++### A2. Deploy the newest release (AGENT)
++
++```bash
++sudo /opt/openre.stream/current/deploy/scripts/deploy.sh release origin/main # prints the new <sha>
++SHA=<that sha>; DEPLOY=/opt/openre.stream/releases/$SHA/deploy/scripts/deploy.sh # and re-paste pf()
++pf --only db # openre.db is sound before anything restarts
++sudo $DEPLOY api $SHA # restarts openre-api + openre-session-coordinator ONLY
++curl -s http://127.0.0.1:4500/api/ready # "status":"ready"
++pf --only release,env,db # current = $SHA, it contains 6dc78a5
++```
++
++- `release` fetches `origin/main` into `/opt/openre.stream/repo`, checks it out to
++ `/opt/openre.stream/releases/<sha>`, runs `npm ci --omit=dev` and gives the directory to `ubuntu`.
++ Nothing restarts.
++- `api` moves `current` and restarts the API and the coordinator. If the API is not ready within
++ 60 s, it switches `current` back by itself (exit 2). Worker units are never touched.
++- The workers stay on `655b98a10aaa` until A3, so `pf --only service` reports that they run another
++ release. That is expected here.
++- **nginx:** `6dc78a5` changes only `deploy/nginx/openre.stream.conf`, and that file is not installed.
++ `/etc/nginx/sites-available/openre.stream.conf` is still the OpenVibe.Sites placeholder (launch
++ rule, README). The release carries the fix, and no nginx action is needed. See B0 for when the
++ vhost gets installed.
++- **Undo:** `sudo /opt/openre.stream/releases/655b98a10aaa/deploy/scripts/deploy.sh api 655b98a10aaa`.
++ Do not run `deploy.sh prune` until the cutover is done, so the old release stays on disk.
++
++### A3. Public RTMP bind (AGENT)
++
++The bind is `OPENRE_RTMP_BIND` in `/etc/openvibe/openre.env` (`server/config.js`, `rtmp.bindHost`).
++A worker reads it only when it starts. **A worker generation is started per release**
++(`openre-rtmp-ingest@<sha>`), so the new release's worker generation is the bind switch:
++
++```bash
++sudo cp -p /etc/openvibe/openre.env /etc/openvibe/openre.env.pre-bind
++sudo sed -i 's/^OPENRE_RTMP_BIND=127\.0\.0\.1$/OPENRE_RTMP_BIND=0.0.0.0/' /etc/openvibe/openre.env
++sudo grep -c '^OPENRE_RTMP_BIND=0\.0\.0\.0$' /etc/openvibe/openre.env # 1
++sudo $DEPLOY workers $SHA # new generation of both workers from $SHA; the old ones drain and exit
++sudo $DEPLOY status
++pf --only service,env,bind # ready generation on $SHA; 0.0.0.0:1936 listening
++```
++
++- The new `openre-rtmp-ingest@$SHA` binds `0.0.0.0:1936` with `SO_REUSEPORT` next to the old
++ generation's `127.0.0.1:1936`. The coordinator then drains the old generation within a few seconds. It
++ closes its listener and exits 0 once idle, and nobody streams to it today. The old restream
++ generation does the same. `test/bind-switch.test.js` proves this sequence with real processes.
++- Until A4 the provider edge still drops 1936, so the public bind is not reachable yet.
++- Never `systemctl restart` a worker instance to pick up an env change while someone is live (it
++ drains, waits up to 30 minutes, then kills).
++- **Undo, before any slot is switched:**
++ `sudo cp -p /etc/openvibe/openre.env.pre-bind /etc/openvibe/openre.env`, then start a loopback
++ generation. Use `sudo /opt/openre.stream/releases/655b98a10aaa/deploy/scripts/deploy.sh workers 655b98a10aaa`
++ for the old code, or, with no session open (`deploy.sh status`),
++ `sudo systemctl restart openre-rtmp-ingest@$SHA.service` for the same code.
++ **Undo in an emergency, with slots switched:** ask the owner to close 1936/tcp at the provider
++ edge (A4). No process restarts. To block it on the host at once:
++ `sudo iptables -I INPUT -p tcp --dport 1936 ! -i lo -j DROP`. Check `sudo iptables -S INPUT` first.
++ Remove the rule with the same line using `-D` instead of `-I`. Never enable ufw for this: it has
++ no rule for SSH on this host.
++
++### A4. Open 1936/tcp at the provider edge (OWNER)
++
++Open **1936/tcp inbound** to `15.204.79.215` in the provider's network firewall, the same place that
++already allows 1935. The host firewall (ufw) is inactive and needs nothing. The DNS record
++`ingest.openre.stream → 15.204.79.215` (DNS only, grey cloud) is already in place.
++
++Check from any machine that is not the host (the agent's workstation works):
++
++```bash
++node scripts/cutover-preflight.js --only dns,port --host-ip 15.204.79.215
++# PASS dns ingest.openre.stream → 15.204.79.215 (this host)
++# PASS port ingest.openre.stream:1936 (15.204.79.215) answers the RTMP handshake from outside the host
++```
++
++Without a checkout, `nc -vz -w 5 ingest.openre.stream 1936` should say "succeeded".
++A **timeout** means the edge still filters. **Connection refused** means the edge is open but nothing
++listens publicly (A3). `--probe-url 'https://<a probe you trust>?host={host}&port={port}'` uses an
++external probe that answers JSON `{ "open": true|false }`. On the host itself the port check is
++always `MANUAL`, because a connection to its own address never crosses the edge.
++
++**Undo:** close the rule at the edge.
++
++### A5. Live settings and the Events subscription (AGENT)
++
++Live reads `OPENRE_URL`, `OPENRE_PUBLIC_URL` and `OPENRE_EVENTS_SECRET` (`server/openre/openre-client.js`,
++`server/openre/mirror.js`, `server/openre/routes.js`). The integration is on when `OPENRE_URL` and
++`OV_OAUTH_CLIENT_SECRET` are set. With no slot switched this changes nothing
++(`test/openre-switch.test.js` in Live).
++
++```bash
++sudo grep -c '^OPENRE_' /etc/openvibe/live.env # 0: none set yet
++sudo cp -p /etc/openvibe/live.env /etc/openvibe/live.env.pre-openre
++S=$(openssl rand -hex 32)
++printf 'OPENRE_URL=http://127.0.0.1:4500\nOPENRE_PUBLIC_URL=https://openre.stream\nOPENRE_EVENTS_SECRET=%s\n' "$S" | sudo tee -a /etc/openvibe/live.env >/dev/null
++unset S
++# the subscription, as Live (consumer "live"), signed with that secret
++sudo node /opt/openre.stream/current/scripts/subscribe-live-events.js --dry-run
++sudo node /opt/openre.stream/current/scripts/subscribe-live-events.js
++# Live picks the env up at start
++curl -s http://127.0.0.1:3000/api/streams | head -c 300; echo # who is live
++sudo DRY_RUN=1 /opt/openvibe.live/deploy/scripts/deploy.sh --restart # the plan: it also deploys origin/main if that moved
++sudo /opt/openvibe.live/deploy/scripts/deploy.sh --restart --wait-idle
++pf --only live-env,events
++```
++
++- The subscription endpoint is `http://127.0.0.1:3000/internal/openre-events`. Events allows
++ `127.0.0.1` endpoints and signs every delivery with signature v2, and Live requires v2. The new
++ subscription gets no history, so the 2026-09-23 rehearsal session is not mirrored.
++- Until Live restarts, deliveries would get 503 and be retried. No slot is switched, so none are sent.
++- A Live restart drops RTMP, WHIP and WebSocket sessions on Live, and they reconnect. `--wait-idle`
++ holds the restart until nobody is live (at most 8 h).
++- OWNER check: `await api('/admin/openre/status')` returns `enabled: true`,
++ `openre_url: "http://127.0.0.1:4500"`, `events_secret_set: true` and `slots: []`.
++- **Undo:** `sudo node /opt/openre.stream/current/scripts/subscribe-live-events.js --disable`, then
++ `sudo cp -p /etc/openvibe/live.env.pre-openre /etc/openvibe/live.env` and
++ `sudo /opt/openvibe.live/deploy/scripts/deploy.sh --restart --wait-idle`. Without `OPENRE_URL`,
++ every slot is ingested by Live again, including switched ones. This is the emergency rollback for
++ all slots at once.
++
++### A6. Full preflight (AGENT)
++
++```bash
++pf # everything PASS; "port" is MANUAL on the host
++```
++
++Then repeat the off-host port check from A4.
++
++### A7. Rehearsal with a test slot and a real restream destination (OWNER + AGENT)
++
++Use a **dedicated test account**, not the owner's own. Once any of a user's slots is on OpenRe, Live
++refuses that user's personal RTMP key (`refusesLiveIngest`).
++
++1. **OWNER:** sign in to Live once with the test account, so it gets a canonical subject. Create an
++ RTMP slot on the Go Live page (for example "openre-rehearsal"). In that slot's restream settings,
++ add **one real destination** with plain `rtmp://`, for example a private or unlisted YouTube
++ stream: `rtmp://a.rtmp.youtube.com/live2` and its key. Add it before step 2 so that the migration
++ imports it, key sealed. After the switch, Live refuses restream edits for the slot (409), and the
++ openre.stream UI is not public yet (B0). Avoid Twitch for the rehearsal: Twitch is rewritten to
++ `rtmps://`, and this host has no OpenSSL ffmpeg (`OPENRE_FFMPEG_OPENSSL_PATH` is unset). Tell the
++ agent the slot id.
++2. **AGENT:** snapshot, then migrate that slot (commands in B2), then `pf --only slot --slot <id>`.
++3. **OWNER:** switch it with `await api('/admin/openre/managed/<id>/ingest-authority', { method: 'PUT', body: { authority: 'openre' } })`.
++ On the Go Live page, press **Regenerate stream key**. The OpenRe key is shown once. In OBS, set
++ the server to `rtmp://ingest.openre.stream:1936/live` and paste the key. Start streaming.
++4. **Check (AGENT and OWNER):**
++ - The channel page on openvibe.live shows the stream, and the player works (Live proxies
++ `flv.internal_url`).
++ - `await api('/admin/openre/status')` lists the session under `live_sessions`.
++ - Sessions and outputs:
++ `sudo sqlite3 -readonly /var/lib/openre/openre.db "SELECT id, state, worker_generation FROM ingest_sessions ORDER BY created_at DESC LIMIT 1; SELECT o.state, o.last_error FROM outputs o ORDER BY o.created_at DESC LIMIT 3; SELECT state, media_vod_id, last_error FROM recordings ORDER BY created_at DESC LIMIT 1;"`
++ should show a `live` session, a `live` output and a `recording` recording.
++ - The destination (YouTube) shows the stream.
++ - Deliveries:
++ `sudo sqlite3 -readonly /var/lib/openvibe-events/events.db "SELECT d.status, count(*) FROM deliveries d JOIN subscriptions s ON s.id = d.subscription_id WHERE s.consumer = 'live' AND s.topic_pattern = 'openre.session.*' GROUP BY 1"`
++ should all be `delivered`.
++5. **During the broadcast (AGENT):**
++ - `sudo $DEPLOY api $SHA`: the stream, the restream and the recording do not drop. OpenRe
++ playback viewers reconnect.
++ - `sudo /opt/openvibe.live/deploy/scripts/deploy.sh --restart` (no `--wait-idle`: the rehearsal
++ itself is live on Live). First check `/api/streams` for anyone else live. The OpenRe session,
++ the restream and the recording keep running. After the restart, Live's 30 s reconcile keeps
++ the channel row live.
++ - `deploy.sh workers` exercises a generation rollover only with a **newer** release (a new
++ instance name). With the same `$SHA` it does nothing. The rollover is covered by
++ `test/rtmp-e2e.test.js`. Repeat it here once a newer release exists.
++6. **OWNER:** stop the stream. The session becomes `ended`, and the Live stream row ends (mirror).
++ The VOD appears in the channel's gallery once Media finalizes it.
++7. **Leave the rehearsal slot switched** as a canary, or roll it back with B-rollback.
++
++## Phase B: per slot
++
++### B0. Before the first broadcaster's slot (OWNER decision)
++
++- **Where broadcasters manage restream destinations after the switch.** Live answers 409 with
++ `manage_url` (https://openre.stream/streams/<id>) to any restream edit for a switched slot, and
++ openre.stream still serves the Sites placeholder. Choose one:
++ - (a) Launch the openre.stream UI: install `deploy/nginx/openre.stream.conf` from the release to
++ `/etc/nginx/sites-available/openre.stream.conf`, run `nginx -t`, reload, and remove the entry
++ from OpenVibe.Sites. The certificate and the OAuth client `openre` with redirect
++ `https://openre.stream/auth/callback` must exist. This touches the README launch rule.
++ - (b) Tell broadcasters that destinations are frozen at the cutover until (a).
++- **Hazard H4, the leaked keys.** 97 of 97 Live stream keys are identical to the pre-leak backup.
++ Each slot's cutover rotates that slot's Live key (at the switch) and issues a new OpenRe key that
++ only the broadcaster sees. It does not rotate the broadcaster's personal key (`users.stream_key`),
++ so step B5 does that. For slots that will not be cut over within days, rotate on Live now: the
++ broadcaster presses Regenerate on the Go Live page for each slot, and Regenerate on the dashboard
++ for the personal key. No admin bulk rotation exists.
++
++### B1. Window (OWNER)
++
++Agree on a maintenance window with the broadcaster. The slot must be offline, because the switch
++refuses a slot that is live on Live. The broadcaster must be reachable and able to change OBS
++during the window.
++
++### B2. Migrate the slot (AGENT)
++
++```bash
++ID=<Live managed stream id>
++pf --only service,db,live-env,events,slot --slot $ID # slot offline, owner has a subject, RTMP slot
++SNAP=/var/lib/openre/live-snapshot-$(date -u +%Y%m%d-%H%M%S).db
++sudo sqlite3 -readonly /opt/openvibe.live/data/live.db ".backup $SNAP"
++sudo sqlite3 "$SNAP" "PRAGMA journal_mode=DELETE;" # a .backup copy is WAL; the migration opens it read-only as ubuntu
++sudo chown ubuntu:ubuntu "$SNAP" && sudo chmod 600 "$SNAP" # it holds every Live user and stream key
++openre_as_service "node scripts/migrate-from-live.js --live-db $SNAP --slots $ID" # dry run
++openre_as_service "node scripts/migrate-from-live.js --live-db $SNAP --slots $ID --apply" # import + checklist
++sudo rm -f "$SNAP"
++pf --only slot --slot $ID # OpenRe <id> (active, mirrored); N destination(s), held ones counted
++```
++
++- The migration imports the slot as a definition with a **new key that nobody has seen**. Old keys
++ are never imported. It also imports the slot's restream destinations, keys sealed. Destinations
++ on private hosts, and the unbound destinations of a user with several slots, are **held**
++ (disabled, with the reason recorded in `migration_map`). It is idempotent.
++- The printed checklist (`--checklist <file>` writes it to a file) is the broadcaster-facing list
++ for B3 to B6.
++- **Undo:** nothing to undo on Live. The OpenRe definition does nothing until the switch. To be
++ tidy, `sudo node /opt/openre.stream/current/scripts/set-definition-state.js --slot $ID --state disabled`.
++
++### B3. Switch (OWNER, admin session)
++
++```js
++await api('/admin/openre/managed/<ID>/ingest-authority', { method: 'PUT', body: { authority: 'openre' } })
++// → { ingest_authority: 'openre', openre_stream_id, rtmp_url: 'rtmp://ingest.openre.stream:1936/live', live_key_rotated: true }
++```
++
++In one transaction, Live points the slot at the OpenRe definition that B2 created and rotates its
++own copy of the slot key. From this moment Live refuses the old key. It also refuses the owner's
++personal key for RTMP, because the owner now has a slot on OpenRe. A 409 means one of: the slot is
++live on Live, it is not an RTMP slot, or the owner has no subject. Fix the cause. Do not use `force`
++for a non-RTMP slot.
++
++### B4. New key (broadcaster)
++
++On the Go Live page for that slot, press **Regenerate stream key**. OpenRe rotates the key and shows
++it once. In OBS, set Server to `rtmp://ingest.openre.stream:1936/live` and Stream Key to that key.
++
++### B5. Personal key (broadcaster; H4)
++
++On the dashboard, press Regenerate for the personal stream key (`POST /api/auth/stream-key/regenerate`).
++`users.stream_key` is not migrated, and a leaked copy must stop working too.
++
++### B6. Test broadcast (broadcaster, AGENT, OWNER)
++
++The broadcaster goes live. Check the same things as A7 step 4. `sudo $DEPLOY status` shows the
++session on the newest generation, and `await api('/admin/openre/status')` lists it. In the OpenRe DB,
++the outputs are `live` and any held destinations are not running.
++
++### B-rollback (per slot)
++
++1. The broadcaster stops streaming. The slot must be offline.
++2. **OWNER:** `await api('/admin/openre/managed/<ID>/ingest-authority', { method: 'PUT', body: { authority: 'live' } })`.
++3. **AGENT:** `sudo node /opt/openre.stream/current/scripts/set-definition-state.js --slot <ID> --state disabled`.
++ OpenRe then refuses the key the broadcaster got in B4, so the slot's destinations can never be
++ pushed twice.
++4. **Broadcaster:** press Regenerate on the Go Live page. The Live key was rotated at the switch.
++ Point OBS back at the Live RTMP URL shown there.
++
++To switch the slot to OpenRe again later, run `set-definition-state.js --slot <ID> --state active`
++first, then B3 to B6.
++
++**All slots at once (emergency):** undo A5. Remove `OPENRE_URL`, or restore `live.env.pre-openre`,
++and restart Live. Every switched slot then behaves as a Live slot. Its Live key was rotated at the
++switch, so each broadcaster regenerates on the Go Live page.
++
++## Phase C: bookkeeping (AGENT, after A2)
++
++- **ovhost inventory.** Copy the `openre` entry of OpenVibe.Host's `host.example.json` into
++ `/etc/openvibe/host.json`: `workerUnits`, and the drill block with `OPENRE_DRILL`. Then run
++ `sudo ovhost validate openre` and `sudo ovhost status openre`. The status lists the worker
++ instances and never restarts them.
++- **First restore drill:** `sudo ovhost drill openre`. It needs a release with `OPENRE_DRILL`, so
++ run it after A2. Only openre-api runs, with `OPENRE_DRILL=1`, on 127.0.0.1:14500, from a restored
++ copy of the database. The coordinator and the workers refuse to start in a drill. The event relay
++ and Media calls are off. Writes, `/play/` and sign-in answer 503.
++- Update `STATUS.json` and README "Status" after each phase.
++- Each slot's Live RTMP key stays in Live's database until Live's RTMP ingest is retired. That
++ happens after the last slot has run on OpenRe for two weeks (ADR-009, register C-40). Port 1935
++ moves to OpenRe then (`OPENRE_RTMP_EXTRA_PORTS=1935`, `deploy.sh workers`).
++
++## Order of actions
++
++| # | Who | Action | Check |
++|---|---|---|---|
++| A1 | OWNER | push OpenRe `main` (and OpenVibe.Host) | `git log origin/main` |
++| A2 | AGENT | `deploy.sh release origin/main`, `deploy.sh api <sha>` | `pf --only release,env,db`, `/api/ready` |
++| A3 | AGENT | `OPENRE_RTMP_BIND=0.0.0.0`, `deploy.sh workers <sha>` | `pf --only service,env,bind` |
++| A4 | **OWNER** | open 1936/tcp at the provider edge | off-host `--only port` PASS |
++| A5 | AGENT | Live env (3 names), `subscribe-live-events.js`, Live `--restart --wait-idle` | `pf --only live-env,events`; OWNER `/admin/openre/status` |
++| A6 | AGENT | full `pf` | all PASS, port MANUAL on host + PASS off host |
++| A7 | **OWNER** + AGENT | rehearsal: test account, slot, real destination; migrate; switch; OBS; deploys during the broadcast | A7 step 4 |
++| C | AGENT | ovhost inventory + first `ovhost drill openre` | drill passed |
++| B0 | **OWNER** | decide on the openre.stream UI; H4 fallback rotations | — |
++| B1–B6 | **OWNER** + broadcaster + AGENT | per slot: window, migrate, switch, regenerate, personal key, test | `pf --only slot --slot <id>`, B6 |
ddiff --git a/scripts/cutover-preflight.js b/scripts/cutover-preflight.js
nnew file mode 100644
iindex 0000000..77ef70c
---- /dev/null
++++ b/scripts/cutover-preflight.js
@@@ -0,0 +1,484 @@
++#!/usr/bin/env node
++'use strict';
++/**
++ * Read-only pre-flight for the RTMP cutover (docs/cutover.md). Run on the host as root:
++ *
++ * sudo node /opt/openre.stream/current/scripts/cutover-preflight.js [--only a,b] [--slot <id>] [--json]
++ *
++ * Checks (in this order; --only / --skip pick some):
++ * release `current` points at the expected release (--expect-release, else origin/main of
++ * /opt/openre.stream/repo as last fetched) and that release contains --require-commit
++ * (default 6dc78a5, the nginx realip fix)
++ * service openre-api /api/ready: ready, not a drill, one ready rtmp-ingest and restream worker
++ * on the current release, a valid coordinator lease, the event relay configured
++ * env /etc/openvibe/openre.env: OPENRE_DRILL unset, the RTMP port and public host
++ * bind OPENRE_RTMP_BIND is public and the RTMP port really listens on a public address
++ * dns the ingest host resolves (A/AAAA) only to this host's addresses
++ * port the ingest port answers an RTMP handshake from outside (see "port" below)
++ * db PRAGMA integrity_check and foreign_key_check of openre.db (opened read-only)
++ * live-env /etc/openvibe/live.env has OPENRE_URL, OPENRE_EVENTS_SECRET (32+ chars) and
++ * OV_OAUTH_CLIENT_SECRET, and the running Live process was started with them
++ * events Events has Live's openre.session.* subscription to /internal/openre-events, enabled,
++ * with the secret Live holds (compared by hash, never printed), and no dead deliveries
++ * slot only with --slot <id>: the Live slot and its OpenRe definition, for the per-slot step
++ *
++ * "port": from the host itself a connection to its own public address never crosses the provider
++ * edge, so the result is MANUAL there. Prove it from outside: run `--only port` on any machine
++ * that is not the host (a checkout with `npm ci` is enough), or pass --probe-url with a probe you
++ * trust (GET, `{host}` and `{port}` replaced, JSON answer with a boolean `open` or `reachable`).
++ *
++ * Nothing is written anywhere: files and databases are opened read-only, no secret value is
++ * printed, the only network traffic is DNS, one GET to /api/ready and one TCP connection with an
++ * RTMP handshake (no publish). Exit 0 when no check FAILs (--strict: also no WARN/MANUAL).
++ */
++const crypto = require('crypto');
++const dnsPromises = require('dns').promises;
++const fs = require('fs');
++const net = require('net');
++const os = require('os');
++const path = require('path');
++const { execFile } = require('child_process');
++
++const DEFAULTS = Object.freeze({
++ root: '/opt/openre.stream',
++ openreEnv: '/etc/openvibe/openre.env',
++ liveEnv: '/etc/openvibe/live.env',
++ openreDb: '/var/lib/openre/openre.db',
++ eventsDb: '/var/lib/openvibe-events/events.db',
++ liveDb: '/opt/openvibe.live/data/live.db',
++ api: 'http://127.0.0.1:4500',
++ openreUrl: 'http://127.0.0.1:4500',
++ liveUnit: 'openvibe-live.service',
++ endpoint: 'http://127.0.0.1:3000/internal/openre-events',
++ topic: 'openre.session.*',
++ requireCommit: '6dc78a5',
++ ingestHost: 'ingest.openre.stream',
++ port: 1936,
++ timeoutMs: 5000,
++});
++const CHECKS = Object.freeze(['release', 'service', 'env', 'bind', 'dns', 'port', 'db', 'live-env', 'events', 'slot']);
++const WORKER_KINDS = Object.freeze(['rtmp-ingest', 'restream']);
++
++// ── helpers ─────────────────────────────────────────────────────────────────
++
++/** KEY=VALUE lines (systemd EnvironmentFile / dotenv); null when the file cannot be read. */
++function parseEnvFile(text) {
++ if (text == null) return null;
++ return require('dotenv').parse(String(text));
++}
++
++const isLoopback = (a) => /^127\./.test(a) || a === '::1' || a === 'localhost' || /^::ffff:127\./.test(a);
++const isWildcard = (a) => a === '0.0.0.0' || a === '::' || a === '*' || a === '[::]';
++const truthy = (v) => /^(1|true|yes|on)$/i.test(String(v || ''));
++const sha256 = (s) => crypto.createHash('sha256').update(String(s)).digest();
++const sameSecret = (a, b) => Boolean(a && b) && crypto.timingSafeEqual(sha256(a), sha256(b));
++
++/** Local addresses of the listeners on a TCP port, from `ss -Hltn`. */
++function parseSsListeners(stdout, port) {
++ const out = [];
++ for (const line of String(stdout || '').split('\n')) {
++ const cols = line.trim().split(/\s+/);
++ const local = cols.find((c) => new RegExp(`:${port}$`).test(c));
++ if (!local) continue;
++ const addr = local.slice(0, local.lastIndexOf(':')).replace(/^\[|\]$/g, '').replace(/%.*$/, '');
++ out.push(addr);
++ }
++ return out;
++}
++
++/** Non-internal addresses of this machine (the host's public addresses on openvibe-ovh). */
++function localAddresses(interfaces = os.networkInterfaces()) {
++ const out = [];
++ for (const list of Object.values(interfaces || {})) for (const i of list || []) if (!i.internal) out.push(i.address);
++ return out;
++}
++
++/** TCP connect + RTMP C0/C1; resolves { ok, rtmp, code, error, ms }. Sends no publish. */
++function rtmpProbe(host, port, timeoutMs = DEFAULTS.timeoutMs) {
++ return new Promise((resolve) => {
++ const started = Date.now();
++ let connected = false;
++ const socket = net.connect({ host, port });
++ const done = (r) => { socket.destroy(); resolve({ ms: Date.now() - started, ...r }); };
++ socket.setTimeout(timeoutMs, () => done(connected
++ ? { ok: true, rtmp: false, error: 'connected, but no RTMP handshake answer' }
++ : { ok: false, code: 'ETIMEDOUT', error: `no answer within ${timeoutMs / 1000}s` }));
++ socket.once('connect', () => {
++ connected = true;
++ const c1 = Buffer.alloc(1536);
++ crypto.randomFillSync(c1, 8);
++ socket.write(Buffer.concat([Buffer.from([0x03]), c1]));
++ });
++ socket.once('data', (d) => done({ ok: true, rtmp: d[0] === 0x03 }));
++ socket.once('error', (err) => done({ ok: false, code: err.code, error: err.message }));
++ });
++}
++
++function runCmd(cmd, args) {
++ return new Promise((resolve) => {
++ execFile(cmd, args, { timeout: 15000, maxBuffer: 1 << 20 }, (err, stdout, stderr) => {
++ resolve({ code: err ? (typeof err.code === 'number' ? err.code : 1) : 0, stdout: String(stdout || ''), stderr: String(stderr || '') });
++ });
++ });
++}
++
++function realDeps() {
++ return {
++ readFile: (p) => { try { return fs.readFileSync(p, 'utf8'); } catch { return null; } },
++ readBinary: (p) => { try { return fs.readFileSync(p); } catch { return null; } },
++ readlink: (p) => { try { return fs.readlinkSync(p); } catch { return null; } },
++ exists: (p) => fs.existsSync(p),
++ run: runCmd,
++ resolve4: (h) => dnsPromises.resolve4(h).catch((e) => (e.code === 'ENODATA' || e.code === 'ENOTFOUND' ? [] : Promise.reject(e))),
++ resolve6: (h) => dnsPromises.resolve6(h).catch((e) => (e.code === 'ENODATA' || e.code === 'ENOTFOUND' ? [] : Promise.reject(e))),
++ interfaces: () => os.networkInterfaces(),
++ probe: rtmpProbe,
++ fetch: (url, opts) => fetch(url, opts),
++ openDb: (p) => {
++ const Database = require('better-sqlite3');
++ return new Database(p, { readonly: true, fileMustExist: true });
++ },
++ };
++}
++
++function withDb(deps, file, fn) {
++ let db;
++ try { db = deps.openDb(file); } catch (err) { return { error: `cannot open ${file} read-only: ${err.message}` }; }
++ try { return fn(db); } finally { try { db.close(); } catch { /* read-only */ } }
++}
++
++// ── checks ──────────────────────────────────────────────────────────────────
++
++async function checkRelease(ctx) {
++ const { deps, opts } = ctx;
++ const target = deps.readlink(path.join(opts.root, 'current'));
++ if (!target) return ['FAIL', `${opts.root}/current is not a symlink`];
++ const current = path.basename(target);
++ ctx.release = current;
++ const repo = path.join(opts.root, 'repo');
++ let expected = opts.expectRelease;
++ let source = '--expect-release';
++ if (!expected) {
++ const r = await deps.run('git', ['-C', repo, 'rev-parse', '--short=12', 'origin/main']);
++ if (r.code !== 0) return ['FAIL', `current = ${current}; cannot read origin/main in ${repo}: ${(r.stderr || '').trim().split('\n')[0]} (run as root, or pass --expect-release)`];
++ expected = r.stdout.trim();
++ source = 'origin/main as last fetched';
++ }
++ const problems = [];
++ if (!current.startsWith(expected) && !expected.startsWith(current)) problems.push(`current = ${current}, expected ${expected} (${source}): deploy it (docs/cutover.md A2)`);
++ if (opts.requireCommit) {
++ const r = await deps.run('git', ['-C', repo, 'merge-base', '--is-ancestor', opts.requireCommit, current]);
++ if (r.code === 1) problems.push(`release ${current} does not contain ${opts.requireCommit}`);
++ else if (/not a valid (object|commit)/i.test(r.stderr || '')) problems.push(`${opts.requireCommit} is not in ${repo} yet (deploy.sh release fetches it), so ${current} cannot contain it`);
++ else if (r.code !== 0) problems.push(`cannot tell whether ${current} contains ${opts.requireCommit}: ${(r.stderr || '').trim().split('\n')[0]}`);
++ }
++ if (problems.length) return ['FAIL', problems.join('; ')];
++ return ['PASS', `current = ${current} (${source})${opts.requireCommit ? `, contains ${opts.requireCommit}` : ''}`];
++}
++
++async function checkService(ctx) {
++ const { deps, opts } = ctx;
++ let res;
++ try {
++ res = await deps.fetch(`${opts.api}/api/ready`, { signal: AbortSignal.timeout(opts.timeoutMs) });
++ } catch (err) {
++ return ['FAIL', `${opts.api}/api/ready unreachable: ${err.message}`];
++ }
++ let body = null;
++ try { body = await res.json(); } catch { body = null; }
++ if (!body) return ['FAIL', `/api/ready answered ${res.status} without JSON`];
++ const problems = [];
++ if (res.status !== 200 || body.status !== 'ready') problems.push(`status ${res.status} ${body.status} (${JSON.stringify(body.checks || {})})`);
++ if (body.mode === 'drill') problems.push('the API runs in drill mode (OPENRE_DRILL)');
++ const workers = Array.isArray(body.workers) ? body.workers : [];
++ const rows = ctx.workerRows || workerRowsFromDb(ctx);
++ for (const kind of WORKER_KINDS) {
++ const ready = workers.filter((w) => w.kind === kind && w.state === 'ready');
++ if (ready.length !== 1) { problems.push(`${ready.length} ready ${kind} workers (want 1)`); continue; }
++ const row = rows && rows.find((w) => w.kind === kind && w.generation === ready[0].generation);
++ if (ctx.release && row && row.release && row.release !== ctx.release) problems.push(`${kind}#${ready[0].generation} runs release ${row.release}, current is ${ctx.release}: deploy.sh workers`);
++ }
++ const draining = workers.filter((w) => w.state === 'draining').map((w) => `${w.kind}#${w.generation}`);
++ if (!body.coordinator || !body.coordinator.lease_valid) problems.push('no valid coordinator lease');
++ if (!body.events || !body.events.configured) problems.push('event relay not configured (EVENTS_URL, OV_OAUTH_CLIENT_SECRET)');
++ if (problems.length) return ['FAIL', problems.join('; ')];
++ const summary = WORKER_KINDS.map((k) => { const w = workers.find((x) => x.kind === k && x.state === 'ready'); return `${k}#${w.generation}`; }).join(', ');
++ const pending = body.events && body.events.pending ? `, ${body.events.pending} events waiting in the outbox` : '';
++ if (draining.length) return ['WARN', `ready: ${summary}; still draining: ${draining.join(', ')}${pending}`];
++ return ['PASS', `ready: ${summary}, coordinator lease valid, event relay on${pending}`];
++}
++
++function workerRowsFromDb(ctx) {
++ const r = withDb(ctx.deps, ctx.opts.openreDb, (db) => db.prepare("SELECT kind, generation, release, state FROM workers WHERE state IN ('starting', 'ready', 'draining')").all());
++ ctx.workerRows = Array.isArray(r) ? r : null;
++ return ctx.workerRows;
++}
++
++function openreEnv(ctx) {
++ if (ctx.openreEnvValues === undefined) ctx.openreEnvValues = parseEnvFile(ctx.deps.readFile(ctx.opts.openreEnv));
++ return ctx.openreEnvValues;
++}
++
++function liveEnv(ctx) {
++ if (ctx.liveEnvValues === undefined) ctx.liveEnvValues = parseEnvFile(ctx.deps.readFile(ctx.opts.liveEnv));
++ return ctx.liveEnvValues;
++}
++
++async function checkEnv(ctx) {
++ const env = openreEnv(ctx);
++ if (!env) return ['FAIL', `cannot read ${ctx.opts.openreEnv} (run as root)`];
++ const problems = [];
++ if (truthy(env.OPENRE_DRILL)) problems.push('OPENRE_DRILL is set: the coordinator and every worker refuse to start');
++ const port = Number(env.OPENRE_RTMP_PORT || 1936);
++ if (port !== ctx.opts.port) problems.push(`OPENRE_RTMP_PORT = ${port}, expected ${ctx.opts.port}`);
++ const host = env.OPENRE_RTMP_PUBLIC_HOST || 'ingest.openre.stream';
++ if (host !== ctx.opts.ingestHost) problems.push(`OPENRE_RTMP_PUBLIC_HOST = ${host}, expected ${ctx.opts.ingestHost}`);
++ const missing = ['OV_OAUTH_CLIENT_SECRET', 'OPENRE_SECRETS_KEY', 'EVENTS_URL'].filter((n) => !env[n]);
++ if (missing.length) problems.push(`not set: ${missing.join(', ')}`);
++ if (problems.length) return ['FAIL', problems.join('; ')];
++ return ['PASS', `OPENRE_DRILL unset, RTMP port ${port}, public host ${host}, secrets and EVENTS_URL set`];
++}
++
++async function checkBind(ctx) {
++ const { deps, opts } = ctx;
++ const env = openreEnv(ctx);
++ if (!env) return ['FAIL', `cannot read ${opts.openreEnv} (run as root)`];
++ const bind = env.OPENRE_RTMP_BIND || '0.0.0.0';
++ const ss = await deps.run('ss', ['-Hltn', `sport = :${opts.port}`]);
++ if (ss.code !== 0) return ['FAIL', `ss failed: ${ss.stderr.trim()}`];
++ const listening = parseSsListeners(ss.stdout, opts.port);
++ const publicListeners = listening.filter((a) => !isLoopback(a));
++ const envPublic = !isLoopback(bind);
++ const detail = `OPENRE_RTMP_BIND=${bind}; listening on ${listening.length ? listening.map((a) => `${a}:${opts.port}`).join(', ') : 'nothing'}`;
++ if (!envPublic) return ['FAIL', `${detail}: set OPENRE_RTMP_BIND=0.0.0.0 and start a worker generation (docs/cutover.md A3)`];
++ if (!publicListeners.length) return ['FAIL', `${detail}: the running generation still has the old bind; start a new one with deploy.sh workers (docs/cutover.md A3)`];
++ if (!isWildcard(bind) && !publicListeners.includes(bind)) return ['WARN', `${detail}: the bind is a specific address`];
++ return ['PASS', detail];
++}
++
++async function checkDns(ctx) {
++ const { deps, opts } = ctx;
++ let a4 = [];
++ let a6 = [];
++ try { [a4, a6] = await Promise.all([deps.resolve4(opts.ingestHost), deps.resolve6(opts.ingestHost)]); } catch (err) {
++ return ['FAIL', `cannot resolve ${opts.ingestHost}: ${err.code || err.message}`];
++ }
++ const all = [...a4, ...a6];
++ if (!all.length) return ['FAIL', `${opts.ingestHost} has no A or AAAA record`];
++ const mine = opts.hostIps.length ? opts.hostIps : localAddresses(deps.interfaces());
++ ctx.onHost = all.some((a) => localAddresses(deps.interfaces()).includes(a));
++ const foreign = all.filter((a) => !mine.includes(a));
++ const detail = `${opts.ingestHost} → ${all.join(', ')}`;
++ if (foreign.length) {
++ const hint = opts.hostIps.length ? `the host is ${opts.hostIps.join(', ')}` : 'not an address of this machine (run on the host, or pass --host-ip <the host\'s address>)';
++ return ['FAIL', `${detail}; ${foreign.join(', ')}: ${hint}. A Cloudflare-proxied (orange) record cannot carry RTMP`];
++ }
++ const bind = (openreEnv(ctx) || {}).OPENRE_RTMP_BIND || '0.0.0.0';
++ if (a6.length && bind === '0.0.0.0') return ['WARN', `${detail}: AAAA records exist but OPENRE_RTMP_BIND=0.0.0.0 listens on IPv4 only`];
++ return ['PASS', `${detail} (this host)`];
++}
++
++async function checkPort(ctx) {
++ const { deps, opts } = ctx;
++ if (opts.probeUrl) {
++ const url = opts.probeUrl.replace(/\{host\}/g, encodeURIComponent(opts.ingestHost)).replace(/\{port\}/g, String(opts.port));
++ let body;
++ try {
++ const res = await deps.fetch(url, { signal: AbortSignal.timeout(opts.timeoutMs * 3) });
++ body = await res.json();
++ if (!res.ok) return ['MANUAL', `probe answered ${res.status}; check by hand (docs/cutover.md A4)`];
++ } catch (err) {
++ return ['MANUAL', `probe ${new URL(url).host} failed (${err.message}); check by hand (docs/cutover.md A4)`];
++ }
++ const open = body && typeof body.open === 'boolean' ? body.open : body && typeof body.reachable === 'boolean' ? body.reachable : null;
++ if (open === true) return ['PASS', `external probe ${new URL(url).host}: ${opts.ingestHost}:${opts.port} open`];
++ if (open === false) return ['FAIL', `external probe ${new URL(url).host}: ${opts.ingestHost}:${opts.port} not reachable — open ${opts.port}/tcp at the provider edge (owner)`];
++ return ['MANUAL', 'the probe answer has no boolean open/reachable; check by hand (docs/cutover.md A4)'];
++ }
++ let addrs = [];
++ try { addrs = await deps.resolve4(opts.ingestHost); } catch { addrs = []; }
++ const target = addrs[0] || opts.ingestHost;
++ const onHost = addrs.some((a) => localAddresses(deps.interfaces()).includes(a));
++ const r = await deps.probe(target, opts.port, opts.timeoutMs);
++ const what = `${opts.ingestHost}:${opts.port}${addrs[0] ? ` (${addrs[0]})` : ''}`;
++ const answer = r.ok ? (r.rtmp ? 'answers the RTMP handshake' : `accepts TCP (${r.error || 'no RTMP answer'})`) : r.code === 'ECONNREFUSED' ? 'refused (nothing listens there)' : `${r.code || 'error'}: ${r.error}`;
++ if (onHost) {
++ return ['MANUAL', `${what} ${answer} from this host, which never crosses the provider edge. From a machine outside: \`node scripts/cutover-preflight.js --only port\` or \`nc -vz -w 5 ${opts.ingestHost} ${opts.port}\``];
++ }
++ if (r.ok && r.rtmp) return ['PASS', `${what} ${answer} from outside the host (${Math.round(r.ms)} ms)`];
++ if (r.ok) return ['WARN', `${what} ${answer}: something accepts TCP but is not an RTMP server`];
++ if (r.code === 'ECONNREFUSED') return ['FAIL', `${what} ${answer}: the edge lets it through, but no public listener (check "bind" on the host)`];
++ return ['FAIL', `${what} ${answer}: filtered — open ${opts.port}/tcp at the provider edge (owner, docs/cutover.md A4) and in the host firewall if one is active`];
++}
++
++async function checkDb(ctx) {
++ const r = withDb(ctx.deps, ctx.opts.openreDb, (db) => {
++ const integrity = db.prepare('PRAGMA integrity_check').all().map((x) => Object.values(x)[0]).join('; ');
++ const fk = db.prepare('PRAGMA foreign_key_check').all();
++ const counts = {};
++ for (const t of ['stream_definitions', 'ingest_keys', 'ingest_sessions', 'destinations', 'migration_map']) {
++ try { counts[t] = db.prepare(`SELECT count(*) AS n FROM ${t}`).get().n; } catch { counts[t] = null; }
++ }
++ return { integrity, fk: fk.length, counts };
++ });
++ if (r.error) return ['FAIL', r.error];
++ const counts = Object.entries(r.counts).map(([k, v]) => `${k} ${v == null ? '?' : v}`).join(', ');
++ if (r.integrity !== 'ok') return ['FAIL', `integrity_check = ${r.integrity.slice(0, 300)}`];
++ if (r.fk) return ['FAIL', `integrity_check ok, but ${r.fk} foreign key violation(s)`];
++ return ['PASS', `integrity_check ok, no foreign key violations (${counts})`];
++}
++
++async function checkLiveEnv(ctx) {
++ const { deps, opts } = ctx;
++ const env = liveEnv(ctx);
++ if (!env) return ['FAIL', `cannot read ${opts.liveEnv} (run as root)`];
++ const problems = [];
++ if (!env.OPENRE_URL) problems.push('OPENRE_URL not set (the integration is off)');
++ else if (env.OPENRE_URL.replace(/\/+$/, '') !== opts.openreUrl) problems.push(`OPENRE_URL = ${env.OPENRE_URL}, expected ${opts.openreUrl}`);
++ if (!env.OPENRE_EVENTS_SECRET) problems.push('OPENRE_EVENTS_SECRET not set');
++ else if (env.OPENRE_EVENTS_SECRET.length < 32) problems.push('OPENRE_EVENTS_SECRET shorter than 32 characters (Events refuses it)');
++ if (!env.OV_OAUTH_CLIENT_SECRET) problems.push('OV_OAUTH_CLIENT_SECRET not set (Live cannot get OpenRe tokens)');
++ const notes = [];
++ if (!env.OPENRE_PUBLIC_URL) notes.push('OPENRE_PUBLIC_URL unset (default https://openre.stream)');
++ // The running process: systemd loaded the env file when Live started. Names only.
++ const show = await deps.run('systemctl', ['show', opts.liveUnit, '--property=MainPID']);
++ const pid = Number((/MainPID=(\d+)/.exec(show.stdout) || [])[1] || 0);
++ if (!pid) problems.push(`${opts.liveUnit} is not running`);
++ else {
++ const raw = deps.readBinary(`/proc/${pid}/environ`);
++ if (!raw) problems.push(`cannot read /proc/${pid}/environ (run as root)`);
++ else {
++ const running = new Map(String(raw).split('\0').filter(Boolean).map((kv) => [kv.slice(0, kv.indexOf('=')), kv.slice(kv.indexOf('=') + 1)]));
++ const stale = ['OPENRE_URL', 'OPENRE_EVENTS_SECRET', 'OPENRE_PUBLIC_URL'].filter((n) => (env[n] || '') !== (running.get(n) || ''));
++ if (stale.length) problems.push(`Live (pid ${pid}) was started before ${stale.join(', ')} changed in ${opts.liveEnv}: restart Live (deploy.sh --restart --wait-idle)`);
++ }
++ }
++ if (problems.length) return ['FAIL', problems.join('; ')];
++ return ['PASS', `OPENRE_URL=${env.OPENRE_URL}, OPENRE_EVENTS_SECRET set, running Live (pid ${pid}) has them${notes.length ? `; ${notes.join('; ')}` : ''}`];
++}
++
++async function checkEvents(ctx) {
++ const { opts } = ctx;
++ const env = liveEnv(ctx) || {};
++ const r = withDb(ctx.deps, opts.eventsDb, (db) => {
++ const subs = db.prepare("SELECT id, enabled, secret FROM subscriptions WHERE consumer = 'live' AND topic_pattern = ? AND endpoint = ?").all(opts.topic, opts.endpoint);
++ const dead = subs.length ? db.prepare("SELECT count(*) AS n FROM deliveries WHERE subscription_id = ? AND status = 'dead'").get(subs[0].id).n : 0;
++ const pending = subs.length ? db.prepare("SELECT count(*) AS n FROM deliveries WHERE subscription_id = ? AND status IN ('pending', 'failed')").get(subs[0].id).n : 0;
++ return { subs, dead, pending };
++ });
++ if (r.error) return ['FAIL', r.error];
++ if (!r.subs.length) return ['FAIL', `no subscription live ${opts.topic} → ${opts.endpoint}: create it (scripts/subscribe-live-events.js, docs/cutover.md A5)`];
++ if (r.subs.length > 1) return ['FAIL', `${r.subs.length} identical subscriptions (${r.subs.map((s) => s.id).join(', ')}): disable all but one`];
++ const s = r.subs[0];
++ const problems = [];
++ if (!s.enabled) problems.push('disabled');
++ if (!env.OPENRE_EVENTS_SECRET) problems.push(`cannot compare the secret: OPENRE_EVENTS_SECRET not readable in ${opts.liveEnv}`);
++ else if (!sameSecret(s.secret, env.OPENRE_EVENTS_SECRET)) problems.push("its secret is not Live's OPENRE_EVENTS_SECRET (Live answers 401 to every delivery)");
++ if (r.dead) problems.push(`${r.dead} dead deliveries (replay them after fixing: POST /api/v1/deliveries/replay)`);
++ if (problems.length) return ['FAIL', `${s.id}: ${problems.join('; ')}`];
++ return [r.pending ? 'WARN' : 'PASS', `${s.id} enabled, secret matches Live's${r.pending ? `, ${r.pending} deliveries waiting or retrying` : ', nothing waiting'}`];
++}
++
++async function checkSlot(ctx) {
++ const { opts } = ctx;
++ const id = Number(opts.slot);
++ const live = withDb(ctx.deps, opts.liveDb, (db) => {
++ const slot = db.prepare('SELECT id, user_id, slug, title, protocol, streaming_method, ingest_authority, openre_stream_id FROM managed_streams WHERE id = ?').get(id);
++ if (!slot) return { slot: null };
++ const liveNow = db.prepare('SELECT id FROM streams WHERE managed_stream_id = ? AND is_live = 1 LIMIT 1').get(id);
++ const subject = db.prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ?").get(slot.user_id);
++ const user = db.prepare('SELECT username, is_banned FROM users WHERE id = ?').get(slot.user_id);
++ return { slot, liveNow, subject: subject && subject.subject_id, user };
++ });
++ if (live.error) return ['FAIL', live.error];
++ if (!live.slot) return ['FAIL', `Live has no managed stream ${id}`];
++ const s = live.slot;
++ const openre = withDb(ctx.deps, opts.openreDb, (db) => {
++ const defs = db.prepare("SELECT d.id, d.state, d.mirror_to_live FROM external_refs r JOIN stream_definitions d ON d.id = r.definition_id WHERE r.service = 'live' AND r.type = 'managed_stream' AND r.ref_id = ?").all(String(id));
++ const map = db.prepare("SELECT source_type, status, reason FROM migration_map WHERE source_system = 'live' AND ((source_type = 'managed_stream' AND source_id = ?) OR source_type = 'restream_destination')").all(String(id));
++ const dests = defs.length ? db.prepare('SELECT enabled, hold_reason FROM destinations WHERE definition_id = ?').all(defs[0].id) : [];
++ const openSessions = defs.length ? db.prepare("SELECT count(*) AS n FROM ingest_sessions WHERE definition_id = ? AND state IN ('starting', 'live', 'ending')").get(defs[0].id).n : 0;
++ return { defs, slotMap: map.find((m) => m.source_type === 'managed_stream') || null, dests, openSessions };
++ });
++ if (openre.error) return ['FAIL', openre.error];
++ const facts = [`slot ${id} (@${live.user ? live.user.username : '?'}, ${s.protocol}${s.streaming_method ? `/${s.streaming_method}` : ''}) authority ${s.ingest_authority || 'live'}`];
++ const problems = [];
++ if (live.user && live.user.is_banned) problems.push('owner is banned');
++ if (!live.subject) problems.push('owner has no canonical subject (they sign in to Live once)');
++ if (live.liveNow) problems.push(`live on Live right now (stream ${live.liveNow.id}): wait for the window`);
++ if (s.protocol !== 'rtmp' && !['rtmp', 'obs'].includes(String(s.streaming_method || '').toLowerCase())) problems.push('not an RTMP slot (OpenRe carries RTMP only)');
++ if (openre.defs.length > 1) problems.push(`${openre.defs.length} OpenRe definitions reference it`);
++ const def = openre.defs[0];
++ if (def) {
++ facts.push(`OpenRe ${def.id} (${def.state}${def.mirror_to_live ? ', mirrored' : ', NOT mirrored to Live'})`);
++ const held = openre.dests.filter((d) => d.hold_reason).length;
++ facts.push(`${openre.dests.length} destination(s)${held ? `, ${held} held` : ''}`);
++ if (s.openre_stream_id && s.openre_stream_id !== def.id) problems.push(`Live points at ${s.openre_stream_id}, OpenRe has ${def.id}`);
++ if (openre.openSessions) facts.push(`${openre.openSessions} open OpenRe session(s)`);
++ } else facts.push(openre.slotMap ? `migration_map: ${openre.slotMap.status}${openre.slotMap.reason ? ` (${openre.slotMap.reason})` : ''}` : 'not migrated yet');
++ if (problems.length) return ['FAIL', `${facts.join('; ')}: ${problems.join('; ')}`];
++ return ['PASS', facts.join('; ')];
++}
++
++const RUNNERS = { release: checkRelease, service: checkService, env: checkEnv, bind: checkBind, dns: checkDns, port: checkPort, db: checkDb, 'live-env': checkLiveEnv, events: checkEvents, slot: checkSlot };
++
++// ── entry ───────────────────────────────────────────────────────────────────
++
++function parseArgs(argv) {
++ const opts = { ...DEFAULTS, only: null, skip: [], hostIps: [], json: false, strict: false, expectRelease: null, probeUrl: null, slot: null };
++ const take = (i) => { if (argv[i + 1] == null) throw new Error(`${argv[i]} needs a value`); return argv[i + 1]; };
++ const names = { '--root': 'root', '--openre-env': 'openreEnv', '--live-env': 'liveEnv', '--openre-db': 'openreDb', '--events-db': 'eventsDb', '--live-db': 'liveDb', '--api': 'api', '--openre-url': 'openreUrl', '--live-unit': 'liveUnit', '--expect-release': 'expectRelease', '--require-commit': 'requireCommit', '--ingest-host': 'ingestHost', '--probe-url': 'probeUrl', '--slot': 'slot' };
++ for (let i = 0; i < argv.length; i++) {
++ const a = argv[i];
++ if (a === '--json') opts.json = true;
++ else if (a === '--strict') opts.strict = true;
++ else if (a === '--only') { opts.only = take(i).split(',').map((s) => s.trim()).filter(Boolean); i++; }
++ else if (a === '--skip') { opts.skip = take(i).split(',').map((s) => s.trim()).filter(Boolean); i++; }
++ else if (a === '--host-ip') { opts.hostIps.push(...take(i).split(',').map((s) => s.trim()).filter(Boolean)); i++; }
++ else if (a === '--port') { opts.port = Number(take(i)); i++; }
++ else if (a === '--timeout-ms') { opts.timeoutMs = Number(take(i)); i++; }
++ else if (names[a]) { opts[names[a]] = take(i); i++; }
++ else if (a === '-h' || a === '--help') opts.help = true;
++ else throw new Error(`unknown argument ${a}`);
++ }
++ if (opts.requireCommit === '' || opts.requireCommit === 'none') opts.requireCommit = null;
++ opts.openreUrl = String(opts.openreUrl).replace(/\/+$/, '');
++ opts.api = String(opts.api).replace(/\/+$/, '');
++ for (const c of [...(opts.only || []), ...opts.skip]) if (!CHECKS.includes(c)) throw new Error(`unknown check ${c} (${CHECKS.join(', ')})`);
++ if (opts.slot != null && !/^\d+$/.test(String(opts.slot))) throw new Error('--slot takes a Live managed stream id');
++ if (!Number.isInteger(opts.port) || opts.port <= 0) throw new Error('--port must be a TCP port');
++ return opts;
++}
++
++async function preflight(opts, deps = realDeps()) {
++ const ctx = { opts, deps };
++ const wanted = (opts.only || CHECKS.filter((c) => c !== 'slot' || opts.slot != null)).filter((c) => !opts.skip.includes(c));
++ const results = [];
++ for (const id of wanted) {
++ if (id === 'slot' && opts.slot == null) { results.push({ id, status: 'SKIP', detail: 'pass --slot <id>' }); continue; }
++ let status;
++ let detail;
++ try { [status, detail] = await RUNNERS[id](ctx); } catch (err) { status = 'FAIL'; detail = `check crashed: ${err.message}`; }
++ results.push({ id, status, detail });
++ }
++ const bad = results.filter((r) => r.status === 'FAIL' || (opts.strict && ['WARN', 'MANUAL'].includes(r.status)));
++ return { ok: bad.length === 0, results };
++}
++
++function format(report) {
++ const lines = report.results.map((r) => `${r.status.padEnd(6)} ${r.id.padEnd(9)} ${r.detail}`);
++ lines.push('', report.ok ? 'preflight: no blocking failure' : `preflight: ${report.results.filter((r) => r.status === 'FAIL').length} failed`);
++ return lines.join('\n');
++}
++
++if (require.main === module) {
++ let opts;
++ try { opts = parseArgs(process.argv.slice(2)); } catch (err) { console.error(`cutover-preflight: ${err.message}`); process.exit(2); }
++ if (opts.help) { console.log(fs.readFileSync(__filename, 'utf8').split('\n').slice(2, 37).join('\n')); process.exit(0); }
++ preflight(opts).then((report) => {
++ console.log(opts.json ? JSON.stringify(report, null, 2) : format(report));
++ process.exit(report.ok ? 0 : 1);
++ }, (err) => { console.error(`cutover-preflight: ${err.stack || err}`); process.exit(2); });
++}
++
++module.exports = { preflight, parseArgs, format, parseEnvFile, parseSsListeners, localAddresses, rtmpProbe, sameSecret, CHECKS, DEFAULTS };
ddiff --git a/scripts/live-credentials.js b/scripts/live-credentials.js
nnew file mode 100644
iindex 0000000..e6f03f4
---- /dev/null
++++ b/scripts/live-credentials.js
@@@ -0,0 +1,43 @@
++'use strict';
++/**
++ * The cutover scripts that act AS LIVE (subscribe-live-events.js, set-definition-state.js) read
++ * Live's env file and ask Network for a client-credentials token with Live's OAuth client. Only
++ * these names are used: OV_OAUTH_CLIENT_ID (default live), OV_OAUTH_CLIENT_SECRET and
++ * OV_NETWORK_INTERNAL_URL (default http://127.0.0.1:4000). No value is ever printed.
++ */
++const fs = require('fs');
++
++const DEFAULT_LIVE_ENV = '/etc/openvibe/live.env';
++
++/** Parse Live's env file (root-only, 0600: the scripts run under sudo). */
++function readLiveEnv(file = DEFAULT_LIVE_ENV) {
++ let text;
++ try { text = fs.readFileSync(file, 'utf8'); } catch (err) {
++ throw new Error(`cannot read ${file} (${err.code}); run as root`);
++ }
++ return require('dotenv').parse(text);
++}
++
++async function readJson(res) {
++ const text = await res.text();
++ try { return text ? JSON.parse(text) : {}; } catch { return { detail: text.slice(0, 200) }; }
++}
++
++/** A service token for Live's client, for one audience (openvibe.events, openvibe.openre). */
++async function liveToken({ env, audience, fetchImpl = globalThis.fetch }) {
++ const clientId = env.OV_OAUTH_CLIENT_ID || 'live';
++ const clientSecret = env.OV_OAUTH_CLIENT_SECRET || '';
++ if (!clientSecret) throw new Error('OV_OAUTH_CLIENT_SECRET is not set in the Live env file');
++ const network = String(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
++ const res = await fetchImpl(`${network}/oauth/token`, {
++ method: 'POST',
++ headers: { 'Content-Type': 'application/x-www-form-urlencoded', Accept: 'application/json' },
++ body: new URLSearchParams({ grant_type: 'client_credentials', client_id: clientId, client_secret: clientSecret, audience }).toString(),
++ signal: AbortSignal.timeout(15000),
++ });
++ const tok = await readJson(res);
++ if (!res.ok || !tok.access_token) throw new Error(`Network refused a token for ${clientId} (${res.status} ${tok.error || tok.code || ''})`.trim());
++ return { token: tok.access_token, clientId };
++}
++
++module.exports = { readLiveEnv, readJson, liveToken, DEFAULT_LIVE_ENV };
ddiff --git a/scripts/migrate-from-live.js b/scripts/migrate-from-live.js
iindex 3a2a114..edb2130 100755
---- a/scripts/migrate-from-live.js
++++ b/scripts/migrate-from-live.js
@@@ -8,9 +8,11 @@
* node scripts/migrate-from-live.js --live-db /path/to/live-snapshot.db --apply # write
* ... [--slots 12,31] [--checklist /path/to/checklist.md]
*
-- * Take the snapshot with `sqlite3 /opt/openvibe.live/data/openvibe.db ".backup /tmp/live-snapshot.db"`
-- * (never point this at the live file). Uses the OpenRe environment (.env or OPENRE_ENV_FILE) for
-- * its own database and OPENRE_SECRETS_KEY. Prints no secret: old keys are not imported and the new
++ * Take the snapshot with `sqlite3 /opt/openvibe.live/data/live.db ".backup /tmp/live-snapshot.db"`
++ * (Live's DB_PATH in production; never point this at the live file). Uses the OpenRe environment
++ * (.env or OPENRE_ENV_FILE) for its own database and OPENRE_SECRETS_KEY; in production set
++ * OPENRE_ENV_FILE=/etc/openvibe/openre.env, OPENRE_DB_PATH=/var/lib/openre/openre.db and
++ * NODE_ENV=production (the units set the last two themselves) — docs/cutover.md has the command. Prints no secret: old keys are not imported and the new
* keys are never shown (the broadcaster rotates to get one). Safe to re-run.
*/
const fs = require('fs');
ddiff --git a/scripts/set-definition-state.js b/scripts/set-definition-state.js
nnew file mode 100644
iindex 0000000..5789b73
---- /dev/null
++++ b/scripts/set-definition-state.js
@@@ -0,0 +1,67 @@
++#!/usr/bin/env node
++'use strict';
++/**
++ * Disable (or re-enable) the OpenRe stream definition of one Live slot, AS LIVE (docs/cutover.md,
++ * per-slot rollback). A disabled definition's keys are refused at the RTMP handshake
++ * (`stream_disabled`), so after a slot goes back to Live nobody can publish to OpenRe with the key
++ * the broadcaster got at the cutover, and OpenRe never pushes that slot's destinations. Running
++ * sessions are not ended (Live holds no openre.session.end grant): do it with the slot offline.
++ *
++ * sudo node /opt/openre.stream/current/scripts/set-definition-state.js --slot 12 --state disabled
++ * sudo node /opt/openre.stream/current/scripts/set-definition-state.js --slot 12 --state active
++ * [--live-env /etc/openvibe/live.env]
++ *
++ * Uses Live's OAuth client (grants openre.stream.read + openre.stream.write on openvibe.openre, both
++ * held since 2026-09-23) and OPENRE_URL from Live's env file (default http://127.0.0.1:4500). Prints
++ * no secret and no key.
++ */
++const { readLiveEnv, readJson, liveToken, DEFAULT_LIVE_ENV } = require('./live-credentials');
++
++function parseArgs(argv) {
++ const o = { liveEnv: DEFAULT_LIVE_ENV, slot: null, state: null };
++ for (let i = 0; i < argv.length; i++) {
++ const a = argv[i];
++ if (a === '--slot') o.slot = argv[++i];
++ else if (a === '--state') o.state = argv[++i];
++ else if (a === '--live-env') o.liveEnv = argv[++i];
++ else throw new Error(`unknown argument ${a}`);
++ }
++ if (!/^\d+$/.test(String(o.slot || ''))) throw new Error('--slot <Live managed stream id> is required');
++ if (!['active', 'disabled'].includes(o.state)) throw new Error('--state must be active or disabled');
++ return o;
++}
++
++async function run({ slot, state, env, fetchImpl = globalThis.fetch, log = console.log }) {
++ const openre = String(env.OPENRE_URL || 'http://127.0.0.1:4500').replace(/\/+$/, '');
++ const { token } = await liveToken({ env, audience: 'openvibe.openre', fetchImpl });
++ const headers = { Authorization: `Bearer ${token}`, Accept: 'application/json' };
++ const ref = `live:managed_stream:${slot}`;
++ const found = await fetchImpl(`${openre}/api/v1/streams?external_ref=${encodeURIComponent(ref)}`, { headers, signal: AbortSignal.timeout(15000) });
++ const list = await readJson(found);
++ if (!found.ok) throw new Error(`OpenRe answered ${found.status}: ${list.code || ''} ${list.detail || ''}`.trim());
++ const d = (list.streams || [])[0];
++ if (!d) throw new Error(`OpenRe has no stream definition for ${ref}`);
++ if (d.state === state) {
++ log(`${d.id} (${ref}) is already ${state}`);
++ return { id: d.id, state, changed: false, open_session: d.session ? d.session.id : null };
++ }
++ const r = await fetchImpl(`${openre}/api/v1/streams/${encodeURIComponent(d.id)}`, {
++ method: 'PATCH',
++ headers: { ...headers, 'Content-Type': 'application/json' },
++ body: JSON.stringify({ state }),
++ signal: AbortSignal.timeout(15000),
++ });
++ const b = await readJson(r);
++ if (!r.ok) throw new Error(`OpenRe answered ${r.status}: ${b.code || ''} ${b.detail || ''}`.trim());
++ log(`${d.id} (${ref}): ${d.state} → ${b.stream ? b.stream.state : state}${d.session ? `; session ${d.session.id} is still ${d.session.state} (not ended by this)` : ''}`);
++ return { id: d.id, state: b.stream ? b.stream.state : state, changed: true, open_session: d.session ? d.session.id : null };
++}
++
++if (require.main === module) {
++ let o;
++ let env;
++ try { o = parseArgs(process.argv.slice(2)); env = readLiveEnv(o.liveEnv); } catch (err) { console.error(`set-definition-state: ${err.message}`); process.exit(2); }
++ run({ slot: o.slot, state: o.state, env }).then(() => process.exit(0), (err) => { console.error(`set-definition-state: ${err.message}`); process.exit(1); });
++}
++
++module.exports = { run, parseArgs };
ddiff --git a/scripts/subscribe-live-events.js b/scripts/subscribe-live-events.js
nnew file mode 100644
iindex 0000000..69b1b5b
---- /dev/null
++++ b/scripts/subscribe-live-events.js
@@@ -0,0 +1,106 @@
++#!/usr/bin/env node
++'use strict';
++/**
++ * Live's OpenVibe.Events subscription to OpenRe's session events (docs/cutover.md step A5):
++ * topic `openre.session.*` → Live's `POST /internal/openre-events`, signed with the secret Live holds
++ * as OPENRE_EVENTS_SECRET. Events names the consumer after the calling service, so this runs AS LIVE:
++ * it reads Live's env file (Live's OAuth client, grant [live, events.subscription.manage,
++ * openvibe.events]) and asks Network for a token with it. No secret is printed. Run on the host as
++ * root (the env file is 0600):
++ *
++ * sudo node /opt/openre.stream/current/scripts/subscribe-live-events.js # create (or report)
++ * sudo node /opt/openre.stream/current/scripts/subscribe-live-events.js --dry-run # list, change nothing
++ * sudo node /opt/openre.stream/current/scripts/subscribe-live-events.js --disable # rollback
++ * sudo node /opt/openre.stream/current/scripts/subscribe-live-events.js --enable # undo a --disable
++ *
++ * [--live-env /etc/openvibe/live.env] [--endpoint http://127.0.0.1:3000/internal/openre-events]
++ *
++ * Env names used (from the Live env file): OV_OAUTH_CLIENT_ID, OV_OAUTH_CLIENT_SECRET,
++ * OV_NETWORK_INTERNAL_URL (scripts/live-credentials.js), EVENTS_URL (default http://127.0.0.1:4300)
++ * and OPENRE_EVENTS_SECRET (32+ characters; generate it with `openssl rand -hex 32` and put it in the
++ * Live env file first). An existing subscription with the same topic and endpoint is reported, not
++ * duplicated; its secret cannot be read back through the API, so `cutover-preflight.js --only events`
++ * compares it with Live's by hash. A new subscription gets no history: sessions from before it
++ * (the 2026-09-23 rehearsal) are not mirrored into Live.
++ */
++const { readLiveEnv, readJson, liveToken, DEFAULT_LIVE_ENV } = require('./live-credentials');
++
++const TOPIC = 'openre.session.*';
++const DEFAULT_ENDPOINT = 'http://127.0.0.1:3000/internal/openre-events';
++
++function parseArgs(argv) {
++ const o = { liveEnv: DEFAULT_LIVE_ENV, endpoint: DEFAULT_ENDPOINT, action: 'create' };
++ for (let i = 0; i < argv.length; i++) {
++ const a = argv[i];
++ if (a === '--dry-run') o.action = 'list';
++ else if (a === '--disable') o.action = 'disable';
++ else if (a === '--enable') o.action = 'enable';
++ else if (a === '--live-env') o.liveEnv = argv[++i];
++ else if (a === '--endpoint') o.endpoint = argv[++i];
++ else throw new Error(`unknown argument ${a}`);
++ }
++ if (!o.liveEnv || !o.endpoint) throw new Error('--live-env and --endpoint need a value');
++ return o;
++}
++
++/**
++ * run({ action, endpoint, env, fetchImpl, log }) -> { action, subscription_id?, created?, existed?, enabled? }
++ * `env` is the parsed Live env file.
++ */
++async function run({ action = 'create', endpoint = DEFAULT_ENDPOINT, env, fetchImpl = globalThis.fetch, log = console.log }) {
++ const events = String(env.EVENTS_URL || 'http://127.0.0.1:4300').replace(/\/+$/, '');
++ const secret = env.OPENRE_EVENTS_SECRET || '';
++ if (!env.OV_OAUTH_CLIENT_SECRET) throw new Error('OV_OAUTH_CLIENT_SECRET is not set in the Live env file');
++ if (action === 'create' && secret.length < 32) throw new Error('OPENRE_EVENTS_SECRET must be set in the Live env file first (32+ characters: openssl rand -hex 32)');
++
++ const { token, clientId } = await liveToken({ env, audience: 'openvibe.events', fetchImpl });
++ const headers = { Authorization: `Bearer ${token}`, Accept: 'application/json' };
++
++ const listRes = await fetchImpl(`${events}/api/v1/subscriptions`, { headers, signal: AbortSignal.timeout(15000) });
++ const list = await readJson(listRes);
++ if (!listRes.ok) throw new Error(`Events answered ${listRes.status} listing subscriptions: ${list.code || ''} ${list.detail || ''}`.trim());
++ const mine = (list.subscriptions || []).filter((s) => s.topic_pattern === TOPIC && s.endpoint === endpoint);
++ const existing = mine[0] || null;
++ const describe = (s) => `${s.id} (${TOPIC} → ${endpoint}, ${s.enabled === false || s.enabled === 0 ? 'disabled' : 'enabled'})`;
++
++ if (action === 'list') {
++ if (!existing) log(`no subscription ${TOPIC} → ${endpoint} for ${clientId}; would create one`);
++ for (const s of mine) log(`exists: ${describe(s)}`);
++ return { action, subscription_id: existing ? existing.id : null, existed: Boolean(existing) };
++ }
++ if (action === 'disable' || action === 'enable') {
++ if (!existing) throw new Error(`no subscription ${TOPIC} → ${endpoint} to ${action}`);
++ const r = await fetchImpl(`${events}/api/v1/subscriptions/${encodeURIComponent(existing.id)}/${action}`, { method: 'POST', headers, signal: AbortSignal.timeout(15000) });
++ const b = await readJson(r);
++ if (!r.ok) throw new Error(`Events answered ${r.status} to ${action}: ${b.code || ''} ${b.detail || ''}`.trim());
++ log(`${action}d: ${describe(b.id ? b : { ...existing, enabled: action === 'enable' })}`);
++ return { action, subscription_id: existing.id, enabled: action === 'enable' };
++ }
++ if (existing) {
++ log(`subscription exists: ${describe(existing)}. Check its secret with: cutover-preflight.js --only events`);
++ return { action, subscription_id: existing.id, existed: true, created: false };
++ }
++ const r = await fetchImpl(`${events}/api/v1/subscriptions`, {
++ method: 'POST',
++ headers: { ...headers, 'Content-Type': 'application/json' },
++ body: JSON.stringify({ topic_pattern: TOPIC, endpoint, secret }),
++ signal: AbortSignal.timeout(15000),
++ });
++ const b = await readJson(r);
++ if (r.status === 409 && b.subscription_id) {
++ log(`subscription exists: ${b.subscription_id}`);
++ return { action, subscription_id: b.subscription_id, existed: true, created: false };
++ }
++ if (!r.ok) throw new Error(`Events answered ${r.status}: ${b.code || ''} ${b.detail || ''}`.trim());
++ log(`subscribed: ${b.id} (${TOPIC} → ${endpoint}), signed with Live's OPENRE_EVENTS_SECRET`);
++ return { action, subscription_id: b.id, existed: false, created: true };
++}
++
++if (require.main === module) {
++ let o;
++ let env;
++ try { o = parseArgs(process.argv.slice(2)); env = readLiveEnv(o.liveEnv); } catch (err) { console.error(`subscribe-live-events: ${err.message}`); process.exit(2); }
++ run({ action: o.action, endpoint: o.endpoint, env }).then(() => process.exit(0), (err) => { console.error(`subscribe-live-events: ${err.message}`); process.exit(1); });
++}
++
++module.exports = { run, parseArgs, TOPIC, DEFAULT_ENDPOINT };
ddiff --git a/test/bind-switch.test.js b/test/bind-switch.test.js
nnew file mode 100644
iindex 0000000..bd7b7a5
---- /dev/null
++++ b/test/bind-switch.test.js
@@@ -0,0 +1,88 @@
++'use strict';
++// The cutover's bind switch (docs/cutover.md step 5), with real worker processes: generation 1
++// listens on 127.0.0.1 only (production today), generation 2 starts with OPENRE_RTMP_BIND=0.0.0.0 on
++// the same port (SO_REUSEPORT lets both bind), the coordinator drains generation 1, which closes its
++// listener and exits 0 by itself, and from then on the port answers the RTMP handshake on the
++// host's other addresses too. No worker is restarted. Needs no ffmpeg.
++const assert = require('assert');
++const { load } = require('../server/config');
++const { openRuntime } = require('../server/store');
++const { createCoordinator } = require('../server/coordinator');
++const { localAddresses, rtmpProbe } = require('../scripts/cutover-preflight');
++const { tmpDir, testEnv, child, waitFor, freePort, suite, silent } = require('./helpers');
++
++const t = suite('bind-switch');
++const procs = [];
++process.on('exit', () => { for (const p of procs) { try { p.kill('SIGKILL'); } catch { /* gone */ } } });
++
++// A non-loopback IPv4 address of this machine (the "public" address in the test).
++const outside = localAddresses().find((a) => /^\d+\.\d+\.\d+\.\d+$/.test(a));
++
++let env;
++let port;
++let g1;
++let g2;
++let rt;
++
++function startWorker(bind) {
++ const p = child('workers/rtmp-ingest.js', { ...env, OPENRE_RTMP_BIND: bind });
++ procs.push(p);
++ return p;
++}
++
++t('generation 1 on 127.0.0.1 only: the host\'s other addresses are refused', async () => {
++ port = await freePort();
++ const internal = await freePort();
++ env = testEnv(tmpDir(), {
++ OPENRE_RTMP_PORT: String(port),
++ OPENRE_RTMP_INTERNAL_PORT_MIN: String(internal),
++ OPENRE_RTMP_INTERNAL_PORT_MAX: String(internal + 20),
++ OPENRE_WORKER_HEARTBEAT_MS: '200',
++ });
++ g1 = startWorker('127.0.0.1');
++ await waitFor(() => /generation 1 ready: publish 127\.0\.0\.1:/.test(g1.output), { what: 'generation 1 ready' });
++ const local = await rtmpProbe('127.0.0.1', port, 2000);
++ assert.ok(local.ok && local.rtmp, 'loopback answers the RTMP handshake');
++ if (outside) {
++ const r = await rtmpProbe(outside, port, 2000);
++ assert.strictEqual(r.ok, false);
++ assert.strictEqual(r.code, 'ECONNREFUSED', `${outside}:${port} must be refused while bound to loopback`);
++ } else console.log(' (no non-loopback IPv4 address here: only loopback is checked)');
++});
++
++t('generation 2 starts with OPENRE_RTMP_BIND=0.0.0.0 on the same port while generation 1 still listens', async () => {
++ g2 = startWorker('0.0.0.0');
++ await waitFor(() => /generation 2 ready: publish 0\.0\.0\.0:/.test(g2.output), { what: 'generation 2 ready' });
++ assert.strictEqual(g1.exitCode, null, 'generation 1 is still running');
++});
++
++t('the coordinator drains generation 1; it closes its listener and exits 0 by itself', async () => {
++ rt = openRuntime({ config: load(env), log: silent });
++ const coordinator = createCoordinator({ rt, media: null, log: silent });
++ const out = coordinator.syncTick();
++ assert.strictEqual(out.drained, 1);
++ const r = await g1.exited;
++ assert.strictEqual(r.code, 0, 'a drained, idle generation exits cleanly');
++ assert.match(g1.output, /public listener closed/);
++ const rows = rt.db.prepare("SELECT generation, state, stop_reason FROM workers WHERE kind = 'rtmp-ingest' ORDER BY generation").all();
++ assert.deepStrictEqual(rows.map((w) => [w.generation, w.state]), [[1, 'stopped'], [2, 'ready']]);
++});
++
++t('the port now answers on every address, served by generation 2', async () => {
++ const local = await rtmpProbe('127.0.0.1', port, 2000);
++ assert.ok(local.ok && local.rtmp, 'loopback still answers (0.0.0.0 covers it)');
++ if (outside) {
++ const r = await rtmpProbe(outside, port, 2000);
++ assert.ok(r.ok && r.rtmp, `${outside}:${port} answers the RTMP handshake`);
++ }
++ assert.strictEqual(g2.exitCode, null);
++});
++
++t('teardown: SIGTERM drains generation 2, which exits 0 when idle', async () => {
++ g2.kill('SIGTERM');
++ const r = await g2.exited;
++ assert.strictEqual(r.code, 0);
++ rt.db.close();
++});
++
++t.run();
ddiff --git a/test/preflight.test.js b/test/preflight.test.js
nnew file mode 100644
iindex 0000000..760259b
---- /dev/null
++++ b/test/preflight.test.js
@@@ -0,0 +1,422 @@
++'use strict';
++// scripts/cutover-preflight.js against a mocked host: files, commands, DNS, interfaces, the port
++// probe and /api/ready are fakes; the three databases are real SQLite files opened read-only.
++// Covers the host as it is today (2026-09-23: everything the cutover still needs fails), a host
++// ready for the first slot, each failure on its own, the off-host port check, the probe URL, the
++// per-slot check, and that no secret value ever reaches the output.
++const assert = require('assert');
++const net = require('net');
++const path = require('path');
++const Database = require('better-sqlite3');
++const { openDb } = require('../server/db');
++const { preflight, parseArgs, format, parseSsListeners, rtmpProbe, sameSecret, CHECKS } = require('../scripts/cutover-preflight');
++const { tmpDir, freePort, suite } = require('./helpers');
++
++const t = suite('preflight');
++
++const HOST_IP = '15.204.79.215';
++const RELEASE = 'aaaaaaaaaaaa';
++const EVENTS_SECRET = 'e'.repeat(64);
++const CLIENT_SECRET = 'c'.repeat(48);
++const SECRETS_KEY = 'f'.repeat(64);
++const LIVE_PID = 4242;
++
++// ── fixtures: three databases ──────────────────────────────────────────────
++
++function makeDbs() {
++ const dir = tmpDir();
++ const openrePath = path.join(dir, 'openre.db');
++ const o = openDb(openrePath);
++ const now = Date.now();
++ const worker = o.prepare(`INSERT INTO workers (id, kind, generation, release, state, endpoints, started_at, heartbeat_at) VALUES (?, ?, ?, ?, ?, '{}', ?, ?)`);
++ worker.run('wrk_1', 'rtmp-ingest', 2, RELEASE, 'ready', now, now);
++ worker.run('wrk_2', 'restream', 2, RELEASE, 'ready', now, now);
++ o.prepare(`INSERT INTO stream_definitions (id, owner_subject, mirror_to_live, created_at, updated_at) VALUES ('str_1', 'usr_1', 1, ?, ?)`).run(now, now);
++ o.prepare(`INSERT INTO external_refs (definition_id, service, type, ref_id, created_at) VALUES ('str_1', 'live', 'managed_stream', '12', ?)`).run(now);
++ o.prepare(`INSERT INTO migration_map (source_system, source_type, source_id, target_type, target_id, status, imported_at) VALUES ('live', 'managed_stream', '12', 'stream', 'str_1', 'imported', ?)`).run(now);
++ o.close();
++
++ const eventsPath = path.join(dir, 'events.db');
++ const e = new Database(eventsPath);
++ e.exec(`CREATE TABLE subscriptions (id TEXT PRIMARY KEY, consumer TEXT NOT NULL, topic_pattern TEXT NOT NULL, endpoint TEXT NOT NULL, secret TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1, retry_policy TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, project_id TEXT, env TEXT NOT NULL DEFAULT 'production');
++ CREATE TABLE deliveries (event_id TEXT NOT NULL, subscription_id TEXT NOT NULL, seq INTEGER NOT NULL, priority INTEGER NOT NULL, attempt INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL, next_attempt_at INTEGER, last_error TEXT, last_status INTEGER, delivered_at INTEGER, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (event_id, subscription_id));`);
++ e.close();
++
++ const livePath = path.join(dir, 'live.db');
++ const l = new Database(livePath);
++ l.exec(`CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT, is_banned INTEGER DEFAULT 0);
++ CREATE TABLE managed_streams (id INTEGER PRIMARY KEY, user_id INTEGER, slug TEXT, title TEXT, protocol TEXT, streaming_method TEXT, ingest_authority TEXT DEFAULT 'live', openre_stream_id TEXT);
++ CREATE TABLE streams (id INTEGER PRIMARY KEY, managed_stream_id INTEGER, is_live INTEGER);
++ CREATE TABLE linked_accounts (id INTEGER PRIMARY KEY, user_id INTEGER, service TEXT, service_user_id TEXT, subject_id TEXT);
++ INSERT INTO users (id, username) VALUES (5, 'rehearsal');
++ INSERT INTO managed_streams (id, user_id, slug, title, protocol, streaming_method) VALUES (12, 5, 'test', 'Test', 'rtmp', 'obs');
++ INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (5, 'network', '77', 'usr_01J0000000000000000000000A');`);
++ l.close();
++ return { dir, openrePath, eventsPath, livePath };
++}
++
++function addSubscription(dbs, { secret = EVENTS_SECRET, enabled = 1, id = 'sub_1' } = {}) {
++ const e = new Database(dbs.eventsPath);
++ e.prepare(`INSERT INTO subscriptions (id, consumer, topic_pattern, endpoint, secret, enabled, created_at, updated_at) VALUES (?, 'live', 'openre.session.*', 'http://127.0.0.1:3000/internal/openre-events', ?, ?, 0, 0)`).run(id, secret, enabled);
++ e.close();
++}
++
++function addDelivery(dbs, status, n = 1) {
++ const e = new Database(dbs.eventsPath);
++ for (let i = 0; i < n; i++) e.prepare("INSERT INTO deliveries (event_id, subscription_id, seq, priority, status, created_at, updated_at) VALUES (?, 'sub_1', ?, 1, ?, 0, 0)").run(`evt_${status}_${i}`, i, status);
++ e.close();
++}
++
++// ── fixtures: the host ─────────────────────────────────────────────────────
++
++const envText = (o) => Object.entries(o).filter(([, v]) => v != null).map(([k, v]) => `${k}=${v}`).join('\n') + '\n';
++
++/** A host that is ready for the first slot. Every field can be overridden per test. */
++function host(dbs, over = {}) {
++ const h = {
++ release: RELEASE,
++ originMain: RELEASE,
++ containsCommit: true,
++ openreEnv: { OV_OAUTH_CLIENT_ID: 'openre', OV_OAUTH_CLIENT_SECRET: CLIENT_SECRET, OPENRE_SECRETS_KEY: SECRETS_KEY, OPENRE_RTMP_PORT: '1936', OPENRE_RTMP_BIND: '0.0.0.0', OPENRE_RTMP_PUBLIC_HOST: 'ingest.openre.stream', EVENTS_URL: 'http://127.0.0.1:4300', MEDIA_API_KEY: 'm'.repeat(40) },
++ liveEnv: { OV_OAUTH_CLIENT_ID: 'live', OV_OAUTH_CLIENT_SECRET: CLIENT_SECRET, OPENRE_URL: 'http://127.0.0.1:4500', OPENRE_PUBLIC_URL: 'https://openre.stream', OPENRE_EVENTS_SECRET: EVENTS_SECRET },
++ runningLiveEnv: null, // default: same as liveEnv
++ ss: 'LISTEN 0 511 0.0.0.0:1936 0.0.0.0:*\nLISTEN 0 511 127.0.0.1:19360 0.0.0.0:*\n',
++ a4: [HOST_IP],
++ a6: [],
++ interfaces: { lo: [{ address: '127.0.0.1', internal: true }], ens3: [{ address: HOST_IP, internal: false }, { address: '2604:2dc0:202:200::7c', internal: false }] },
++ probe: { ok: true, rtmp: true, ms: 12 },
++ ready: { status: 200, body: { status: 'ready', checks: { db: true, key: true }, workers: [{ kind: 'restream', generation: 2, state: 'ready' }, { kind: 'rtmp-ingest', generation: 2, state: 'ready' }], coordinator: { holder: 'coordinator:1', lease_valid: true }, events: { configured: true, pending: 0, rejected: 0 } } },
++ probeUrl: null,
++ ...over,
++ };
++ const calls = [];
++ const files = {
++ '/etc/openvibe/openre.env': h.openreEnv ? envText(h.openreEnv) : null,
++ '/etc/openvibe/live.env': h.liveEnv ? envText(h.liveEnv) : null,
++ };
++ const running = h.runningLiveEnv || h.liveEnv || {};
++ const deps = {
++ calls,
++ readFile: (p) => (p in files ? files[p] : null),
++ readBinary: (p) => (p === `/proc/${LIVE_PID}/environ` ? Buffer.from(`PATH=/usr/bin\0${Object.entries(running).map(([k, v]) => `${k}=${v}`).join('\0')}\0`) : null),
++ readlink: (p) => (p === '/opt/openre.stream/current' && h.release ? `/opt/openre.stream/releases/${h.release}` : null),
++ run: async (cmd, args) => {
++ calls.push([cmd, ...args]);
++ if (cmd === 'git' && args.includes('rev-parse')) return { code: 0, stdout: `${h.originMain}\n`, stderr: '' };
++ if (cmd === 'git' && args.includes('merge-base')) {
++ if (h.containsCommit === 'unknown') return { code: 128, stdout: '', stderr: `fatal: Not a valid object name ${args[args.length - 2]}\n` };
++ return { code: h.containsCommit ? 0 : 1, stdout: '', stderr: '' };
++ }
++ if (cmd === 'ss') return { code: 0, stdout: h.ss, stderr: '' };
++ if (cmd === 'systemctl') return { code: 0, stdout: `MainPID=${h.livePid === undefined ? LIVE_PID : h.livePid}\n`, stderr: '' };
++ return { code: 127, stdout: '', stderr: `no ${cmd}` };
++ },
++ resolve4: async () => h.a4,
++ resolve6: async () => h.a6,
++ interfaces: () => h.interfaces,
++ probe: async (hostName, port) => { calls.push(['probe', hostName, port]); return h.probe; },
++ fetch: async (url) => {
++ calls.push(['fetch', url]);
++ if (url.endsWith('/api/ready')) {
++ if (h.ready instanceof Error) throw h.ready;
++ return { status: h.ready.status, ok: h.ready.status < 300, json: async () => h.ready.body };
++ }
++ if (h.probeUrl) return { status: h.probeUrl.status || 200, ok: (h.probeUrl.status || 200) < 300, json: async () => h.probeUrl.body };
++ throw new Error(`unexpected fetch ${url}`);
++ },
++ openDb: h.openDb || ((p) => new Database(p, { readonly: true, fileMustExist: true })),
++ };
++ deps.h = h;
++ return deps;
++}
++
++function opts(dbs, argv = []) {
++ return parseArgs(['--openre-db', dbs.openrePath, '--events-db', dbs.eventsPath, '--live-db', dbs.livePath, ...argv]);
++}
++
++const byId = (report) => Object.fromEntries(report.results.map((r) => [r.id, r]));
++
++function assertNoSecret(report) {
++ const text = JSON.stringify(report) + format(report);
++ for (const s of [EVENTS_SECRET, CLIENT_SECRET, SECRETS_KEY, 'm'.repeat(40), 'x'.repeat(64)]) assert.ok(!text.includes(s), 'a secret value reached the output');
++}
++
++// ── tests ──────────────────────────────────────────────────────────────────
++
++t('a host ready for the first slot: every check passes; on the host the port check is MANUAL', async () => {
++ const dbs = makeDbs();
++ addSubscription(dbs);
++ const deps = host(dbs);
++ const report = await preflight(opts(dbs), deps);
++ const r = byId(report);
++ assert.deepStrictEqual(Object.keys(r), CHECKS.filter((c) => c !== 'slot'), 'slot runs only with --slot');
++ for (const id of ['release', 'service', 'env', 'bind', 'dns', 'db', 'live-env', 'events']) assert.strictEqual(r[id].status, 'PASS', `${id}: ${r[id].detail}`);
++ assert.strictEqual(r.port.status, 'MANUAL');
++ assert.match(r.port.detail, /never crosses the provider edge/);
++ assert.match(r.port.detail, /--only port/);
++ assert.strictEqual(report.ok, true);
++ assert.strictEqual((await preflight(opts(dbs, ['--strict']), host(dbs))).ok, false, '--strict counts MANUAL as blocking');
++ assertNoSecret(report);
++ // read-only: the only commands are git reads, ss and systemctl show
++ for (const c of deps.calls.filter((x) => !['probe', 'fetch'].includes(x[0]))) {
++ assert.ok(['git rev-parse', 'git merge-base', 'ss -Hltn', 'systemctl show'].some((p) => c.join(' ').replace(/-C \S+ /, '').startsWith(p)), `unexpected command ${c.join(' ')}`);
++ }
++});
++
++t('the host as it is on 2026-09-23: old release, loopback bind, Live env and subscription missing', async () => {
++ const dbs = makeDbs();
++ const deps = host(dbs, {
++ release: '655b98a10aaa',
++ originMain: '450f87c3b2a1',
++ containsCommit: false,
++ openreEnv: { ...host(dbs).h.openreEnv, OPENRE_RTMP_BIND: '127.0.0.1' },
++ liveEnv: { OV_OAUTH_CLIENT_ID: 'live', OV_OAUTH_CLIENT_SECRET: CLIENT_SECRET },
++ ss: 'LISTEN 0 511 127.0.0.1:1936 0.0.0.0:*\n',
++ });
++ const report = await preflight(opts(dbs), deps);
++ const r = byId(report);
++ assert.strictEqual(report.ok, false);
++ assert.strictEqual(r.release.status, 'FAIL');
++ assert.match(r.release.detail, /current = 655b98a10aaa, expected 450f87c3b2a1/);
++ assert.match(r.release.detail, /does not contain 6dc78a5/);
++ assert.strictEqual(r.bind.status, 'FAIL');
++ assert.match(r.bind.detail, /OPENRE_RTMP_BIND=127\.0\.0\.1; listening on 127\.0\.0\.1:1936: set OPENRE_RTMP_BIND=0\.0\.0\.0/);
++ assert.strictEqual(r['live-env'].status, 'FAIL');
++ assert.match(r['live-env'].detail, /OPENRE_URL not set/);
++ assert.match(r['live-env'].detail, /OPENRE_EVENTS_SECRET not set/);
++ assert.strictEqual(r.events.status, 'FAIL');
++ assert.match(r.events.detail, /no subscription live openre\.session\.\* → http:\/\/127\.0\.0\.1:3000\/internal\/openre-events/);
++ assert.strictEqual(r.dns.status, 'PASS', 'DNS is already done');
++ assert.strictEqual(r.db.status, 'PASS');
++ assertNoSecret(report);
++});
++
++t('release: --expect-release wins over origin/main; --require-commit none skips the ancestry check', async () => {
++ const dbs = makeDbs();
++ const deps = host(dbs, { originMain: 'bbbbbbbbbbbb', containsCommit: false });
++ let r = byId(await preflight(opts(dbs, ['--only', 'release', '--expect-release', RELEASE, '--require-commit', 'none']), deps));
++ assert.strictEqual(r.release.status, 'PASS');
++ assert.ok(!deps.calls.some((c) => c.includes('rev-parse')), 'origin/main is not read when a release is given');
++ r = byId(await preflight(opts(dbs, ['--only', 'release']), host(dbs, { containsCommit: 'unknown' })));
++ assert.match(r.release.detail, /6dc78a5 is not in \/opt\/openre\.stream\/repo yet \(deploy\.sh release fetches it\)/, 'production on 2026-09-23: the repo never fetched 6dc78a5');
++ r = byId(await preflight(opts(dbs, ['--only', 'release']), host(dbs, { release: null })));
++ assert.match(r.release.detail, /not a symlink/);
++});
++
++t('bind: public in the env but the running generation still listens on loopback → deploy.sh workers', async () => {
++ const dbs = makeDbs();
++ const r = byId(await preflight(opts(dbs, ['--only', 'bind']), host(dbs, { ss: 'LISTEN 0 511 127.0.0.1:1936 0.0.0.0:*\n' })));
++ assert.strictEqual(r.bind.status, 'FAIL');
++ assert.match(r.bind.detail, /deploy\.sh workers/);
++ const r6 = byId(await preflight(opts(dbs, ['--only', 'bind']), host(dbs, { ss: 'LISTEN 0 511 [::]:1936 [::]:*\nLISTEN 0 511 0.0.0.0:19360 0.0.0.0:*\n', openreEnv: { ...host(dbs).h.openreEnv, OPENRE_RTMP_BIND: '::' } })));
++ assert.strictEqual(r6.bind.status, 'PASS', r6.bind.detail);
++});
++
++t('env: OPENRE_DRILL in the production env file fails; so do a wrong port or host', async () => {
++ const dbs = makeDbs();
++ const base = host(dbs).h.openreEnv;
++ let r = byId(await preflight(opts(dbs, ['--only', 'env']), host(dbs, { openreEnv: { ...base, OPENRE_DRILL: '1' } })));
++ assert.strictEqual(r.env.status, 'FAIL');
++ assert.match(r.env.detail, /OPENRE_DRILL is set/);
++ r = byId(await preflight(opts(dbs, ['--only', 'env']), host(dbs, { openreEnv: { ...base, OPENRE_RTMP_PORT: '1935', OPENRE_RTMP_PUBLIC_HOST: 'openre.stream' } })));
++ assert.match(r.env.detail, /OPENRE_RTMP_PORT = 1935/);
++ assert.match(r.env.detail, /OPENRE_RTMP_PUBLIC_HOST = openre\.stream/);
++ r = byId(await preflight(opts(dbs, ['--only', 'env,bind']), host(dbs, { openreEnv: null })));
++ assert.match(r.env.detail, /cannot read \/etc\/openvibe\/openre\.env \(run as root\)/);
++ assert.strictEqual(r.bind.status, 'FAIL');
++});
++
++t('dns: a proxied (foreign) address fails; AAAA with an IPv4-only bind warns; off the host --host-ip decides', async () => {
++ const dbs = makeDbs();
++ let r = byId(await preflight(opts(dbs, ['--only', 'dns']), host(dbs, { a4: ['104.21.3.4'] })));
++ assert.strictEqual(r.dns.status, 'FAIL');
++ assert.match(r.dns.detail, /Cloudflare-proxied/);
++ r = byId(await preflight(opts(dbs, ['--only', 'dns']), host(dbs, { a6: ['2604:2dc0:202:200::7c'] })));
++ assert.strictEqual(r.dns.status, 'WARN');
++ assert.match(r.dns.detail, /IPv4 only/);
++ r = byId(await preflight(opts(dbs, ['--only', 'dns']), host(dbs, { a4: [] })));
++ assert.match(r.dns.detail, /no A or AAAA record/);
++ const laptop = { lo: [{ address: '127.0.0.1', internal: true }], wlan0: [{ address: '10.0.0.9', internal: false }] };
++ r = byId(await preflight(opts(dbs, ['--only', 'dns']), host(dbs, { interfaces: laptop })));
++ assert.strictEqual(r.dns.status, 'FAIL');
++ assert.match(r.dns.detail, /--host-ip/);
++ r = byId(await preflight(opts(dbs, ['--only', 'dns', '--host-ip', HOST_IP]), host(dbs, { interfaces: laptop })));
++ assert.strictEqual(r.dns.status, 'PASS');
++});
++
++t('port from outside the host: handshake = PASS, refused = no public listener, timeout = provider edge', async () => {
++ const dbs = makeDbs();
++ const laptop = { wlan0: [{ address: '10.0.0.9', internal: false }] };
++ const run = (probe) => preflight(opts(dbs, ['--only', 'port']), host(dbs, { interfaces: laptop, probe })).then(byId);
++ let r = await run({ ok: true, rtmp: true, ms: 40 });
++ assert.strictEqual(r.port.status, 'PASS');
++ assert.match(r.port.detail, /ingest\.openre\.stream:1936 \(15\.204\.79\.215\) answers the RTMP handshake from outside/);
++ r = await run({ ok: false, code: 'ECONNREFUSED', error: 'connect ECONNREFUSED' });
++ assert.strictEqual(r.port.status, 'FAIL');
++ assert.match(r.port.detail, /no public listener/);
++ r = await run({ ok: false, code: 'ETIMEDOUT', error: 'no answer within 5s' });
++ assert.strictEqual(r.port.status, 'FAIL');
++ assert.match(r.port.detail, /open 1936\/tcp at the provider edge \(owner, docs\/cutover\.md A4\)/);
++ r = await run({ ok: true, rtmp: false, error: 'connected, but no RTMP handshake answer' });
++ assert.strictEqual(r.port.status, 'WARN');
++});
++
++t('port with --probe-url: open/closed from the probe, anything else MANUAL', async () => {
++ const dbs = makeDbs();
++ const url = 'https://probe.example/check?host={host}&port={port}';
++ const run = (probeUrl) => {
++ const deps = host(dbs, { probeUrl });
++ return preflight(opts(dbs, ['--only', 'port', '--probe-url', url]), deps).then((rep) => ({ r: byId(rep), deps }));
++ };
++ let { r, deps } = await run({ body: { open: true } });
++ assert.strictEqual(r.port.status, 'PASS');
++ assert.ok(deps.calls.some((c) => c[0] === 'fetch' && c[1] === 'https://probe.example/check?host=ingest.openre.stream&port=1936'));
++ assert.ok(!deps.calls.some((c) => c[0] === 'probe'), 'no direct connection when a probe URL is given');
++ ({ r } = await run({ body: { reachable: false } }));
++ assert.strictEqual(r.port.status, 'FAIL');
++ ({ r } = await run({ body: { hello: 'world' } }));
++ assert.strictEqual(r.port.status, 'MANUAL');
++ ({ r } = await run({ status: 500, body: {} }));
++ assert.strictEqual(r.port.status, 'MANUAL');
++});
++
++t('service: not ready, drill mode, a worker on an old release, no lease, relay off, draining', async () => {
++ const dbs = makeDbs();
++ const ready = host(dbs).h.ready;
++ const run = (body, status = 200, argv = []) => preflight(opts(dbs, ['--only', 'release,service', ...argv]), host(dbs, { ready: { status, body } })).then(byId);
++ let r = await run({ ...ready.body, status: 'not_ready', checks: { db: true, key: false } }, 503);
++ assert.strictEqual(r.service.status, 'FAIL');
++ assert.match(r.service.detail, /status 503 not_ready/);
++ r = await run({ ...ready.body, mode: 'drill' });
++ assert.match(r.service.detail, /drill mode/);
++ r = await run({ ...ready.body, coordinator: { lease_valid: false }, events: { configured: false } });
++ assert.match(r.service.detail, /no valid coordinator lease/);
++ assert.match(r.service.detail, /event relay not configured/);
++ r = await run({ ...ready.body, workers: [...ready.body.workers, { kind: 'rtmp-ingest', generation: 1, state: 'draining' }] });
++ assert.strictEqual(r.service.status, 'WARN');
++ assert.match(r.service.detail, /still draining: rtmp-ingest#1/);
++ r = await run({ ...ready.body, workers: [{ kind: 'restream', generation: 2, state: 'ready' }] });
++ assert.match(r.service.detail, /0 ready rtmp-ingest workers/);
++ // the ready generation runs another release than `current`
++ const db = new Database(dbs.openrePath);
++ db.prepare("UPDATE workers SET release = '655b98a10aaa' WHERE kind = 'rtmp-ingest'").run();
++ db.close();
++ r = await run(ready.body);
++ assert.strictEqual(r.service.status, 'FAIL');
++ assert.match(r.service.detail, /rtmp-ingest#2 runs release 655b98a10aaa, current is aaaaaaaaaaaa: deploy\.sh workers/);
++ const down = byId(await preflight(opts(dbs, ['--only', 'service']), host(dbs, { ready: new Error('connect ECONNREFUSED 127.0.0.1:4500') })));
++ assert.match(down.service.detail, /unreachable/);
++});
++
++t('live-env: Live running without the new settings must be restarted; a short secret is refused', async () => {
++ const dbs = makeDbs();
++ const env = host(dbs).h.liveEnv;
++ let r = byId(await preflight(opts(dbs, ['--only', 'live-env']), host(dbs, { runningLiveEnv: { OV_OAUTH_CLIENT_SECRET: CLIENT_SECRET } })));
++ assert.strictEqual(r['live-env'].status, 'FAIL');
++ assert.match(r['live-env'].detail, /started before OPENRE_URL, OPENRE_EVENTS_SECRET, OPENRE_PUBLIC_URL changed/);
++ assert.match(r['live-env'].detail, /deploy\.sh --restart --wait-idle/);
++ r = byId(await preflight(opts(dbs, ['--only', 'live-env']), host(dbs, { liveEnv: { ...env, OPENRE_EVENTS_SECRET: 'short', OPENRE_URL: 'http://127.0.0.1:4501' } })));
++ assert.match(r['live-env'].detail, /shorter than 32/);
++ assert.match(r['live-env'].detail, /OPENRE_URL = http:\/\/127\.0\.0\.1:4501, expected http:\/\/127\.0\.0\.1:4500/);
++ r = byId(await preflight(opts(dbs, ['--only', 'live-env']), host(dbs, { livePid: 0 })));
++ assert.match(r['live-env'].detail, /openvibe-live\.service is not running/);
++ assertNoSecret({ results: Object.values(r) });
++});
++
++t('events: wrong secret, disabled, duplicates, dead deliveries fail; waiting deliveries warn; secrets never printed', async () => {
++ let dbs = makeDbs();
++ addSubscription(dbs, { secret: 'x'.repeat(64) });
++ let r = byId(await preflight(opts(dbs, ['--only', 'events']), host(dbs)));
++ assert.strictEqual(r.events.status, 'FAIL');
++ assert.match(r.events.detail, /secret is not Live's OPENRE_EVENTS_SECRET/);
++ assertNoSecret({ results: Object.values(r) });
++
++ dbs = makeDbs();
++ addSubscription(dbs, { enabled: 0 });
++ addDelivery(dbs, 'dead', 2);
++ r = byId(await preflight(opts(dbs, ['--only', 'events']), host(dbs)));
++ assert.match(r.events.detail, /disabled/);
++ assert.match(r.events.detail, /2 dead deliveries/);
++
++ dbs = makeDbs();
++ addSubscription(dbs);
++ addSubscription(dbs, { id: 'sub_2' });
++ r = byId(await preflight(opts(dbs, ['--only', 'events']), host(dbs)));
++ assert.match(r.events.detail, /2 identical subscriptions/);
++
++ dbs = makeDbs();
++ addSubscription(dbs);
++ addDelivery(dbs, 'failed', 3);
++ r = byId(await preflight(opts(dbs, ['--only', 'events']), host(dbs)));
++ assert.strictEqual(r.events.status, 'WARN');
++ assert.match(r.events.detail, /3 deliveries waiting or retrying/);
++
++ r = byId(await preflight(opts(dbs, ['--only', 'events', '--events-db', '/nonexistent/events.db']), host(dbs)));
++ assert.match(r.events.detail, /cannot open \/nonexistent\/events\.db read-only/);
++});
++
++t('db: integrity_check and foreign keys from a read-only handle', async () => {
++ const dbs = makeDbs();
++ let r = byId(await preflight(opts(dbs, ['--only', 'db']), host(dbs)));
++ assert.strictEqual(r.db.status, 'PASS');
++ assert.match(r.db.detail, /stream_definitions 1, ingest_keys 0, ingest_sessions 0, destinations 0, migration_map 1/);
++ const corrupt = () => ({
++ prepare: (sql) => ({ all: () => (/integrity_check/.test(sql) ? [{ integrity_check: '*** in database main *** Page 7 is never used' }] : []), get: () => ({ n: 0 }) }),
++ close() {},
++ });
++ r = byId(await preflight(opts(dbs, ['--only', 'db']), host(dbs, { openDb: corrupt })));
++ assert.strictEqual(r.db.status, 'FAIL');
++ assert.match(r.db.detail, /Page 7 is never used/);
++ // the real handle really is read-only
++ const ro = new Database(dbs.openrePath, { readonly: true, fileMustExist: true });
++ assert.throws(() => ro.prepare("DELETE FROM migration_map").run(), /readonly/);
++ ro.close();
++});
++
++t('slot: offline RTMP slot with a subject and a migrated definition passes; live now / no subject fail', async () => {
++ const dbs = makeDbs();
++ let r = byId(await preflight(opts(dbs, ['--only', 'slot', '--slot', '12']), host(dbs)));
++ assert.strictEqual(r.slot.status, 'PASS', r.slot.detail);
++ assert.match(r.slot.detail, /slot 12 \(@rehearsal, rtmp\/obs\) authority live; OpenRe str_1 \(active, mirrored\); 0 destination\(s\)/);
++ const l = new Database(dbs.livePath);
++ l.prepare('INSERT INTO streams (managed_stream_id, is_live) VALUES (12, 1)').run();
++ l.prepare('DELETE FROM linked_accounts').run();
++ l.close();
++ r = byId(await preflight(opts(dbs, ['--only', 'slot', '--slot', '12']), host(dbs)));
++ assert.strictEqual(r.slot.status, 'FAIL');
++ assert.match(r.slot.detail, /live on Live right now/);
++ assert.match(r.slot.detail, /no canonical subject/);
++ r = byId(await preflight(opts(dbs, ['--only', 'slot', '--slot', '99']), host(dbs)));
++ assert.match(r.slot.detail, /no managed stream 99/);
++ r = byId(await preflight(opts(dbs, ['--only', 'slot']), host(dbs)));
++ assert.strictEqual(r.slot.status, 'SKIP');
++});
++
++t('arguments, ss parsing and the secret comparison', () => {
++ assert.throws(() => parseArgs(['--only', 'nope']), /unknown check nope/);
++ assert.throws(() => parseArgs(['--slot', 'abc']), /--slot/);
++ assert.throws(() => parseArgs(['--frobnicate']), /unknown argument/);
++ assert.deepStrictEqual(parseArgs(['--host-ip', '1.2.3.4,5.6.7.8', '--host-ip', '9.9.9.9']).hostIps, ['1.2.3.4', '5.6.7.8', '9.9.9.9']);
++ assert.deepStrictEqual(parseSsListeners('LISTEN 0 511 127.0.0.1:1936 0.0.0.0:*\nLISTEN 0 511 [::]:1936 [::]:*\nLISTEN 0 511 *:1936 *:*\nLISTEN 0 511 127.0.0.1:19360 0.0.0.0:*\n', 1936), ['127.0.0.1', '::', '*']);
++ assert.strictEqual(sameSecret('a'.repeat(40), 'a'.repeat(40)), true);
++ assert.strictEqual(sameSecret('a'.repeat(40), 'a'.repeat(39)), false);
++ assert.strictEqual(sameSecret('', ''), false);
++});
++
++t('rtmpProbe: a real socket — RTMP answer, a non-RTMP server, and a refused port', async () => {
++ const rtmp = net.createServer((s) => s.once('data', () => s.write(Buffer.concat([Buffer.from([0x03]), Buffer.alloc(1536)]))));
++ const other = net.createServer((s) => s.once('data', () => s.write('HTTP/1.1 400 Bad Request\r\n\r\n')));
++ await new Promise((r) => rtmp.listen(0, '127.0.0.1', r));
++ await new Promise((r) => other.listen(0, '127.0.0.1', r));
++ const a = await rtmpProbe('127.0.0.1', rtmp.address().port, 2000);
++ assert.deepStrictEqual([a.ok, a.rtmp], [true, true]);
++ const b = await rtmpProbe('127.0.0.1', other.address().port, 2000);
++ assert.deepStrictEqual([b.ok, b.rtmp], [true, false]);
++ const closed = await freePort();
++ const c = await rtmpProbe('127.0.0.1', closed, 2000);
++ assert.deepStrictEqual([c.ok, c.code], [false, 'ECONNREFUSED']);
++ rtmp.close();
++ other.close();
++});
++
++t.run();
ddiff --git a/test/set-definition-state.test.js b/test/set-definition-state.test.js
nnew file mode 100644
iindex 0000000..12f0343
---- /dev/null
++++ b/test/set-definition-state.test.js
@@@ -0,0 +1,72 @@
++'use strict';
++// scripts/set-definition-state.js (per-slot rollback) against a real openre-api: a stub Network
++// hands out Live's service token, the script finds the slot's definition by its live:managed_stream
++// reference and disables it; the key the broadcaster got is then refused at the RTMP handshake.
++// Re-enabling restores it. Also: unknown slot, no-op, and nothing secret printed.
++const assert = require('assert');
++const { run, parseArgs } = require('../scripts/set-definition-state');
++const { bootApi, request, serviceToken, suite, OWNER } = require('./helpers');
++
++const t = suite('set-definition-state');
++const CLIENT_SECRET = 'c'.repeat(48);
++const LIVE_CAPS = ['openre.stream.read', 'openre.stream.write', 'openre.key.rotate', 'openre.session.read'];
++let api;
++let env;
++let key;
++let streamId;
++const tokenRequests = [];
++
++/** fetch: Network's token endpoint is stubbed (Live's real grants), everything else is the real API. */
++async function fetchImpl(url, opts = {}) {
++ if (url === 'http://network.test/oauth/token') {
++ const p = new URLSearchParams(opts.body);
++ tokenRequests.push(Object.fromEntries(p));
++ const token = serviceToken(p.get('client_id'), LIVE_CAPS, { aud: p.get('audience') });
++ return new Response(JSON.stringify({ access_token: token, token_type: 'Bearer', expires_in: 300 }), { status: 200 });
++ }
++ return fetch(url, opts);
++}
++
++const logs = () => { const lines = []; const log = (s) => lines.push(String(s)); log.lines = lines; return log; };
++
++t('setup: a definition made for Live slot 12, as Live does at the switch', async () => {
++ api = await bootApi();
++ env = { OV_OAUTH_CLIENT_ID: 'live', OV_OAUTH_CLIENT_SECRET: CLIENT_SECRET, OV_NETWORK_INTERNAL_URL: 'http://network.test', OPENRE_URL: api.base };
++ const created = await request(api.base, 'POST', '/api/v1/streams', {
++ token: serviceToken('live', LIVE_CAPS),
++ headers: { 'X-OV-Subject': OWNER },
++ body: { title: 'Slot 12', mirror_to_live: true, external_refs: [{ service: 'live', type: 'managed_stream', id: '12' }] },
++ });
++ assert.strictEqual(created.status, 201);
++ streamId = created.body.stream.id;
++ key = created.body.key.key;
++ assert.ok(api.rt.store.definitions.resolveIngestKey(key, 'rtmp').definition);
++});
++
++t('--state disabled: the definition is disabled and its key refused at the handshake', async () => {
++ const log = logs();
++ const r = await run({ slot: '12', state: 'disabled', env, fetchImpl, log });
++ assert.deepStrictEqual(r, { id: streamId, state: 'disabled', changed: true, open_session: null });
++ assert.deepStrictEqual(tokenRequests.at(-1), { grant_type: 'client_credentials', client_id: 'live', client_secret: CLIENT_SECRET, audience: 'openvibe.openre' });
++ assert.deepStrictEqual(api.rt.store.definitions.resolveIngestKey(key, 'rtmp'), { error: 'stream_disabled' });
++ assert.match(log.lines.join('\n'), new RegExp(`${streamId} \\(live:managed_stream:12\\): active → disabled`));
++ for (const l of log.lines) { assert.ok(!l.includes(key)); assert.ok(!l.includes(CLIENT_SECRET)); }
++ const again = await run({ slot: '12', state: 'disabled', env, fetchImpl, log });
++ assert.strictEqual(again.changed, false);
++});
++
++t('--state active: the same key works again (re-switching the slot later)', async () => {
++ const r = await run({ slot: '12', state: 'active', env, fetchImpl, log: () => {} });
++ assert.strictEqual(r.state, 'active');
++ assert.ok(api.rt.store.definitions.resolveIngestKey(key, 'rtmp').definition);
++});
++
++t('a slot without a definition is an error; arguments are checked', async () => {
++ await assert.rejects(run({ slot: '99', state: 'disabled', env, fetchImpl, log: () => {} }), /no stream definition for live:managed_stream:99/);
++ assert.throws(() => parseArgs(['--slot', 'x', '--state', 'disabled']), /--slot/);
++ assert.throws(() => parseArgs(['--slot', '12', '--state', 'archived']), /--state must be active or disabled/);
++ assert.deepStrictEqual(parseArgs(['--slot', '12', '--state', 'disabled']), { liveEnv: '/etc/openvibe/live.env', slot: '12', state: 'disabled' });
++ await api.close();
++});
++
++t.run();
ddiff --git a/test/subscribe-live-events.test.js b/test/subscribe-live-events.test.js
nnew file mode 100644
iindex 0000000..fab0716
---- /dev/null
++++ b/test/subscribe-live-events.test.js
@@@ -0,0 +1,116 @@
++'use strict';
++// scripts/subscribe-live-events.js against a stub Network and Events (fetch is injected): Live's
++// token, the subscription created once with Live's secret, reported when it exists, --dry-run,
++// --disable/--enable for rollback, and no secret in anything it prints.
++const assert = require('assert');
++const { run, parseArgs, TOPIC, DEFAULT_ENDPOINT } = require('../scripts/subscribe-live-events');
++const { suite } = require('./helpers');
++
++const t = suite('subscribe-live-events');
++const SECRET = 's'.repeat(64);
++const CLIENT_SECRET = 'c'.repeat(48);
++const ENV = { OV_OAUTH_CLIENT_ID: 'live', OV_OAUTH_CLIENT_SECRET: CLIENT_SECRET, OV_NETWORK_INTERNAL_URL: 'http://network.test', EVENTS_URL: 'http://events.test/', OPENRE_EVENTS_SECRET: SECRET };
++
++function stub({ subscriptions = [], tokenStatus = 200, createStatus = 201 } = {}) {
++ const calls = [];
++ const subs = subscriptions.map((s) => ({ ...s }));
++ const reply = (status, body) => ({ status, ok: status < 300, text: async () => JSON.stringify(body) });
++ const fetchImpl = async (url, opts = {}) => {
++ const method = opts.method || 'GET';
++ calls.push({ method, url, headers: opts.headers || {}, body: opts.body || null });
++ if (url === 'http://network.test/oauth/token') {
++ const p = new URLSearchParams(opts.body);
++ assert.strictEqual(p.get('grant_type'), 'client_credentials');
++ assert.strictEqual(p.get('client_id'), 'live');
++ assert.strictEqual(p.get('audience'), 'openvibe.events');
++ return tokenStatus === 200 ? reply(200, { access_token: 'tok-live', token_type: 'Bearer', expires_in: 300 }) : reply(tokenStatus, { error: 'invalid_client' });
++ }
++ assert.strictEqual(opts.headers.Authorization, 'Bearer tok-live');
++ if (url === 'http://events.test/api/v1/subscriptions' && method === 'GET') return reply(200, { subscriptions: subs });
++ if (url === 'http://events.test/api/v1/subscriptions' && method === 'POST') {
++ const b = JSON.parse(opts.body);
++ if (createStatus === 409) return reply(409, { code: 'events.subscription_exists', subscription_id: 'sub_existing' });
++ const row = { id: 'sub_new', consumer: 'live', topic_pattern: b.topic_pattern, endpoint: b.endpoint, enabled: true };
++ subs.push(row);
++ return reply(201, { ...row, secret: b.secret });
++ }
++ const m = /^http:\/\/events\.test\/api\/v1\/subscriptions\/([^/]+)\/(enable|disable)$/.exec(url);
++ if (m && method === 'POST') {
++ const s = subs.find((x) => x.id === m[1]);
++ s.enabled = m[2] === 'enable';
++ return reply(200, s);
++ }
++ return reply(404, { code: 'not_found' });
++ };
++ return { fetchImpl, calls, subs };
++}
++
++const logs = () => { const lines = []; const log = (s) => lines.push(String(s)); log.lines = lines; return log; };
++const noSecret = (lines) => {
++ for (const l of lines) for (const s of [SECRET, CLIENT_SECRET, 'tok-live']) assert.ok(!l.includes(s), `printed a secret: ${l}`);
++};
++
++t('creates openre.session.* → Live\'s /internal/openre-events with Live\'s secret, as Live', async () => {
++ const s = stub({ subscriptions: [{ id: 'sub_other', topic_pattern: 'media.vod.*', endpoint: 'http://127.0.0.1:3000/internal/events', enabled: true }] });
++ const log = logs();
++ const r = await run({ env: ENV, fetchImpl: s.fetchImpl, log });
++ assert.deepStrictEqual(r, { action: 'create', subscription_id: 'sub_new', existed: false, created: true });
++ const post = s.calls.find((c) => c.method === 'POST' && c.url.endsWith('/api/v1/subscriptions'));
++ assert.deepStrictEqual(JSON.parse(post.body), { topic_pattern: TOPIC, endpoint: DEFAULT_ENDPOINT, secret: SECRET });
++ assert.strictEqual(TOPIC, 'openre.session.*');
++ assert.strictEqual(DEFAULT_ENDPOINT, 'http://127.0.0.1:3000/internal/openre-events');
++ assert.match(log.lines.join('\n'), /subscribed: sub_new \(openre\.session\.\* → http:\/\/127\.0\.0\.1:3000\/internal\/openre-events\)/);
++ noSecret(log.lines);
++});
++
++t('an existing subscription is reported, not duplicated (and a 409 race is handled)', async () => {
++ let s = stub({ subscriptions: [{ id: 'sub_1', topic_pattern: TOPIC, endpoint: DEFAULT_ENDPOINT, enabled: true }] });
++ let log = logs();
++ let r = await run({ env: ENV, fetchImpl: s.fetchImpl, log });
++ assert.deepStrictEqual(r, { action: 'create', subscription_id: 'sub_1', existed: true, created: false });
++ assert.ok(!s.calls.some((c) => c.method === 'POST' && c.url.endsWith('/subscriptions')), 'no second subscription');
++ assert.match(log.lines.join('\n'), /cutover-preflight\.js --only events/);
++ s = stub({ createStatus: 409 });
++ log = logs();
++ r = await run({ env: ENV, fetchImpl: s.fetchImpl, log });
++ assert.strictEqual(r.subscription_id, 'sub_existing');
++});
++
++t('--dry-run lists and changes nothing', async () => {
++ const s = stub();
++ const log = logs();
++ const r = await run({ action: 'list', env: ENV, fetchImpl: s.fetchImpl, log });
++ assert.strictEqual(r.existed, false);
++ assert.deepStrictEqual(s.calls.map((c) => c.method), ['POST', 'GET'], 'only the token and the list');
++ assert.match(log.lines.join('\n'), /would create one/);
++});
++
++t('--disable and --enable (rollback and its undo)', async () => {
++ const s = stub({ subscriptions: [{ id: 'sub_1', topic_pattern: TOPIC, endpoint: DEFAULT_ENDPOINT, enabled: true }] });
++ const log = logs();
++ let r = await run({ action: 'disable', env: ENV, fetchImpl: s.fetchImpl, log });
++ assert.deepStrictEqual(r, { action: 'disable', subscription_id: 'sub_1', enabled: false });
++ assert.strictEqual(s.subs[0].enabled, false);
++ r = await run({ action: 'enable', env: ENV, fetchImpl: s.fetchImpl, log });
++ assert.strictEqual(s.subs[0].enabled, true);
++ await assert.rejects(run({ action: 'disable', env: ENV, fetchImpl: stub().fetchImpl, log }), /no subscription openre\.session\.\* .* to disable/);
++ noSecret(log.lines);
++});
++
++t('refuses without a usable secret before calling anything; a refused token is an error', async () => {
++ const s = stub();
++ await assert.rejects(run({ env: { ...ENV, OPENRE_EVENTS_SECRET: 'short' }, fetchImpl: s.fetchImpl, log: () => {} }), /OPENRE_EVENTS_SECRET must be set/);
++ await assert.rejects(run({ env: { ...ENV, OV_OAUTH_CLIENT_SECRET: '' }, fetchImpl: s.fetchImpl, log: () => {} }), /OV_OAUTH_CLIENT_SECRET is not set/);
++ assert.strictEqual(s.calls.length, 0);
++ await assert.rejects(run({ env: ENV, fetchImpl: stub({ tokenStatus: 401 }).fetchImpl, log: () => {} }), /Network refused a token for live \(401 invalid_client\)/);
++});
++
++t('arguments', () => {
++ assert.strictEqual(parseArgs([]).action, 'create');
++ assert.strictEqual(parseArgs(['--dry-run']).action, 'list');
++ assert.strictEqual(parseArgs(['--disable']).action, 'disable');
++ assert.strictEqual(parseArgs(['--live-env', '/tmp/x.env']).liveEnv, '/tmp/x.env');
++ assert.throws(() => parseArgs(['--nope']), /unknown argument/);
++});
++
++t.run();