openvibers/openrestream Commits

Deploy (systemd units per process, worker generations as template instances, nginx for openre.stream, deploy.sh), .env.example, README with the per-protocol port status and cutover runbook, STATUS alpha, CI, capability and service manifest proposals, Live integration patch; contracts v0.13.0, shared v1.2.1; extra public RTMP ports

OP OpenVibers committed 23 Sept 2026 · 34dea5e270 · 25 files, +2143 −70


Changes

ddiff --git a/.env.example b/.env.example
nnew file mode 100644
iindex 0000000..15a464c
---- /dev/null
++++ b/.env.example
@@@ -0,0 +1,102 @@
++# ═══════════════════════════════════════════════════════════════
++# OpenRe.Stream — Environment Configuration
++# Copy to .env for development. In production this is /etc/openvibe/openre.env (0600), read by
++# every unit (openre-api, openre-session-coordinator, openre-rtmp-ingest@, openre-restream-worker@);
++# the units set NODE_ENV, PORT, OPENRE_RELEASE and OPENRE_DB_PATH themselves.
++# Names only here: never commit a value.
++# ═══════════════════════════════════════════════════════════════
++
++# ── openre-api ─────────────────────────────────────────────────
++PORT=4500
++HOST=127.0.0.1
++NODE_ENV=development
++#BASE_URL=https://openre.stream
++# Hops in front of Node that set X-Forwarded-For (Cloudflare → nginx → Node).
++#TRUST_PROXY=2
++#COOKIE_SECURE=true
++
++# ── Identity: OpenVibe.Network ────────────────────────────────
++OV_NETWORK_URL=https://openvibe.network
++OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000
++#OV_NETWORK_ISSUER=https://openvibe.network
++# Pin the Network public key instead of fetching /api/.well-known/jwks (PEM, \n escapes allowed).
++#OV_NETWORK_PUBLIC_KEY=
++# OAuth client `openre`: browser sign-in for the UI (redirect https://openre.stream/auth/callback)
++# and OpenRe's own client-credentials tokens (Events, Media).
++OV_OAUTH_CLIENT_ID=openre
++OV_OAUTH_CLIENT_SECRET=
++#OV_OAUTH_REDIRECT_URI=https://openre.stream/auth/callback
++# A browser's Network JWT is accepted when its aud contains one of these.
++#OPENRE_USER_AUDIENCES=openvibe.openre,openvibe.network
++
++# ── Store (SQLite, shared by every OpenRe process; WAL) ───────
++OPENRE_DB_PATH=./data/openre.db
++
++# ── Secrets ───────────────────────────────────────────────────
++# 32 bytes (64 hex chars or base64): encrypts restream destination keys and SRT passphrases at rest.
++# Required for destinations; generate with `openssl rand -hex 32`.
++OPENRE_SECRETS_KEY=
++# The previous key while rotating OPENRE_SECRETS_KEY (decryption only).
++#OPENRE_SECRETS_KEY_PREVIOUS=
++
++# ── RTMP ingest (openre-rtmp-ingest) ──────────────────────────
++# Public port. Live's own ingest keeps 1935 until its RTMP is retired (README "Port plan").
++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
++#OPENRE_RTMP_BIND=0.0.0.0
++# What encoders are told: rtmp://<host>[:<port>]/live
++#OPENRE_RTMP_PUBLIC_HOST=ingest.openre.stream
++#OPENRE_RTMP_PUBLIC_PORT=1936
++# Loopback ports each worker generation takes for its RTMP play + HTTP-FLV endpoints.
++#OPENRE_RTMP_INTERNAL_PORT_MIN=19360
++#OPENRE_RTMP_INTERNAL_PORT_MAX=19399
++#OPENRE_RTMP_CHUNK_SIZE=60000
++#OPENRE_RTMP_MAX_PUBLISHERS=64
++
++# ── Workers, leases, drain ────────────────────────────────────
++#OPENRE_WORKER_HEARTBEAT_MS=2000
++# A worker silent this long is lost: its sessions fail and its outputs move to another worker.
++#OPENRE_WORKER_LEASE_MS=15000
++# A draining generation keeps its sessions at most this long (default 24 h), then ends them.
++#OPENRE_DRAIN_MAX_MS=86400000
++#OPENRE_COORDINATOR_INTERVAL_MS=1000
++#OPENRE_RESTREAM_POLL_MS=1000
++
++# ── Restream outputs (openre-restream-worker) ─────────────────
++#OPENRE_FFMPEG_PATH=ffmpeg
++# Static OpenSSL ffmpeg used for rtmps:// destinations (as Live's bin/ffmpeg).
++#OPENRE_FFMPEG_OPENSSL_PATH=
++#OPENRE_MAX_DESTINATIONS=10
++# Allow destinations on loopback/private/link-local hosts (off: SSRF rule).
++#OPENRE_DEST_ALLOW_PRIVATE=false
++#OPENRE_OUTPUT_LOG_LIMIT=200
++#OPENRE_RESTART_BASE_MS=5000
++#OPENRE_RESTART_MAX_MS=120000
++#OPENRE_MAX_RESTARTS=30
++#OPENRE_RAPID_CRASH_MS=5000
++#OPENRE_RAPID_CRASH_GIVEUP=4
++#OPENRE_LIVE_ACK_TIMEOUT_MS=20000
++#OPENRE_STABLE_MS=30000
++#OPENRE_OUTPUT_START_DELAY_MS=3000
++
++# ── Events (OpenVibe.Events; relay runs in the coordinator) ───
++# Unset = relay off (rows still commit and wait). Grant: [openre, events.event.publish, openvibe.events].
++#EVENTS_URL=http://127.0.0.1:4300
++#EVENTS_PUBLISH=on
++#EVENTS_RELAY_INTERVAL_MS=1000
++
++# ── Recording requests (OpenVibe.Media) ───────────────────────
++MEDIA_URL=http://127.0.0.1:4100
++# The Media tenant recordings are filed under ('live' keeps them in the Live channel's gallery).
++MEDIA_APP_ID=live
++# 'key' = the tenant's app key below (what Media's VOD routes accept today); 'service' = a Network
++# service token (needs Media to name a capability on its VOD ingest routes first).
++#OPENRE_MEDIA_AUTH=key
++MEDIA_API_KEY=
++# OPENRE_RECORDING=off disables recording requests.
++#OPENRE_RECORDING=on
++#OPENRE_RECORDING_START_DELAY_MS=2000
++
++# ── Links ─────────────────────────────────────────────────────
++#OV_LIVE_URL=https://openvibe.live
ddiff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
nnew file mode 100644
iindex 0000000..2fd78a6
---- /dev/null
++++ b/.github/workflows/ci.yml
@@@ -0,0 +1,25 @@
++name: ci
++on:
++ push:
++ branches: [main]
++ pull_request:
++jobs:
++ test:
++ runs-on: ubuntu-latest
++ steps:
++ - uses: actions/checkout@v4
++ - uses: actions/setup-node@v4
++ with:
++ node-version: 22.22.1 # production runs Node 22 (>= 22.12 for SO_REUSEPORT)
++ cache: npm
++ # The RTMP end-to-end test publishes and restreams with a real ffmpeg.
++ - run: sudo apt-get update -qq && sudo apt-get install -y -qq ffmpeg
++ - run: npm ci --no-audit --no-fund
++ - run: npm test
++ - name: contract check (openvibe-contracts)
++ # openre's service manifest in openvibe-contracts is still the charter placeholder
++ # (contractRanges ^0.1.0, no capabilities) until the lead releases
++ # docs/service-manifest-proposal.json and docs/capabilities-proposal/. Remove
++ # continue-on-error in the commit that bumps openvibe-contracts to that release.
++ continue-on-error: true
++ run: npx openvibe-contracts-check --service openre --src server --src workers
ddiff --git a/README.md b/README.md
iindex 48464b0..0bd3567 100644
---- a/README.md
++++ b/README.md
@@@ -2,75 +2,236 @@
> Ingest and restream: stream definitions, keys, sessions, transport workers, outputs and output health.
--**Status:** placeholder — planning only, no runnable code yet.
--**Domain:** `openre.stream`
--**Plan:** OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §10 and §10.5.
++**Status:** alpha (roadmap Wave 7). Runs and is tested end to end with real RTMP; not deployed, and no broadcaster uses it yet. The domain keeps its placeholder page on OpenVibe.Sites until the launch rule below is met.
++**Domain:** `openre.stream` (UI + API), `ingest.openre.stream` (RTMP, DNS only)
++**Plan:** OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §10, §10.5, §15.10; ADR-009 (binding), ADR-004, ADR-006, ADR-007.
**License:** AGPL-3.0 (same as every OpenVibe service).
## Purpose
--The ingest/restream control plane and runtime extracted from OpenVibe.Live. Live keeps the creator/channel product and observes sessions through OpenRe APIs and events; transport workers survive Live deploys.
++The ingest/restream control plane and runtime extracted from OpenVibe.Live. Live keeps the creator/channel product and observes sessions through OpenRe APIs and events; transport workers survive Live deploys **and OpenRe API deploys**. This is the permanent fix for "restarting Live drops live RTMP streams" (hazard H1).
## Owns
--- stream input definitions and keys, ingest sessions, transport worker ownership and generations, output/restream destinations, output health/logs, recording requests to Media, optional mirroring into Live (explicit consent)
++- stream (input) definitions, owned by a canonical subject (`usr_…`), with typed references to other services' entities (`live:managed_stream:12`, `live:user:45`)
++- ingest keys: SHA-256 only, shown once at creation/rotation; rotation revokes the old key immediately or after a grace period (`openre.key.rotate`)
++- ingest sessions: `starting → live → ending → ended | failed`, with the owning worker, its generation and a lease
++- transport workers and their generations (who owns which session; new sessions reach the newest ready generation; old ones drain)
++- restream destinations (URL + key sealed with AES-256-GCM, never returned in full), outputs with health, progress and bounded logs, a circuit breaker per destination
++- recording *requests* to OpenVibe.Media, and the migration map from Live ids
++- the lifecycle events `openre.session.started|ended|failed`, `openre.output.healthy|failed`, `openre.recording.requested`, `openre.key.rotated`
## Does not own
--- channel presentation, discovery, watch pages (Live)
--- recording bytes and finalisation (Media)
--- chat
++- channel presentation, discovery, watch pages, the creator dashboard (Live)
++- recording bytes and finalisation (Media: it records, finalises, thumbnails and fires `vod.ready`)
++- chat, money, identity
--## Planned surfaces
++## Depends on
--- deployment units: `openre-api`, `openre-rtmp-ingest`, `openre-webrtc-ingest`, `openre-sfu`, `openre-restream-worker`, `openre-session-coordinator` (one repo)
--- standalone UI: definitions, keys/ingest URLs, destinations add/edit/delete/test, session lifecycle, output health
++- OpenVibe.Network (JWKS for service tokens and user JWTs; OAuth client `openre`; client-credentials tokens for Events/Media)
++- OpenVibe.Events (event relay; optional: rows wait in the outbox)
++- OpenVibe.Media (recording requests; optional: sessions work without it)
++- OpenVibe.Contracts (ids, problem+json, capability checks), OpenVibe.SDK v0.2.2 (outbox, token client), OpenVibe.Shared (chrome)
++- node-media-server 2.7.4 (the RTMP session code Live runs) and the system ffmpeg
--## Data (authority tables / families)
++## What is ported, per protocol
++
++| Transport | Live source | In OpenRe | State |
++|---|---|---|---|
++| RTMP ingest | `server/streaming/rtmp-server.js` (node-media-server) | `workers/rtmp-ingest.js` — same NMS session code; OpenRe's own keys; publish renamed to `/live/<session id>` so no key reaches a play URL; loopback RTMP play + HTTP-FLV per generation; SO_REUSEPORT drain | **ported, tested with real ffmpeg** |
++| Restream (RTMP source) | `server/streaming/restream-manager.js` | `workers/restream/*` — same ffmpeg arguments (codec copy from HTTP-FLV), Twitch→RTMPS, Kick `/app`, SRT options, live ACK by `-progress`, backoff, rapid-crash circuit breaker, destination cooldown | **ported, tested with real ffmpeg** |
++| Restream from JSMPEG / WebRTC sources | same | encoder args ported (`encodingArgs`), sources not (they need those ingests) | not ported |
++| Recording | `server/streaming/recorder.js` + `server/media-client.js` | `server/store/recordings.js` + `server/media-client.js` — same Media API v1 calls (create VOD, `ingest/rtmp`, finalize, delete shell / clips-only) driven by the coordinator; Media pulls the loopback play URL | **ported (RTMP)**, tested against a stub Media that really pulls the URL |
++| WHIP / WebRTC ingest | `whip-handler.js` (werift) | `workers/webrtc-ingest.js`: worker interface + coordinator registration only | **not ported** |
++| SFU | `webrtc-sfu.js` + `broadcast-server.js` (mediasoup) | `workers/sfu.js`: worker interface only | **not ported** |
++| JSMPEG | `jsmpeg-relay.js` | `workers/jsmpeg.js`: worker interface only | **not ported** |
++| OAuth-linked destinations (per go-live Twitch/YouTube key refresh, YouTube broadcast creation) | `restream-manager._refreshDestFromConnection` | no (Live owns platform OAuth); OpenRe pushes to the stored key | **not ported** |
++| Viewer counts from platforms, chat relay, PowerChat | restream-manager / integrations | no (product features, stay in Live) | not in scope |
--- see above (PostgreSQL for durable session metadata; Redis/etcd only for leases/routing)
++## Processes
--## Capabilities and events
++```
++ ┌───────────────── one SQLite database (WAL), /var/lib/openre/openre.db ─────────────────┐
++openre-api (4500) openre-session-coordinator openre-rtmp-ingest@<release> openre-restream-worker@<release>
++ API + UI + /play proxy leases, generations, drain, 0.0.0.0:1936 publish (reuseport) one ffmpeg per output, pulls the
++ definitions, keys, output assignment, recording 127.0.0.1 play + HTTP-FLV session's HTTP-FLV from its worker
++ destinations, end requests requests (Media), event relay per generation
++```
--- `openre.stream.*`, `openre.session.*`, `openre.output.*`, `openre.key.rotate`
++- **No process depends on another at runtime.** Workers need only the database. The API owns no transport; restarting it never touches a session (`test/api-restart.test.js`, `test/rtmp-e2e.test.js`).
++- **Generations.** Each worker process registers as generation N+1 of its kind. When a newer generation is `ready`, the coordinator marks older ones `draining` (deadline `OPENRE_DRAIN_MAX_MS`, default 24 h). A draining RTMP worker closes its public listener (the kernel sends new encoder connections to the new generation through the shared port) and keeps its publishers; a draining restream worker keeps its outputs. Idle → the process exits 0 by itself. At the deadline, remaining sessions are ended (encoders reconnect to the newest generation) and remaining outputs are handed over.
++- **Leases.** Every heartbeat (2 s) renews the worker and the lease of every session it owns in one transaction. A worker silent for `OPENRE_WORKER_LEASE_MS` (15 s) is `lost`: its sessions fail (`openre.session.failed`), its outputs go back to `pending` for another worker. A lost worker that wakes up drops its transports and exits.
++- **Isolation.** An output only ever changes its own row. A dead destination fails (`openre.output.failed`, destination cooldown 15 min → 24 h) and the session stays live (`test/rtmp-e2e.test.js`, `test/sessions.test.js`, `test/coordinator.test.js`).
++- **Store.** SQLite per ADR-007 (single host). Several writer processes is the trigger ADR-007 names for PostgreSQL; all SQL is in `server/store/*`, so that move replaces one layer. Redis/etcd are not used; leases live in the same database.
--Events: ``openre.session.started|ended``, ``openre.output.healthy|failed``, ``openre.recording.requested``
++## Running it
--## Depends on
++```bash
++npm install
++cp .env.example .env # set OPENRE_SECRETS_KEY (openssl rand -hex 32)
++npm run dev # openre-api on http://127.0.0.1:4500
++npm run coordinator # in other terminals
++npm run rtmp-ingest
++npm run restream-worker
++npm test # every test/*.test.js; the RTMP e2e test needs ffmpeg on PATH
++```
++
++Node 22 (≥ 22.12 for `reusePort`): `fnm exec --using=22.22.1 npm test`.
++
++## Auth
++
++Services call with an OpenVibe.Network client-credentials token (audience `openvibe.openre`), verified offline against the Network JWKS; each route checks one capability. A service may send `X-OV-Subject: usr_…` to act for one owner (it is then limited to that owner's streams). Browsers/owners use the Network user JWT (`ov_token` cookie from `/auth/login`, or Bearer) and act on their own streams; Network role `admin` is staff (read all, end sessions).
++
++| Capability | Routes |
++|---|---|
++| `openre.stream.read` | `GET /api/v1/streams[?external_ref=live:managed_stream:12]`, `GET /api/v1/streams/:id`, `GET /api/v1/streams/:id/keys` |
++| `openre.stream.write` | `POST /api/v1/streams` (first key, shown once), `PATCH`/`DELETE /api/v1/streams/:id` |
++| `openre.key.rotate` | `POST /api/v1/streams/:id/keys/rotate {grace_seconds, end_sessions}` (new key, shown once) |
++| `openre.session.read` | `GET /api/v1/sessions`, `GET /api/v1/sessions/:id`, `GET /api/v1/sessions/:id/playback`, `GET /api/v1/workers` |
++| `openre.session.end` | `POST /api/v1/sessions/:id/end` |
++| `openre.output.read` | `GET /api/v1/streams/:id/destinations`, `GET /api/v1/sessions/:id/outputs`, `GET /api/v1/destinations/:id/logs`, `GET /api/v1/outputs/:id/logs` |
++| `openre.output.write` | `POST /api/v1/streams/:id/destinations`, `PATCH`/`DELETE /api/v1/destinations/:id`, `POST /api/v1/destinations/:id/test|start|stop` |
++
++The ids are proposed in [docs/capabilities-proposal/](docs/capabilities-proposal/) with the service manifest ([docs/service-manifest-proposal.json](docs/service-manifest-proposal.json)); they are not in `openvibe-contracts` yet. Until that release, `server/auth/index.js` grants them with the contracts rule (exact id or `family.*`) and defers to `capabilities.check()` once contracts know the id. Errors are RFC 9457 problem+json with a stable `code`.
++
++**Grants the lead adds in Network** (`[client, capability, audience]`):
++
++- `[live, openre.stream.read, openvibe.openre]`, `[live, openre.stream.write, openvibe.openre]`, `[live, openre.key.rotate, openvibe.openre]`, `[live, openre.session.read, openvibe.openre]`
++- `[openre, events.event.publish, openvibe.events]`
++- `[live, events.subscription.manage, openvibe.events]` if Live creates its own `openre.session.*` subscription (or an operator creates it)
++- Media, only when `OPENRE_MEDIA_AUTH=service`: `[openre, media.object.upload, openvibe.media]` with namespace `live` — and Media must first name that capability on its VOD create/ingest/finalize/delete routes (today they accept only the tenant app key; see "Recording").
++- OAuth client `openre` (authorization code + client credentials) with redirect `https://openre.stream/auth/callback`.
++
++## Events
++
++Written to `event_outbox` in the same SQLite transaction as the change (openvibe-sdk `createOutbox`); the coordinator's relay publishes them with OpenRe's service token. Envelopes validate against `events.event-envelope@1`; `visibility: internal`; no payload ever carries an ingest key or a destination key.
++
++| Type | Subject | When |
++|---|---|---|
++| `openre.session.started` | `ingest_session` | a session reached live (payload: owner, protocol, worker + generation, `external_refs`, `mirror_to_live`, playback descriptor) |
++| `openre.session.ended` | `ingest_session` | a session that had been live ended (duration, reason) |
++| `openre.session.failed` | `ingest_session` | worker lost, lease expired, … (`was_live`, reason) |
++| `openre.output.healthy` / `openre.output.failed` | `output` | an output confirmed live / gave up (error, cooldown) |
++| `openre.recording.requested` | `recording` | Media accepted the recording request (Media app + VOD id) |
++| `openre.key.rotated` | `stream` | new key id + hint, retired key ids, grace end |
++
++## Playback
++
++`GET /api/v1/sessions/:id/playback` returns a descriptor: `flv.internal_url` (loopback HTTP-FLV on the worker that holds the session — for services on this host), `flv.public_url` (`https://openre.stream/play/<session>.flv`, proxied by the API for non-private streams), `rtmp.internal_url` (loopback RTMP play, what Media records from). `webrtc` and `hls` are `null` (not produced). Live's player keeps using its own `/api/streams/rtmp-proxy/:id.flv`, which for an OpenRe session proxies `flv.internal_url` (see the Live patch).
++
++## Recording
++
++When a session is live and its definition says `recording_mode` `vod` or `clips`, the coordinator asks Media, exactly as Live's recorder does: `POST /api/v1/<MEDIA_APP_ID>/vods` (title, Live `user_id`/`managed_stream_id` from the typed references so the VOD lands in the Live channel's gallery, visibility, `meta.openre_session_id`), then `POST …/vods/:id/ingest/rtmp { rtmp_url: rtmp://127.0.0.1:<play port>/live/<session id> }`. When the session ends: `POST …/finalize` (clips-only recordings are then deleted, as Live does). A disk-space refusal deletes the empty VOD shell and retries every 5 min while the session is live. Media finalises. Auth today is the Media tenant key (`MEDIA_API_KEY`, the `live` app's key — the interim compromise, as Live holds it too); the target is `OPENRE_MEDIA_AUTH=service` once Media accepts service tokens on those routes. Known gap: the VOD's `stream_id` (a Live id) is not set, because Live creates its `streams` row from the event; Live links the VOD by slot, user and time.
++
++## Security notes
++
++- Ingest keys: 256-bit random, `ork_` + 43 base64url characters, SHA-256 at rest, compared by indexed lookup inside the RTMP handshake. Never logged (only the last four characters), never in an event, never in a URL after the handshake (the publish is renamed to the session id).
++- Destination keys and SRT passphrases: AES-256-GCM (`OPENRE_SECRETS_KEY`, rotation via `OPENRE_SECRETS_KEY_PREVIOUS`); APIs return `****` + last four only; ffmpeg command lines are logged redacted.
++- SSRF (plan anti-goal 13): destination URLs must be `rtmp://`, `rtmps://` or `srt://` with a hostname, no credentials, and no loopback/private/link-local/CGNAT/multicast host — checked on write and again, with DNS resolution, right before ffmpeg starts (`OPENRE_DEST_ALLOW_PRIVATE` opts out, e.g. a LAN box). Live allowed private hosts; the migration holds such destinations with the reason recorded.
++- The RTMP public listener accepts publish only; play is refused there. Loopback play/FLV ports accept only `/live/ses_<ULID>` paths.
++- UI forms carry a CSRF token derived from the session cookie; cookies are SameSite=Lax.
++
++## Port plan
++
++| Port | Owner | Until |
++|---|---|---|
++| 1935/tcp | Live's in-process RTMP (node-media-server) | Live's RTMP ingest is retired (after the last slot has run on OpenRe for two weeks, ADR-009) |
++| 9935/tcp (loopback) | Live's HTTP-FLV | same |
++| **1936/tcp** | OpenRe `openre-rtmp-ingest` (all generations, SO_REUSEPORT) | permanent: URLs handed out as `rtmp://ingest.openre.stream:1936/live` keep working |
++| 19360–19399/tcp (loopback) | OpenRe per-generation RTMP play + HTTP-FLV | permanent |
++| 4500/tcp (loopback) | `openre-api` | permanent |
++
++The switch to 1935: once Live no longer listens on 1935, set `OPENRE_RTMP_EXTRA_PORTS=1935` and deploy a worker generation (`deploy.sh workers`); from then on `rtmp://ingest.openre.stream/live` works too, and 1936 keeps working. `ingest.openre.stream` is a DNS-only record (RTMP cannot go through Cloudflare's proxy); open 1936/tcp at the host firewall and the provider edge.
++
++## Deploying
++
++Layout: `/opt/openre.stream/releases/<sha>` (full checkouts), `current` → the release the API and coordinator run, env `/etc/openvibe/openre.env` (see [.env.example](.env.example)), store `/var/lib/openre/openre.db`, units in [deploy/systemd/](deploy/systemd/), nginx [deploy/nginx/openre.stream.conf](deploy/nginx/openre.stream.conf), script [deploy/scripts/deploy.sh](deploy/scripts/deploy.sh):
++
++```bash
++sudo deploy/scripts/deploy.sh release origin/main # new release dir + npm ci
++sudo deploy/scripts/deploy.sh api # restarts openre-api + coordinator ONLY
++sudo deploy/scripts/deploy.sh workers # starts a new worker generation; old ones drain
++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.
++- 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)
++
++[docs/live-patch.diff](docs/live-patch.diff) is the Live side, made against OpenVibe.Live `c384787` and verified there (applies with `git apply`; Live's `npm test` 62/62 on Node 22.22.1). **Switch off (the default) changes nothing**: with `OPENRE_URL` unset, or a slot on `'live'`, every changed code path returns what it returned before (`test/openre-switch.test.js` in the patch). It adds:
++
++- `server/openre/openre-client.js` — service-token client for OpenRe (streams by `external_ref`, create, rotate, sessions, playback descriptors).
++- `managed_streams.ingest_authority` (`'live'` default | `'openre'`) + `openre_stream_id`, and an `openre_sessions` projection table (additive, in `initDb`).
++- Live's RTMP `prePublish` refuses a slot key when the slot is on `'openre'` (a stream can never be ingested twice). Only RTMP moves; WHIP/browser/JSMPEG for that slot stay on Live.
++- Go Live / stream-key UI for `'openre'` slots: `GET /api/streams/managed` returns the slot without its Live key; `GET /api/streams/managed/:id/profile` and `GET /api/streams/:id/endpoint` show OpenRe's RTMP URL and a key hint (never a key); `POST /api/streams/managed/:id/regenerate-key` rotates on OpenRe and shows the new key once; `GET /api/streams/:id/rtmp-status` reads the mirror. Small fallbacks in `broadcast.js` / `broadcast-workspace.js` display the hint.
++- `POST /internal/openre-events` — OpenVibe.Events delivery endpoint (SDK `parseDelivery` + `createInbox`, consumer `live-openre-mirror`): `openre.session.started` creates or attaches the slot's `streams` row (control config, go-live notifications, Live's own `live.stream.started`), `ended`/`failed` end it; revision-ordered, exactly once; only for switched slots whose definition has `mirror_to_live` (consent). A 30 s reconcile asks OpenRe about mirrored live sessions (heartbeat refresh; ends rows OpenRe ended). Live's stale-stream cleanup skips mirrored sessions OpenRe confirmed in the last 30 min, and Live's boot-time restream resume skips them.
++- Playback: `GET /api/streams/rtmp-proxy/:id.flv` proxies the session's `flv.internal_url` from OpenRe's playback descriptor (validated as a loopback `…/live/ses_….flv` URL), so Live's player is unchanged.
++- Restream routes refuse (409 + `manage_url`) edits/start/stop for destinations of `'openre'` slots: those restreams run and are managed on openre.stream.
++- Admin: `GET /api/admin/openre/status`, `PUT /api/admin/openre/managed/:id/ingest-authority {authority, force}` — `'openre'` needs the slot offline, an RTMP slot (or `force`), the owner's canonical subject; it finds or creates the OpenRe definition and, in one transaction, flips the slot and rotates Live's own key for it. `'live'` flips back.
++- Env: `OPENRE_URL`, `OPENRE_PUBLIC_URL`, `OPENRE_EVENTS_SECRET` (in Live's `.env.example`). Unsetting `OPENRE_URL` is the emergency rollback for every switched slot at once.
++
++Not in the patch (follow-ups): live thumbnails, AI audio/vision taps and the RobotStreamer publisher for OpenRe sessions (they read Live's local HTTP-FLV by key; they can read `flv.internal_url` instead); Stream Manager destination editing through OpenRe's API (today: a link to openre.stream); platform viewer counts for OpenRe outputs.
++
++## 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`).
++
++## Cutover runbook
++
++ADR-009: one protocol at a time (RTMP → WHIP → JSMPEG → SFU), per slot, behind the switch, in a maintenance window agreed with the broadcaster; every key that existed before is rotated at the RTMP cutover; Live's ingest code stays until the last protocol has run on OpenRe for two weeks.
++
++### RTMP (ready)
++
++Once, before the first slot:
++
++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.
--- OpenVibe.Network
--- OpenVibe.Media
--- OpenVibe.Events
--- OpenVibe.Contracts
++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).
--## Acceptance (must be true before "done")
++### WHIP, JSMPEG, SFU (not ready)
--- OpenRe ingests and restreams without visiting Live
--- deploying Live's web/API during a broadcast does not interrupt transport or recording
--- deploying the OpenRe API does not terminate worker-owned transports (new sessions route to the newest ready generation; old workers drain)
--- destination failure never terminates the source session
++Each needs its transport ported into its worker (`workers/webrtc-ingest.js`, `workers/jsmpeg.js`, `workers/sfu.js` have the interface and registration), a restream source for it, a playback descriptor (`webrtc`), recording (RTP to Media for WebRTC; JSMPEG has no Media ingest), and `OPENRE_PROTOCOLS` in Live's `server/openre/authority.js` extended. The same per-slot switch, window and rollback apply.
--## Bootstrap / extraction source
++## Acceptance
--Live's mediasoup/werift/Node-Media-Server/FFmpeg ingest and restream code — the change is lifecycle and ownership, one protocol at a time.
++| Criterion (ADR-009 / plan W7) | Evidence |
++|---|---|
++| OpenRe ingests and restreams without visiting Live | `test/rtmp-e2e.test.js`: real ffmpeg publish → OpenRe worker → ffmpeg restream to an RTMP sink; no Live process involved |
++| Deploying the OpenRe API does not end worker-owned transports | `test/api-restart.test.js` (API process SIGTERM + restart while a worker process holds a live session; lease keeps renewing) and `test/rtmp-e2e.test.js` (same with a real encoder, restream and FLV playback afterwards) |
++| New sessions route to the newest ready generation; old workers drain | `test/rtmp-e2e.test.js` (second RTMP worker process: new publish on generation 2, generation 1 keeps its session, closes its listener and exits 0 when it ends), `test/coordinator.test.js` |
++| A destination failure never terminates the source session | `test/rtmp-e2e.test.js` (dead destination fails, session and encoder stay), `test/sessions.test.js`, `test/coordinator.test.js` (lost restream worker) |
++| Recording finalisation stays in Media | `test/recording-events.test.js`, `test/rtmp-e2e.test.js` (Media is asked to create/ingest/finalise; Media's ffmpeg pulls the loopback URL) |
++| Every stream key that existed before is rotated | `test/migration.test.js` (old keys never imported; new keys unseen), Live patch rotates Live's own slot key at the switch; checklist step for personal keys |
++| Deploying Live during a broadcast does not interrupt transport or recording | by construction (no OpenRe process talks to Live); proven for real only after the Live patch is deployed — not claimed here |
++| Switch off = no Live behaviour change | Live's full `npm test` with the patch applied, plus `test/openre-switch.test.js` |
## Launch rule
This repository does not make the product real, and the domain keeps its placeholder page on
[OpenVibers/OpenVibe.Sites](https://github.com/OpenVibers/OpenVibe.Sites) until all of the
--following exist here (plan §12.12):
--
--1. an owning runtime with health/readiness endpoints and observability;
--2. canonical identity/auth integration (OpenVibe.Network subjects, scoped service principals);
--3. server-rendered or static public routes that are useful without JavaScript;
--4. real persistence and end-to-end workflows;
--5. capability and event registration against `OpenVibe.Contracts`;
--6. a migration/seed strategy, a security/threat review, and sitemap/robots/feed behaviour;
--7. acceptance tests proving the advertised functionality.
--
--The launch release removes the domain from `OpenVibe.Sites/sites.json`, switches routing and
--registers maturity in the ecosystem registry atomically. A placeholder is never counted as an
--implemented service.
++following exist here (plan §12.12): an owning runtime with health/readiness endpoints and
++observability; canonical identity/auth integration; server-rendered public routes useful without
++JavaScript; real persistence and end-to-end workflows; capability and event registration against
++`OpenVibe.Contracts`; a migration/seed strategy, a security/threat review, and
++sitemap/robots/feed behaviour; acceptance tests proving the advertised functionality. Today the
++first three and the last are in place; capability registration, a security review and the
++production deploy are not.
---
ddiff --git a/STATUS.json b/STATUS.json
iindex 370d438..d47a6be 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -1,7 +1,24 @@
{
"repository": "OpenVibers/OpenRe.Stream",
-- "stage": "placeholder",
-- "code": false,
++ "stage": "alpha",
++ "code": true,
++ "wave": "W7",
"plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026)",
-- "updated": "2026-09-21"
++ "runtime": "node 22, express, better-sqlite3, node-media-server 2.7.4, ffmpeg",
++ "port": 4500,
++ "rtmpPort": 1936,
++ "domain": "openre.stream",
++ "contracts": "openvibe-contracts v0.13.0",
++ "protocols": {
++ "rtmp": "ingest ported (openre-rtmp-ingest)",
++ "restream": "RTMP source to RTMP/RTMPS/SRT destinations ported (openre-restream-worker)",
++ "whip": "not ported (worker interface only)",
++ "webrtc-sfu": "not ported (worker interface only)",
++ "jsmpeg": "not ported (worker interface only)"
++ },
++ "capabilities": ["openre.stream.read", "openre.stream.write", "openre.key.rotate", "openre.session.read", "openre.session.end", "openre.output.read", "openre.output.write"],
++ "capabilitiesRegistered": false,
++ "livePatch": "docs/live-patch.diff (not applied)",
++ "deployed": false,
++ "updated": "2026-09-22"
}
ddiff --git a/deploy/nginx/openre.stream.conf b/deploy/nginx/openre.stream.conf
nnew file mode 100644
iindex 0000000..8b59a46
---- /dev/null
++++ b/deploy/nginx/openre.stream.conf
@@@ -0,0 +1,97 @@
++## ═══════════════════════════════════════════════════════════════
++## openre.stream — Nginx Configuration (REFERENCE)
++##
++## Deployed to /etc/nginx/sites-available/openre.stream.conf — keep them in sync and reload
++## nginx after changing it. Certificate: openre.stream is its own domain (not under the
++## openvibe.network wildcard); issue one with certbot for openre.stream before enabling.
++##
++## openre-api (port 4500, unit openre-api) serves:
++## / the standalone UI (server-rendered; works without JavaScript)
++## /auth/* OpenVibe.Network sign-in (OAuth client `openre`)
++## /api/v1/* the API (service tokens or the owner's Network JWT on every route)
++## /api/health, /api/ready
++## /play/<session>.flv HTTP-FLV playback of a live, non-private session (long-lived)
++##
++## RTMP does NOT go through nginx or Cloudflare: ingest.openre.stream is a DNS-only (grey cloud)
++## A record to this host, and the rtmp-ingest workers listen on TCP 1936 themselves (README
++## "Port plan"). Open 1936/tcp in the host firewall and at the provider edge.
++## ═══════════════════════════════════════════════════════════════
++
++limit_req_zone $binary_remote_addr zone=openre_api:10m rate=20r/s;
++limit_conn_zone $binary_remote_addr zone=openre_play:10m;
++
++server {
++ listen 80;
++ listen [::]:80;
++ server_name openre.stream www.openre.stream;
++ location /.well-known/acme-challenge/ { root /var/www/certbot; }
++ location / { return 301 https://openre.stream$request_uri; }
++}
++
++server {
++ listen 443 ssl;
++ listen [::]:443 ssl;
++ http2 on;
++ server_name www.openre.stream;
++ ssl_certificate /etc/letsencrypt/live/openre.stream/fullchain.pem;
++ ssl_certificate_key /etc/letsencrypt/live/openre.stream/privkey.pem;
++ return 301 https://openre.stream$request_uri;
++}
++
++server {
++ listen 443 ssl;
++ listen [::]:443 ssl;
++ http2 on;
++ server_name openre.stream;
++
++ ssl_certificate /etc/letsencrypt/live/openre.stream/fullchain.pem;
++ ssl_certificate_key /etc/letsencrypt/live/openre.stream/privkey.pem;
++ ssl_protocols TLSv1.2 TLSv1.3;
++
++ access_log /var/log/nginx/openre.stream.access.log;
++ error_log /var/log/nginx/openre.stream.error.log;
++
++ client_max_body_size 256k;
++
++ add_header X-Content-Type-Options nosniff always;
++ add_header Referrer-Policy strict-origin-when-cross-origin always;
++ add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
++
++ # Live playback: no buffering, long reads. Viewers reconnect across an API deploy; the
++ # ingest (a separate worker process) is never affected.
++ location /play/ {
++ limit_conn openre_play 8;
++ proxy_pass http://127.0.0.1:4500;
++ proxy_http_version 1.1;
++ proxy_set_header Connection "";
++ proxy_set_header Host $host;
++ proxy_set_header X-Real-IP $remote_addr;
++ proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
++ proxy_set_header X-Forwarded-Proto $scheme;
++ proxy_buffering off;
++ proxy_cache off;
++ gzip off;
++ proxy_read_timeout 12h;
++ send_timeout 12h;
++ }
++
++ location /api/ {
++ limit_req zone=openre_api burst=40 nodelay;
++ proxy_pass http://127.0.0.1:4500;
++ proxy_http_version 1.1;
++ proxy_set_header Host $host;
++ proxy_set_header X-Real-IP $remote_addr;
++ proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
++ proxy_set_header X-Forwarded-Proto $scheme;
++ proxy_set_header traceparent $http_traceparent;
++ }
++
++ location / {
++ proxy_pass http://127.0.0.1:4500;
++ proxy_http_version 1.1;
++ proxy_set_header Host $host;
++ proxy_set_header X-Real-IP $remote_addr;
++ proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
++ proxy_set_header X-Forwarded-Proto $scheme;
++ }
++}
ddiff --git a/deploy/scripts/deploy.sh b/deploy/scripts/deploy.sh
nnew file mode 100755
iindex 0000000..a772472
---- /dev/null
++++ b/deploy/scripts/deploy.sh
@@@ -0,0 +1,110 @@
++#!/usr/bin/env bash
++# ═══════════════════════════════════════════════════════════════
++# OpenRe.Stream deploy (run on the host as root; the lead runs it, never CI).
++#
++# deploy/scripts/deploy.sh release [<git-ref>] check out <ref> (default origin/main) into
++# /opt/openre.stream/releases/<sha>, npm ci
++# deploy/scripts/deploy.sh api [<sha>] point `current` at a release, restart
++# openre-api + openre-session-coordinator.
++# NEVER touches a transport worker.
++# deploy/scripts/deploy.sh workers [<sha>] start a new generation of both workers from a
++# release; older generations drain and exit on
++# their own (sessions are not interrupted)
++# deploy/scripts/deploy.sh status units, generations, open sessions
++# deploy/scripts/deploy.sh prune remove releases no unit uses (keeps 5)
++#
++# Layout: /opt/openre.stream/releases/<sha>/ (full checkout + node_modules), `current` symlink
++# for the API and coordinator, one openre-rtmp-ingest@<sha> / openre-restream-worker@<sha>
++# instance per worker generation. Env: /etc/openvibe/openre.env. Store: /var/lib/openre/openre.db.
++# ═══════════════════════════════════════════════════════════════
++set -euo pipefail
++
++ROOT=/opt/openre.stream
++REPO=https://github.com/OpenVibers/OpenRe.Stream.git
++API=http://127.0.0.1:4500
++NODE_BIN=${NODE_BIN:-node}
++
++say() { printf '[openre-deploy] %s\n' "$*" >&2; }
++die() { say "ERROR: $*"; exit 1; }
++
++latest_release() { ls -1t "$ROOT/releases" 2>/dev/null | head -1; }
++
++cmd_release() {
++ local ref=${1:-origin/main}
++ mkdir -p "$ROOT/releases" "$ROOT/repo"
++ if [ ! -d "$ROOT/repo/.git" ]; then git clone --quiet "$REPO" "$ROOT/repo"; fi
++ git -C "$ROOT/repo" fetch --quiet --tags origin
++ local sha
++ sha=$(git -C "$ROOT/repo" rev-parse --short=12 "$ref")
++ local dir="$ROOT/releases/$sha"
++ if [ -d "$dir" ]; then say "release $sha already exists"; echo "$sha"; return; fi
++ git -C "$ROOT/repo" worktree add --detach --quiet "$dir" "$sha"
++ (cd "$dir" && npm ci --omit=dev --no-audit --no-fund --quiet)
++ chown -R ubuntu:ubuntu "$dir"
++ say "release $sha ready at $dir"
++ echo "$sha"
++}
++
++wait_ready() {
++ for _ in $(seq 1 60); do
++ if curl -sf --max-time 2 "$API/api/ready" >/dev/null; then return 0; fi
++ sleep 1
++ done
++ return 1
++}
++
++cmd_api() {
++ local sha=${1:-$(latest_release)}
++ [ -n "$sha" ] && [ -d "$ROOT/releases/$sha" ] || die "no release $sha"
++ local prev
++ prev=$(readlink "$ROOT/current" 2>/dev/null || true)
++ ln -sfn "$ROOT/releases/$sha" "$ROOT/current.new" && mv -Tf "$ROOT/current.new" "$ROOT/current"
++ say "current -> $sha (was ${prev:-none}); restarting openre-api and openre-session-coordinator only"
++ systemctl restart openre-api.service
++ systemctl restart openre-session-coordinator.service
++ if ! wait_ready; then
++ say "openre-api not ready after 60 s; rolling back to ${prev:-nothing}"
++ if [ -n "$prev" ]; then
++ ln -sfn "$prev" "$ROOT/current.new" && mv -Tf "$ROOT/current.new" "$ROOT/current"
++ systemctl restart openre-api.service openre-session-coordinator.service
++ fi
++ exit 2
++ fi
++ say "openre-api ready on $sha"
++}
++
++cmd_workers() {
++ local sha=${1:-$(latest_release)}
++ [ -n "$sha" ] && [ -d "$ROOT/releases/$sha" ] || die "no release $sha"
++ say "starting worker generation from release $sha (older generations drain by themselves)"
++ systemctl enable --now "openre-rtmp-ingest@$sha.service" "openre-restream-worker@$sha.service"
++ # Disable (not stop) older instances so a reboot does not bring them back; they exit on their own.
++ for unit in $(systemctl list-units --plain --no-legend 'openre-rtmp-ingest@*' 'openre-restream-worker@*' | awk '{print $1}'); do
++ case "$unit" in *"@$sha.service") ;; *) systemctl disable "$unit" >/dev/null 2>&1 || true; say "draining: $unit";; esac
++ done
++}
++
++cmd_status() {
++ systemctl --no-pager --plain list-units 'openre-*' || true
++ curl -s --max-time 3 "$API/api/ready" | "$NODE_BIN" -e 'let s="";process.stdin.on("data",d=>s+=d).on("end",()=>{try{const r=JSON.parse(s);console.log(JSON.stringify({status:r.status,workers:r.workers,coordinator:r.coordinator,events:r.events},null,2))}catch{console.log(s)}})' || true
++}
++
++cmd_prune() {
++ local keep used
++ used=$( (readlink "$ROOT/current"; systemctl list-units --plain --no-legend 'openre-*@*' | awk '{print $1}' | sed -E 's/.*@(.*)\.service/\1/') | xargs -n1 basename 2>/dev/null | sort -u)
++ keep=$(ls -1t "$ROOT/releases" | head -5)
++ for r in $(ls -1 "$ROOT/releases"); do
++ if echo "$used $keep" | grep -qw "$r"; then continue; fi
++ say "removing release $r"
++ git -C "$ROOT/repo" worktree remove --force "$ROOT/releases/$r"
++ done
++}
++
++case "${1:-}" in
++ release) shift; cmd_release "$@" ;;
++ api) shift; cmd_api "$@" ;;
++ workers) shift; cmd_workers "$@" ;;
++ status) cmd_status ;;
++ prune) cmd_prune ;;
++ *) sed -n '2,20p' "$0"; exit 1 ;;
++esac
ddiff --git a/deploy/systemd/openre-api.service b/deploy/systemd/openre-api.service
nnew file mode 100644
iindex 0000000..cb2a9a9
---- /dev/null
++++ b/deploy/systemd/openre-api.service
@@@ -0,0 +1,31 @@
++[Unit]
++Description=OpenRe.Stream API and UI (openre-api): definitions, keys, destinations, sessions, playback proxy
++# The API owns no transport. Restarting it (every API deploy) never touches the transport
++# workers: they are separate units with no dependency on this one, in either direction.
++After=network.target
++Wants=openvibe-network.service
++
++[Service]
++Type=simple
++User=ubuntu
++Group=ubuntu
++WorkingDirectory=/opt/openre.stream/current
++EnvironmentFile=/etc/openvibe/openre.env
++Environment=NODE_ENV=production
++Environment=PORT=4500
++Environment=OPENRE_DB_PATH=/var/lib/openre/openre.db
++Environment=PATH=/usr/local/bin:/usr/bin:/bin
++ExecStart=/usr/bin/env node server/index.js
++Restart=always
++RestartSec=3
++KillSignal=SIGTERM
++TimeoutStopSec=15
++NoNewPrivileges=true
++PrivateTmp=true
++ProtectSystem=strict
++ProtectHome=true
++StateDirectory=openre
++ReadWritePaths=/var/lib/openre
++
++[Install]
++WantedBy=multi-user.target
ddiff --git a/deploy/systemd/[email protected] b/deploy/systemd/[email protected]
nnew file mode 100644
iindex 0000000..3d3694a
---- /dev/null
++++ b/deploy/systemd/[email protected]
@@@ -0,0 +1,37 @@
++[Unit]
++Description=OpenRe.Stream restream worker (ffmpeg per destination), release %i
++# One instance per release (instance name = the release directory). A worker deploy STARTS a new
++# instance; the coordinator gives it every new output and drains the older generation, whose
++# outputs run until their sessions end (or OPENRE_DRAIN_MAX_MS: then they move over) before it exits 0.
++# Nothing here depends on openre-api or openre-session-coordinator: an API deploy never restarts
++# this unit. Never `systemctl restart` it during a broadcast; see README "Deploying".
++After=network.target
++
++[Service]
++Type=simple
++User=ubuntu
++Group=ubuntu
++WorkingDirectory=/opt/openre.stream/releases/%i
++EnvironmentFile=/etc/openvibe/openre.env
++Environment=NODE_ENV=production
++Environment=OPENRE_RELEASE=%i
++Environment=OPENRE_DB_PATH=/var/lib/openre/openre.db
++Environment=PATH=/usr/local/bin:/usr/bin:/bin
++ExecStart=/usr/bin/env node workers/restream-worker.js
++# A crash restarts the instance as a new generation; a clean exit (drained) stays stopped.
++Restart=on-failure
++RestartSec=2
++# SIGTERM starts a drain. KillMode=mixed: the ffmpeg children never get it from systemd directly.
++KillSignal=SIGTERM
++KillMode=mixed
++# An explicit stop (or host shutdown) waits this long for outputs to end, then kills.
++TimeoutStopSec=30min
++LimitNOFILE=65536
++NoNewPrivileges=true
++PrivateTmp=true
++ProtectSystem=strict
++ProtectHome=true
++ReadWritePaths=/var/lib/openre
++
++[Install]
++WantedBy=multi-user.target
ddiff --git a/deploy/systemd/[email protected] b/deploy/systemd/[email protected]
nnew file mode 100644
iindex 0000000..b21a0bc
---- /dev/null
++++ b/deploy/systemd/[email protected]
@@@ -0,0 +1,37 @@
++[Unit]
++Description=OpenRe.Stream RTMP ingest worker, release %i
++# One instance per release (instance name = the release directory). A worker deploy STARTS a new
++# instance; the coordinator then drains the older generation, which closes its public listener,
++# keeps its publishers until they leave (or OPENRE_DRAIN_MAX_MS), and exits 0 by itself.
++# Nothing here depends on openre-api or openre-session-coordinator: an API deploy never restarts
++# this unit. Never `systemctl restart` it during a broadcast; see README "Deploying".
++After=network.target
++
++[Service]
++Type=simple
++User=ubuntu
++Group=ubuntu
++WorkingDirectory=/opt/openre.stream/releases/%i
++EnvironmentFile=/etc/openvibe/openre.env
++Environment=NODE_ENV=production
++Environment=OPENRE_RELEASE=%i
++Environment=OPENRE_DB_PATH=/var/lib/openre/openre.db
++Environment=PATH=/usr/local/bin:/usr/bin:/bin
++ExecStart=/usr/bin/env node workers/rtmp-ingest.js
++# A crash restarts the instance as a new generation; a clean exit (drained) stays stopped.
++Restart=on-failure
++RestartSec=2
++# SIGTERM starts a drain; it does not drop publishers. Only the main process gets it.
++KillSignal=SIGTERM
++KillMode=mixed
++# An explicit stop (or host shutdown) waits this long for publishers to leave, then kills.
++TimeoutStopSec=30min
++LimitNOFILE=65536
++NoNewPrivileges=true
++PrivateTmp=true
++ProtectSystem=strict
++ProtectHome=true
++ReadWritePaths=/var/lib/openre
++
++[Install]
++WantedBy=multi-user.target
ddiff --git a/deploy/systemd/openre-session-coordinator.service b/deploy/systemd/openre-session-coordinator.service
nnew file mode 100644
iindex 0000000..2c44ae7
---- /dev/null
++++ b/deploy/systemd/openre-session-coordinator.service
@@@ -0,0 +1,28 @@
++[Unit]
++Description=OpenRe.Stream session coordinator: worker leases and generations, output assignment, recording requests, event relay
++# Stateless between ticks (everything is in the database): restarting it never ends a session.
++After=network.target
++
++[Service]
++Type=simple
++User=ubuntu
++Group=ubuntu
++WorkingDirectory=/opt/openre.stream/current
++EnvironmentFile=/etc/openvibe/openre.env
++Environment=NODE_ENV=production
++Environment=OPENRE_DB_PATH=/var/lib/openre/openre.db
++Environment=PATH=/usr/local/bin:/usr/bin:/bin
++ExecStart=/usr/bin/env node workers/coordinator.js
++Restart=always
++RestartSec=3
++KillSignal=SIGTERM
++TimeoutStopSec=15
++NoNewPrivileges=true
++PrivateTmp=true
++ProtectSystem=strict
++ProtectHome=true
++StateDirectory=openre
++ReadWritePaths=/var/lib/openre
++
++[Install]
++WantedBy=multi-user.target
ddiff --git a/docs/capabilities-proposal/openre.key.rotate.json b/docs/capabilities-proposal/openre.key.rotate.json
nnew file mode 100644
iindex 0000000..80119b0
---- /dev/null
++++ b/docs/capabilities-proposal/openre.key.rotate.json
@@@ -0,0 +1,21 @@
++{
++ "id": "openre.key.rotate",
++ "version": "1.0",
++ "owner": "openre",
++ "status": "active",
++ "visibility": "first-party",
++ "description": "Rotate a stream's ingest key: a new key is issued and returned once; the old one is revoked immediately or after a grace period, and running sessions can optionally be ended.",
++ "permissions": [
++ "openre:keys"
++ ],
++ "resourceConstraints": [
++ "owner"
++ ],
++ "quotaClass": "openre-write",
++ "events": [
++ "openre.key.rotated"
++ ],
++ "implementedBy": [
++ "POST /api/v1/streams/:id/keys/rotate"
++ ]
++}
ddiff --git a/docs/capabilities-proposal/openre.output.read.json b/docs/capabilities-proposal/openre.output.read.json
nnew file mode 100644
iindex 0000000..1a7ffd9
---- /dev/null
++++ b/docs/capabilities-proposal/openre.output.read.json
@@@ -0,0 +1,22 @@
++{
++ "id": "openre.output.read",
++ "version": "1.0",
++ "owner": "openre",
++ "status": "active",
++ "visibility": "first-party",
++ "description": "Read restream destinations (keys masked to their last four characters), outputs with health and progress, and bounded logs.",
++ "permissions": [
++ "openre:read"
++ ],
++ "resourceConstraints": [
++ "owner"
++ ],
++ "quotaClass": "openre-read",
++ "events": [],
++ "implementedBy": [
++ "GET /api/v1/streams/:id/destinations",
++ "GET /api/v1/sessions/:id/outputs",
++ "GET /api/v1/destinations/:id/logs",
++ "GET /api/v1/outputs/:id/logs"
++ ]
++}
ddiff --git a/docs/capabilities-proposal/openre.output.write.json b/docs/capabilities-proposal/openre.output.write.json
nnew file mode 100644
iindex 0000000..facbdfa
---- /dev/null
++++ b/docs/capabilities-proposal/openre.output.write.json
@@@ -0,0 +1,27 @@
++{
++ "id": "openre.output.write",
++ "version": "1.0",
++ "owner": "openre",
++ "status": "active",
++ "visibility": "first-party",
++ "description": "Create, update, delete and test restream destinations (stream keys are write-only and sealed at rest; URLs must be rtmp/rtmps/srt with public hosts), and start or stop an output for the live session.",
++ "permissions": [
++ "openre:outputs"
++ ],
++ "resourceConstraints": [
++ "owner"
++ ],
++ "quotaClass": "openre-write",
++ "events": [
++ "openre.output.healthy",
++ "openre.output.failed"
++ ],
++ "implementedBy": [
++ "POST /api/v1/streams/:id/destinations",
++ "PATCH /api/v1/destinations/:id",
++ "DELETE /api/v1/destinations/:id",
++ "POST /api/v1/destinations/:id/test",
++ "POST /api/v1/destinations/:id/start",
++ "POST /api/v1/destinations/:id/stop"
++ ]
++}
ddiff --git a/docs/capabilities-proposal/openre.session.end.json b/docs/capabilities-proposal/openre.session.end.json
nnew file mode 100644
iindex 0000000..cb4fe94
---- /dev/null
++++ b/docs/capabilities-proposal/openre.session.end.json
@@@ -0,0 +1,21 @@
++{
++ "id": "openre.session.end",
++ "version": "1.0",
++ "owner": "openre",
++ "status": "active",
++ "visibility": "first-party",
++ "description": "Ask the worker that owns a session to disconnect its encoder (the session then ends normally).",
++ "permissions": [
++ "openre:sessions"
++ ],
++ "resourceConstraints": [
++ "owner"
++ ],
++ "quotaClass": "openre-write",
++ "events": [
++ "openre.session.ended"
++ ],
++ "implementedBy": [
++ "POST /api/v1/sessions/:id/end"
++ ]
++}
ddiff --git a/docs/capabilities-proposal/openre.session.read.json b/docs/capabilities-proposal/openre.session.read.json
nnew file mode 100644
iindex 0000000..c9ae19e
---- /dev/null
++++ b/docs/capabilities-proposal/openre.session.read.json
@@@ -0,0 +1,22 @@
++{
++ "id": "openre.session.read",
++ "version": "1.0",
++ "owner": "openre",
++ "status": "active",
++ "visibility": "first-party",
++ "description": "Read ingest sessions: lifecycle and transitions, worker and generation, media info, recording state and the playback descriptor; list worker generations.",
++ "permissions": [
++ "openre:read"
++ ],
++ "resourceConstraints": [
++ "owner"
++ ],
++ "quotaClass": "openre-read",
++ "events": [],
++ "implementedBy": [
++ "GET /api/v1/sessions",
++ "GET /api/v1/sessions/:id",
++ "GET /api/v1/sessions/:id/playback",
++ "GET /api/v1/workers"
++ ]
++}
ddiff --git a/docs/capabilities-proposal/openre.stream.read.json b/docs/capabilities-proposal/openre.stream.read.json
nnew file mode 100644
iindex 0000000..7d9336b
---- /dev/null
++++ b/docs/capabilities-proposal/openre.stream.read.json
@@@ -0,0 +1,21 @@
++{
++ "id": "openre.stream.read",
++ "version": "1.0",
++ "owner": "openre",
++ "status": "active",
++ "visibility": "first-party",
++ "description": "Read stream definitions: ingest URLs, key metadata (hint, status; never a key), settings and external references. Owners read their own; a service reads every stream, or the streams of the X-OV-Subject it acts for.",
++ "permissions": [
++ "openre:read"
++ ],
++ "resourceConstraints": [
++ "owner"
++ ],
++ "quotaClass": "openre-read",
++ "events": [],
++ "implementedBy": [
++ "GET /api/v1/streams",
++ "GET /api/v1/streams/:id",
++ "GET /api/v1/streams/:id/keys"
++ ]
++}
ddiff --git a/docs/capabilities-proposal/openre.stream.write.json b/docs/capabilities-proposal/openre.stream.write.json
nnew file mode 100644
iindex 0000000..51d14e6
---- /dev/null
++++ b/docs/capabilities-proposal/openre.stream.write.json
@@@ -0,0 +1,21 @@
++{
++ "id": "openre.stream.write",
++ "version": "1.0",
++ "owner": "openre",
++ "status": "active",
++ "visibility": "first-party",
++ "description": "Create, update and archive stream definitions (creation issues the first ingest key, returned once). Archiving is refused while a session is open.",
++ "permissions": [
++ "openre:write"
++ ],
++ "resourceConstraints": [
++ "owner"
++ ],
++ "quotaClass": "openre-write",
++ "events": [],
++ "implementedBy": [
++ "POST /api/v1/streams",
++ "PATCH /api/v1/streams/:id",
++ "DELETE /api/v1/streams/:id"
++ ]
++}
ddiff --git a/docs/live-patch.diff b/docs/live-patch.diff
nnew file mode 100644
iindex 0000000..a511d6d
---- /dev/null
++++ b/docs/live-patch.diff
@@@ -0,0 +1,1192 @@
++diff --git a/.env.example b/.env.example
++index 5b3d8ac..b30f33b 100644
++--- a/.env.example
+++++ b/.env.example
++@@ -41,6 +41,17 @@ INTERNAL_API_KEY=change-me-to-a-random-secret
++ # OpenVibe.Events (durable events): stream lifecycle is published through Live's outbox with the
++ # OAuth client above (grant events.event.publish). Unset = off. EVENTS_PUBLISH=off also disables it.
++ #EVENTS_URL=http://127.0.0.1:4300
+++
+++# OpenRe.Stream (roadmap Wave 7): the ingest/restream authority. Unset = off, and every stream slot
+++# is ingested by Live exactly as before. When set, an admin can switch one slot at a time
+++# (PUT /api/admin/openre/managed/:id/ingest-authority); only RTMP moves. Live's service principal
+++# (the OAuth client above) needs openre.stream.read, openre.stream.write, openre.key.rotate and
+++# openre.session.read on audience openvibe.openre.
+++#OPENRE_URL=http://127.0.0.1:4500
+++#OPENRE_PUBLIC_URL=https://openre.stream
+++# Secret of Live's OpenVibe.Events subscription to openre.session.* (endpoint
+++# http://127.0.0.1:3000/internal/openre-events); sessions are mirrored into `streams` from it.
+++#OPENRE_EVENTS_SECRET=
++ # Path to the Network's RS256 public key (offline JWT verification)
++ #OV_NETWORK_PUBLIC_KEY=./data/keys/openvibe-network-public.pem
++
++diff --git a/public/js/broadcast-workspace.js b/public/js/broadcast-workspace.js
++index 5ecdf41..1ebe063 100644
++--- a/public/js/broadcast-workspace.js
+++++ b/public/js/broadcast-workspace.js
++@@ -261,6 +261,8 @@ async function _wsLoadProfile(managedStreamId) {
++ _wsState.selectedMs = { ..._wsState.selectedMs, ...data.managed_stream };
++ }
++ _wsState.streamKey = data.stream_key || null;
+++ // OpenRe-ingested slot: no key in the profile, only a hint (Regenerate shows a new key once).
+++ _wsState.streamKeyHint = data.stream_key_hint || null;
++ _wsState.profile = data.broadcast_settings || {};
++ _wsState.profile.vibeCoding = _wsNormalizeVibeSettings(_wsReadStoredVibeSettings(_wsState.profile));
++ _wsState.whipUrlBase = data.whip_url_base || null;
++@@ -270,6 +272,7 @@ async function _wsLoadProfile(managedStreamId) {
++ } catch {
++ _wsState.profile = {};
++ _wsState.streamKey = _wsState.selectedMs?.stream_key || null;
+++ _wsState.streamKeyHint = null;
++ _wsState.rtmpUrl = null;
++ }
++ _wsState.dirty = false;
++@@ -299,7 +302,7 @@ function _wsRenderPanel() {
++ const vibe = _wsGetVibeSettings();
++ const isLive = _wsIsManagedStreamLive(ms.id);
++ const method = ms.streaming_method || 'browser';
++- const streamKey = _wsState.streamKey || '';
+++ const streamKey = _wsState.streamKey || _wsState.streamKeyHint || '';
++ const vibePublisherUrl = _wsGetVibePublisherUrl();
++ const vibeFeedUrl = _wsGetVibeFeedUrl(ms);
++ const vibeSlotRef = ms.slug || ms.id;
++diff --git a/public/js/broadcast.js b/public/js/broadcast.js
++index f0db572..4610588 100644
++--- a/public/js/broadcast.js
+++++ b/public/js/broadcast.js
++@@ -1791,7 +1791,8 @@ async function showRTMPInstructions(stream) {
++ const data = await api(`/streams/${stream.id}/endpoint`);
++ const ep = data.endpoint || {};
++ const rtmpUrl = ep.rtmpUrl || `rtmp://${location.hostname}:1935/live`;
++- const streamKey = ep.streamKey || data.stream_key || 'N/A';
+++ // An OpenRe-ingested slot never returns its key here, only a hint (Regenerate shows a new key once).
+++ const streamKey = ep.streamKey || data.stream_key || ep.streamKeyHint || 'N/A';
++ document.getElementById('bc-rtmp-url').textContent = rtmpUrl;
++ document.getElementById('bc-rtmp-key').textContent = streamKey;
++ // Mirror into IRL Pro guide fields
++diff --git a/server/db/database.js b/server/db/database.js
++index e8f5903..60c20bf 100644
++--- a/server/db/database.js
+++++ b/server/db/database.js
++@@ -2704,6 +2704,10 @@ function initDb() {
++ database.exec('CREATE INDEX IF NOT EXISTS idx_vibe_events_stream ON vibe_coding_events(stream_id)');
++ } catch (e) { console.warn('[DB] vibe_coding migration:', e.message); }
++
+++ // OpenRe ingest switch (roadmap Wave 7): managed_streams.ingest_authority ('live' by default)
+++ // and the openre_sessions mirror. Additive only; see server/openre/schema.js.
+++ try { require('../openre/schema').ensure(database); } catch (e) { console.warn('[DB] openre schema:', e.message); }
+++
++ // Versioned migrations run last, so none of them can run before the tables they touch exist.
++ // A critical (money) migration that fails throws here and stops the boot rather than serving
++ // half-converted balances.
++diff --git a/server/index.js b/server/index.js
++index 4bba504..a26d028 100644
++--- a/server/index.js
+++++ b/server/index.js
++@@ -221,6 +221,8 @@ function webrtcStreamHasActiveProducer(streamId) {
++
++ function hasActiveLiveFeed(stream) {
++ if (!stream) return false;
+++ // OpenRe.Stream holds the ingest (a mirrored session it confirmed recently): not stale.
+++ if (require('./openre/mirror').hasLiveSession(stream.id)) return true;
++ const streamKey = getStreamKey(stream);
++ if (stream.protocol === 'rtmp') {
++ return !!streamKey && rtmpServer.isReceiving(streamKey);
++@@ -545,6 +547,9 @@ app.use('/data/arena', express.static(path.resolve(process.env.ARENA_IMAGE_PATH
++ // OpenVibe.Media → Live webhook (vod.ready / clip.ready …). Mounted BEFORE the
++ // /internal router because it authenticates with an HMAC signature, not X-Internal-Key.
++ app.post('/internal/media-webhook', require('./media-proxy/webhook'));
+++// OpenVibe.Events → Live: OpenRe session lifecycle mirrored into `streams` (signed delivery,
+++// OPENRE_EVENTS_SECRET; server/openre/mirror.js). Also before /internal (no X-Internal-Key).
+++app.post('/internal/openre-events', require('./openre/mirror').webhookHandler);
++
++ // Internal (server-to-server) routes — allow openvibe.network to call into this service
++ // Summary numbers for the Network's navigation service (ordering sites by real use).
++@@ -650,6 +655,7 @@ app.use('/api/recap', require('./recap/routes')); // after-show repor
++ app.use('/api/comments', commentRoutes);
++ app.use('/api/controls', controlRoutes);
++ app.use('/api/onvif', onvifRoutes);
+++app.use('/api/admin/openre', require('./openre/routes'));
++ app.use('/api/admin', adminRoutes);
++ app.use('/api/mod', require('./admin/mod-routes'));
++ app.use('/api/channels', require('./admin/channel-mod-routes'));
++@@ -1153,6 +1159,8 @@ async function start() {
++ // 5b. Resume enabled restreams for streams that survived the restart.
++ // WHIP/RTMP broadcasters have no browser session to re-start them manually.
++ for (const stream of db.getLiveStreams()) {
+++ // OpenRe restreams its own sessions; Live must not start a second push for them.
+++ if (require('./openre/mirror').ownsStream(stream.id)) continue;
++ restreamManager.resumeForStream(stream.id, stream.user_id, {
++ protocol: stream.protocol,
++ streamKey: stream.managed_stream_key,
++@@ -1220,6 +1228,9 @@ async function start() {
++ // 6e. Start periodic viewer count polling for restream destinations
++ restreamManager.startViewerCountPolling();
++
+++ // 6g. OpenRe mirror reconcile (only runs when OPENRE_URL is set; idle with nothing mirrored).
+++ try { require('./openre/mirror').start(); } catch (e) { console.warn('[OpenRe] mirror reconcile not started:', e.message); }
+++
++ // 6f. VOD storage/offload is owned by OpenVibe.Media now — nothing to start here.
++
++ // 7. Start HTTP server.
++diff --git a/server/openre/authority.js b/server/openre/authority.js
++new file mode 100644
++index 0000000..865fff1
++--- /dev/null
+++++ b/server/openre/authority.js
++@@ -0,0 +1,136 @@
+++'use strict';
+++/**
+++ * The per-slot ingest switch: managed_streams.ingest_authority = 'live' (default) | 'openre'.
+++ *
+++ * 'openre' means OpenRe.Stream ingests this slot's RTMP: Live's RTMP server refuses the slot's
+++ * key (so a stream can never be ingested twice), the Go Live UI shows OpenRe's ingest URL and
+++ * rotates OpenRe's key, and the slot's sessions reach `streams` through the mirror (mirror.js).
+++ * Only RTMP moves (OPENRE_PROTOCOLS); WHIP, browser/WebRTC and JSMPEG stay on Live until
+++ * OpenRe carries them.
+++ *
+++ * Switch off = zero behaviour change: every helper here returns what Live did before unless the
+++ * slot says 'openre' AND OpenRe is configured (openre-client enabled()).
+++ */
+++const crypto = require('crypto');
+++const db = require('../db/database');
+++const client = require('./openre-client');
+++
+++const OPENRE_PROTOCOLS = new Set(['rtmp']);
+++
+++function authorityOf(slot) {
+++ if (!slot || slot.ingest_authority !== 'openre') return 'live';
+++ return client.enabled() ? 'openre' : 'live';
+++}
+++
+++function slotById(id) {
+++ return id ? db.getManagedStreamById(id) : null;
+++}
+++
+++/** authorityOf(slotById(id)) === 'openre', without reading the slot when OpenRe is not configured. */
+++function slotIsOpenre(id) {
+++ if (!id || !client.enabled()) return false;
+++ return authorityOf(slotById(id)) === 'openre';
+++}
+++
+++/** Live's RTMP publish handler asks this before accepting a slot key. */
+++function refusesLiveIngest({ managedStream, protocol = 'rtmp' }) {
+++ try {
+++ if (!managedStream || !OPENRE_PROTOCOLS.has(protocol)) return false;
+++ // Unsetting OPENRE_URL is an emergency rollback to Live for every switched slot at once:
+++ // the slot then behaves as a Live slot again, with the Live key rotated at the switch
+++ // (only visible to the owner through Regenerate on the Go Live page).
+++ return authorityOf(managedStream) === 'openre';
+++ } catch {
+++ return false;
+++ }
+++}
+++
+++/** A slot as the owner's API returns it. Unchanged (same object) for Live-ingested slots. */
+++function serializeSlot(slot) {
+++ if (authorityOf(slot) !== 'openre') return slot;
+++ return { ...slot, stream_key: null, ingest_authority: 'openre', stream_key_managed_by: 'openre', openre_manage_url: client.manageUrl(slot.openre_stream_id) };
+++}
+++
+++/**
+++ * What the Go Live UI shows for an OpenRe slot: OpenRe's RTMP URL and the hint of its current key
+++ * (the key itself is only ever shown once, by a rotation).
+++ */
+++async function ingestFor(slot, subject) {
+++ const stream = slot.openre_stream_id ? await client.getStream(slot.openre_stream_id, { subject }) : await client.streamForSlot(slot.id);
+++ const rtmp = stream && stream.ingest && stream.ingest.rtmp;
+++ return {
+++ ingest_authority: 'openre',
+++ stream_key_managed_by: 'openre',
+++ openre_stream_id: stream ? stream.id : null,
+++ rtmp_url: rtmp ? rtmp.url : null,
+++ stream_key_hint: rtmp && rtmp.key_hint
+++ ? `Shown once: press Regenerate for a new key (current key ends in …${rtmp.key_hint})`
+++ : 'Press Regenerate to get your OpenRe stream key',
+++ openre_manage_url: client.manageUrl(stream ? stream.id : slot.openre_stream_id),
+++ };
+++}
+++
+++/** Regenerate on an OpenRe slot: OpenRe rotates, the new key is returned once. */
+++async function rotateFor(slot, subject) {
+++ const streamId = slot.openre_stream_id || (await client.streamForSlot(slot.id) || {}).id;
+++ if (!streamId) throw new client.OpenReError('this slot has no OpenRe stream yet', 409);
+++ const r = await client.rotateKey(streamId, { subject });
+++ return { stream_key: r.key.key, stream_key_managed_by: 'openre', rtmp_url: r.ingest && r.ingest.rtmp ? r.ingest.rtmp.url : null };
+++}
+++
+++function recordingModeFor(slot) {
+++ try {
+++ return db.resolveStreamRecordingMode({ user_id: slot.user_id, managed_stream_id: slot.id });
+++ } catch { return 'vod'; }
+++}
+++
+++/**
+++ * The admin switch (PUT /api/admin/openre/managed/:id/ingest-authority).
+++ * 'openre': needs OpenRe configured, the slot offline on Live, the owner's canonical subject;
+++ * finds or creates the OpenRe definition, then (one transaction) flips the slot and
+++ * rotates Live's own key for it, so no key that existed before works anywhere.
+++ * 'live': flips back (rollback). Live's key was rotated at the switch: the broadcaster
+++ * regenerates it on the Go Live page.
+++ */
+++async function setAuthority(slotId, authority, { force = false } = {}) {
+++ const slot = slotById(slotId);
+++ if (!slot) return { status: 404, error: 'Managed stream not found' };
+++ if (!['live', 'openre'].includes(authority)) return { status: 400, error: "authority must be 'live' or 'openre'" };
+++ if (authority === 'live') {
+++ db.run("UPDATE managed_streams SET ingest_authority = 'live', updated_at = CURRENT_TIMESTAMP WHERE id = ?", [slot.id]);
+++ return { status: 200, body: { managed_stream_id: slot.id, ingest_authority: 'live', next: 'The broadcaster regenerates the stream key on the Go Live page (the Live key was rotated when the slot moved to OpenRe).' } };
+++ }
+++ if (!client.enabled()) return { status: 409, error: 'OpenRe is not configured on this Live (OPENRE_URL, OV_OAUTH_CLIENT_SECRET)' };
+++ if (slot.ingest_authority === 'openre') return { status: 200, body: { managed_stream_id: slot.id, ingest_authority: 'openre', openre_stream_id: slot.openre_stream_id, unchanged: true } };
+++ const live = db.get('SELECT id FROM streams WHERE managed_stream_id = ? AND is_live = 1 LIMIT 1', [slot.id]);
+++ if (live) return { status: 409, error: `Slot ${slot.id} is live on Live (stream ${live.id}); switch it in a maintenance window, while it is offline` };
+++ const method = String(slot.streaming_method || '').toLowerCase();
+++ if (!force && slot.protocol !== 'rtmp' && !['rtmp', 'obs'].includes(method)) {
+++ return { status: 409, error: `Slot ${slot.id} is a ${slot.protocol}${method ? `/${method}` : ''} slot; OpenRe only ingests RTMP today (pass force to switch anyway)` };
+++ }
+++ const subject = require('../auth/identity-sync').subjectOf(slot.user_id);
+++ if (!subject) return { status: 409, error: 'The owner has no canonical subject yet (they need to sign in to Live once)' };
+++ let stream = await client.streamForSlot(slot.id);
+++ if (!stream) {
+++ let visibility = 'public';
+++ try { visibility = db.resolveStreamVodVisibility({ user_id: slot.user_id, managed_stream_id: slot.id }); } catch { /* public */ }
+++ stream = await client.createStreamForSlot(slot, { subject, recordingMode: recordingModeFor(slot), recordingVisibility: visibility });
+++ }
+++ const newLiveKey = crypto.randomBytes(20).toString('hex');
+++ db.getDb().transaction(() => {
+++ db.run("UPDATE managed_streams SET ingest_authority = 'openre', openre_stream_id = ?, stream_key = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?", [stream.id, newLiveKey, slot.id]);
+++ })();
+++ return {
+++ status: 200,
+++ body: {
+++ managed_stream_id: slot.id,
+++ ingest_authority: 'openre',
+++ openre_stream_id: stream.id,
+++ rtmp_url: stream.ingest && stream.ingest.rtmp ? stream.ingest.rtmp.url : null,
+++ live_key_rotated: true,
+++ next: 'The broadcaster presses Regenerate on the Go Live page (OpenRe issues the key) and pastes the new server and key into OBS.',
+++ },
+++ };
+++}
+++
+++module.exports = { OPENRE_PROTOCOLS, authorityOf, slotById, slotIsOpenre, refusesLiveIngest, serializeSlot, ingestFor, rotateFor, setAuthority };
++diff --git a/server/openre/mirror.js b/server/openre/mirror.js
++new file mode 100644
++index 0000000..365f61d
++--- /dev/null
+++++ b/server/openre/mirror.js
++@@ -0,0 +1,205 @@
+++'use strict';
+++/**
+++ * OpenRe sessions → Live's `streams` table (roadmap Wave 7, ADR-009: Live observes sessions through
+++ * OpenRe events and APIs). Watch pages, discovery, go-live notifications and analytics keep working
+++ * because an OpenRe-ingested session becomes an ordinary live `streams` row.
+++ *
+++ * POST /internal/openre-events the endpoint of Live's OpenVibe.Events subscription (topic
+++ * openre.session.*), signed with OPENRE_EVENTS_SECRET. Each event
+++ * is applied exactly once (openvibe-sdk inbox, consumer
+++ * 'live-openre-mirror'), in revision order per session.
+++ * reconcile (every 30 s) for mirrored live sessions only: asks OpenRe for the session,
+++ * keeps the row's heartbeat fresh, ends it if OpenRe says it ended
+++ * (covers a missed event). Does nothing when no row is mirrored.
+++ *
+++ * Consent: a session is mirrored only when its slot is switched to OpenRe on Live AND the OpenRe
+++ * definition says mirror_to_live (plan §15.10 "explicit visibility and consent").
+++ */
+++const db = require('../db/database');
+++const client = require('./openre-client');
+++const { authorityOf } = require('./authority');
+++
+++const CONSUMER = 'live-openre-mirror';
+++const RECONCILE_MS = 30000;
+++// A mirrored session OpenRe has not confirmed for this long is left to Live's stale cleanup.
+++const CONFIRM_WINDOW_MIN = 30;
+++let inbox = null;
+++let timer = null;
+++
+++function ensureInbox() {
+++ if (inbox) return inbox;
+++ const { createInbox } = require('openvibe-sdk/events');
+++ inbox = createInbox(db.getDb());
+++ inbox.ensureSchema();
+++ return inbox;
+++}
+++
+++function slotRef(payload) {
+++ const refs = Array.isArray(payload && payload.external_refs) ? payload.external_refs : [];
+++ const r = refs.find(x => x && x.service === 'live' && x.type === 'managed_stream');
+++ const id = r ? parseInt(r.id, 10) : NaN;
+++ return Number.isInteger(id) && id > 0 ? id : null;
+++}
+++
+++const toSqlTime = (iso) => { const d = iso ? new Date(iso) : new Date(); return Number.isNaN(d.getTime()) ? null : d.toISOString().replace('T', ' ').slice(0, 19); };
+++
+++/**
+++ * Apply one event (inside the inbox transaction). Returns { outcome, after? } where after() runs
+++ * once the transaction committed (go-live notifications must not fire for a rolled-back row).
+++ */
+++function apply(event) {
+++ const type = String(event && event.event_type || '');
+++ const p = event.payload || {};
+++ const sessionId = String(p.session_id || (event.subject && event.subject.id) || '');
+++ const revision = Number(event.subject && event.subject.revision) || 0;
+++ if (!/^openre\.session\.(started|ended|failed)$/.test(type) || !/^ses_[0-9A-HJKMNP-TV-Z]{26}$/.test(sessionId)) return { outcome: 'ignored' };
+++ const known = db.get('SELECT * FROM openre_sessions WHERE session_id = ?', [sessionId]);
+++ if (known && known.revision >= revision) return { outcome: 'stale' };
+++
+++ if (type === 'openre.session.started') {
+++ const slotId = slotRef(p);
+++ const slot = slotId ? db.getManagedStreamById(slotId) : null;
+++ if (!slot || authorityOf(slot) !== 'openre' || !p.mirror_to_live) return { outcome: 'not_mirrored' };
+++ const user = db.getUserById(slot.user_id);
+++ if (!user || user.is_banned) return { outcome: 'not_mirrored' };
+++ // A row made by the Go Live page waiting for RTMP is used, like Live's own RTMP ingest does.
+++ const waiting = db.get(`SELECT s.id FROM streams s LEFT JOIN openre_sessions o ON o.stream_id = s.id
+++ WHERE s.managed_stream_id = ? AND s.is_live = 1 AND s.protocol = 'rtmp' AND o.session_id IS NULL ORDER BY s.id DESC LIMIT 1`, [slot.id]);
+++ let streamId;
+++ let created = false;
+++ if (waiting) {
+++ streamId = waiting.id;
+++ db.run('UPDATE streams SET started_at = CURRENT_TIMESTAMP, last_heartbeat = CURRENT_TIMESTAMP WHERE id = ?', [streamId]);
+++ } else {
+++ const channel = db.ensureChannel(user.id);
+++ streamId = Number(db.createStream({
+++ user_id: user.id,
+++ channel_id: channel && channel.id,
+++ managed_stream_id: slot.id,
+++ control_config_id: slot.control_config_id || null,
+++ title: slot.title || `${user.display_name || user.username}'s Stream`,
+++ description: slot.description || '',
+++ category: slot.category || null,
+++ protocol: 'rtmp',
+++ is_nsfw: slot.is_nsfw ? 1 : 0,
+++ }).lastInsertRowid);
+++ db.run('UPDATE streams SET last_heartbeat = CURRENT_TIMESTAMP WHERE id = ?', [streamId]);
+++ created = true;
+++ }
+++ db.run(`INSERT INTO openre_sessions (session_id, managed_stream_id, stream_id, state, revision, started_at, confirmed_at, updated_at)
+++ VALUES (?, ?, ?, 'live', ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+++ ON CONFLICT(session_id) DO UPDATE SET stream_id = excluded.stream_id, state = 'live', revision = excluded.revision, confirmed_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP`,
+++ [sessionId, slot.id, streamId, revision, toSqlTime(p.started_at)]);
+++ const ended = db.endOtherLiveStreamsForSlot(slot.id, streamId);
+++ return {
+++ outcome: created ? 'created' : 'attached',
+++ after: () => {
+++ for (const sid of ended) { try { require('../streaming/broadcast-server').endStream(sid); } catch { /* */ } }
+++ try {
+++ const channel = db.getChannelByUserId(user.id);
+++ const configId = slot.control_config_id || (channel && channel.active_control_config_id);
+++ if (configId) db.applyConfigToStream(configId, streamId);
+++ } catch (e) { console.warn('[OpenRe] control config for mirrored stream failed:', e.message); }
+++ const stream = db.getStreamById(streamId) || { id: streamId };
+++ try { require('../streaming/golive-notify').notifyFollowersGoLive(user, stream); } catch (e) { console.warn('[OpenRe] go-live notify failed:', e.message); }
+++ try { require('../streaming/live-events').announceGoLive(stream, user); } catch { /* */ }
+++ console.log(`[OpenRe] session ${sessionId} mirrored into stream ${streamId} (${user.username}, slot ${slot.id})`);
+++ },
+++ };
+++ }
+++
+++ // ended / failed
+++ const state = type === 'openre.session.ended' ? 'ended' : 'failed';
+++ const streamId = known ? known.stream_id : null;
+++ db.run(`INSERT INTO openre_sessions (session_id, managed_stream_id, stream_id, state, revision, ended_at, updated_at)
+++ VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
+++ ON CONFLICT(session_id) DO UPDATE SET state = excluded.state, revision = excluded.revision, ended_at = excluded.ended_at, updated_at = CURRENT_TIMESTAMP`,
+++ [sessionId, known ? known.managed_stream_id : slotRef(p), streamId, state, revision, toSqlTime(p.ended_at)]);
+++ if (!streamId) return { outcome: 'recorded' };
+++ const row = db.getStreamById(streamId);
+++ if (row && row.is_live) db.endStream(streamId);
+++ return {
+++ outcome: 'ended',
+++ after: () => {
+++ try { db.computeAndCacheStreamAnalytics(streamId); } catch { /* */ }
+++ try { require('../streaming/broadcast-server').endStream(streamId); } catch { /* */ }
+++ console.log(`[OpenRe] session ${sessionId} ${state}: stream ${streamId} ended`);
+++ },
+++ };
+++}
+++
+++/** Express handler for POST /internal/openre-events (needs req.rawBody from express.json verify). */
+++function webhookHandler(req, res) {
+++ const secret = process.env.OPENRE_EVENTS_SECRET || '';
+++ if (!secret) return res.status(503).json({ error: 'OPENRE_EVENTS_SECRET is not set' });
+++ const { parseDelivery } = require('openvibe-sdk/events');
+++ const delivery = parseDelivery(req.rawBody, req.headers, secret);
+++ if (!delivery) return res.status(401).json({ error: 'bad signature' });
+++ let result;
+++ try {
+++ let after = null;
+++ result = ensureInbox().once(CONSUMER, delivery.event.event_id, () => {
+++ const r = apply(delivery.event);
+++ after = r.after || null;
+++ return r.outcome;
+++ });
+++ if (!result.duplicate && after) { try { after(); } catch (e) { console.warn('[OpenRe] post-commit step failed:', e.message); } }
+++ } catch (err) {
+++ console.error('[OpenRe] event apply failed:', err.message);
+++ return res.status(500).json({ error: 'apply failed' }); // Events retries
+++ }
+++ return res.status(204).end();
+++}
+++
+++/** Stream id → the mirrored OpenRe session (or null). */
+++function sessionForStream(streamId) {
+++ try { return db.get('SELECT * FROM openre_sessions WHERE stream_id = ? ORDER BY updated_at DESC LIMIT 1', [streamId]) || null; } catch { return null; }
+++}
+++
+++/** Does OpenRe hold a live ingest for this stream (confirmed recently)? Used by Live's stale cleanup. */
+++function hasLiveSession(streamId) {
+++ try {
+++ return Boolean(db.get(`SELECT 1 FROM openre_sessions WHERE stream_id = ? AND state = 'live'
+++ AND confirmed_at > datetime('now', '-${CONFIRM_WINDOW_MIN} minutes')`, [streamId]));
+++ } catch { return false; }
+++}
+++
+++/** Does OpenRe own this stream at all (so Live must not start its own restream/recording for it)? */
+++function ownsStream(streamId) {
+++ return Boolean(sessionForStream(streamId));
+++}
+++
+++async function reconcileOnce() {
+++ let rows = [];
+++ try { rows = db.all("SELECT * FROM openre_sessions WHERE state = 'live'"); } catch { return 0; }
+++ if (!rows.length || !client.enabled()) return 0;
+++ let n = 0;
+++ for (const r of rows) {
+++ let s;
+++ try { s = await client.getSession(r.session_id); } catch (err) {
+++ if (err.status === 404) s = { state: 'failed', revision: r.revision + 1, ended_at: null };
+++ else continue; // OpenRe unreachable: keep the last known state (bounded by CONFIRM_WINDOW_MIN)
+++ }
+++ if (s.state === 'live' || s.state === 'starting') {
+++ db.run("UPDATE openre_sessions SET confirmed_at = CURRENT_TIMESTAMP WHERE session_id = ?", [r.session_id]);
+++ if (r.stream_id) db.run('UPDATE streams SET last_heartbeat = CURRENT_TIMESTAMP WHERE id = ? AND is_live = 1', [r.stream_id]);
+++ } else if (['ending', 'ended', 'failed'].includes(s.state)) {
+++ const event = { event_type: `openre.session.${s.state === 'failed' ? 'failed' : 'ended'}`, subject: { type: 'ingest_session', id: r.session_id, revision: Math.max(Number(s.revision) || 0, r.revision + 1) }, payload: { session_id: r.session_id, ended_at: s.ended_at } };
+++ const out = db.getDb().transaction(() => apply(event))();
+++ if (out.after) out.after();
+++ n++;
+++ }
+++ }
+++ return n;
+++}
+++
+++function start() {
+++ if (timer || !client.enabled()) return;
+++ timer = setInterval(() => { reconcileOnce().catch(err => console.warn('[OpenRe] reconcile failed:', err.message)); }, RECONCILE_MS);
+++ if (timer.unref) timer.unref();
+++}
+++
+++function stop() { if (timer) clearInterval(timer); timer = null; }
+++function _reset() { stop(); inbox = null; }
+++
+++module.exports = { apply, webhookHandler, sessionForStream, hasLiveSession, ownsStream, reconcileOnce, start, stop, CONSUMER, _reset };
++diff --git a/server/openre/openre-client.js b/server/openre/openre-client.js
++new file mode 100644
++index 0000000..6862412
++--- /dev/null
+++++ b/server/openre/openre-client.js
++@@ -0,0 +1,127 @@
+++'use strict';
+++/**
+++ * OpenRe.Stream client (roadmap Wave 7). Live asks OpenRe for stream definitions, ingest URLs,
+++ * key rotation, sessions and playback descriptors; it never touches an OpenRe worker.
+++ *
+++ * Auth: Live's Network service principal (OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET),
+++ * client-credentials tokens for audience openvibe.openre. Grants Live needs on openre:
+++ * openre.stream.read, openre.stream.write, openre.key.rotate, openre.session.read
+++ * Calls made for a streamer carry X-OV-Subject: <their usr_ subject>.
+++ *
+++ * Env:
+++ * OPENRE_URL OpenRe API, host-local (http://127.0.0.1:4500). Unset = integration off:
+++ * every slot behaves exactly as before, whatever its ingest_authority.
+++ * OPENRE_PUBLIC_URL for links to the standalone UI (default https://openre.stream)
+++ * OPENRE_EVENTS_SECRET the signing secret of Live's OpenVibe.Events subscription to openre.*
+++ */
+++const { createServiceTokenClient } = require('openvibe-sdk/auth');
+++
+++const TIMEOUT_MS = 5000;
+++let tokens = null;
+++const cache = new Map(); // playback descriptors, 10 s
+++
+++function settings() {
+++ return {
+++ url: String(process.env.OPENRE_URL || '').replace(/\/+$/, ''),
+++ publicUrl: String(process.env.OPENRE_PUBLIC_URL || 'https://openre.stream').replace(/\/+$/, ''),
+++ networkInternalUrl: String(process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, ''),
+++ clientId: process.env.OV_OAUTH_CLIENT_ID || 'live',
+++ clientSecret: process.env.OV_OAUTH_CLIENT_SECRET || '',
+++ };
+++}
+++
+++/** True when Live is configured to talk to OpenRe at all. */
+++function enabled() {
+++ const s = settings();
+++ return Boolean(s.url && s.clientSecret);
+++}
+++
+++class OpenReError extends Error {
+++ constructor(message, status, body) { super(message); this.status = status || 0; this.body = body || null; this.code = body && body.code; }
+++}
+++
+++function tokenClient() {
+++ const s = settings();
+++ if (!tokens || tokens._for !== `${s.clientId}@${s.networkInternalUrl}`) {
+++ tokens = createServiceTokenClient({ tokenUrl: `${s.networkInternalUrl}/oauth/token`, clientId: s.clientId, clientSecret: s.clientSecret, audience: 'openvibe.openre' });
+++ tokens._for = `${s.clientId}@${s.networkInternalUrl}`;
+++ }
+++ return tokens;
+++}
+++
+++async function request(method, path, { body, subject, timeoutMs = TIMEOUT_MS } = {}) {
+++ if (!enabled()) throw new OpenReError('OpenRe integration is not configured (OPENRE_URL, OV_OAUTH_CLIENT_SECRET)', 0);
+++ const s = settings();
+++ const headers = { Accept: 'application/json', Authorization: `Bearer ${await tokenClient().getToken()}` };
+++ if (subject) headers['X-OV-Subject'] = subject;
+++ if (body !== undefined) headers['Content-Type'] = 'application/json';
+++ let res;
+++ let text = '';
+++ try {
+++ res = await fetch(`${s.url}${path}`, { method, headers, body: body === undefined ? undefined : JSON.stringify(body), signal: AbortSignal.timeout(timeoutMs) });
+++ text = await res.text().catch(() => '');
+++ } catch (err) {
+++ throw new OpenReError(`OpenRe unreachable (${method} ${path}): ${err.message}`, 0);
+++ }
+++ let json = null;
+++ if (text) { try { json = JSON.parse(text); } catch { json = null; } }
+++ if (!res.ok) throw new OpenReError((json && (json.detail || json.error)) || `OpenRe ${res.status} on ${method} ${path}`, res.status, json);
+++ return json;
+++}
+++
+++/** The OpenRe stream definition serving a Live slot, or null. */
+++async function streamForSlot(managedStreamId) {
+++ const r = await request('GET', `/api/v1/streams?external_ref=${encodeURIComponent(`live:managed_stream:${managedStreamId}`)}`);
+++ return (r && r.streams && r.streams[0]) || null;
+++}
+++
+++/** Create the definition for a slot, owned by the streamer's canonical subject. The key it
+++ * returns is dropped unseen: the streamer gets a usable key by rotating. */
+++async function createStreamForSlot(slot, { subject, recordingMode, recordingVisibility }) {
+++ const r = await request('POST', '/api/v1/streams', {
+++ subject,
+++ body: {
+++ title: slot.title || 'Stream',
+++ recording_mode: recordingMode,
+++ recording_visibility: recordingVisibility,
+++ mirror_to_live: true,
+++ external_refs: [
+++ { service: 'live', type: 'managed_stream', id: String(slot.id), label: slot.slug || slot.title || null },
+++ { service: 'live', type: 'user', id: String(slot.user_id), label: slot.username || null },
+++ ],
+++ },
+++ });
+++ return r.stream;
+++}
+++
+++async function rotateKey(streamId, { subject, graceSeconds = 0 } = {}) {
+++ return request('POST', `/api/v1/streams/${encodeURIComponent(streamId)}/keys/rotate`, { subject, body: { grace_seconds: graceSeconds } });
+++}
+++
+++async function getStream(streamId, { subject } = {}) {
+++ const r = await request('GET', `/api/v1/streams/${encodeURIComponent(streamId)}`, { subject });
+++ return r.stream;
+++}
+++
+++async function getSession(sessionId) {
+++ const r = await request('GET', `/api/v1/sessions/${encodeURIComponent(sessionId)}`);
+++ return r.session;
+++}
+++
+++/** Playback descriptor for a session (cached 10 s: the FLV proxy asks on every viewer connect). */
+++async function playback(sessionId) {
+++ const hit = cache.get(sessionId);
+++ if (hit && hit.at > Date.now() - 10000) return hit.value;
+++ const r = await request('GET', `/api/v1/sessions/${encodeURIComponent(sessionId)}/playback`);
+++ cache.set(sessionId, { at: Date.now(), value: r.playback });
+++ if (cache.size > 500) for (const [k, v] of cache) if (v.at < Date.now() - 60000) cache.delete(k);
+++ return r.playback;
+++}
+++
+++function manageUrl(streamId) {
+++ return `${settings().publicUrl}/streams/${encodeURIComponent(streamId || '')}`;
+++}
+++
+++function _reset() { tokens = null; cache.clear(); }
+++
+++module.exports = { enabled, settings, request, streamForSlot, createStreamForSlot, rotateKey, getStream, getSession, playback, manageUrl, OpenReError, _reset };
++diff --git a/server/openre/routes.js b/server/openre/routes.js
++new file mode 100644
++index 0000000..836fd8c
++--- /dev/null
+++++ b/server/openre/routes.js
++@@ -0,0 +1,49 @@
+++'use strict';
+++/**
+++ * Admin routes for the OpenRe ingest switch (mounted at /api/admin/openre, admin only). The lead
+++ * flips a slot during a maintenance window agreed with the broadcaster (OpenRe.Stream README,
+++ * "RTMP cutover runbook").
+++ *
+++ * GET /api/admin/openre/status configuration + switched slots
+++ * PUT /api/admin/openre/managed/:id/ingest-authority { authority: 'openre'|'live', force? }
+++ */
+++const express = require('express');
+++const db = require('../db/database');
+++const { requireAdmin } = require('../auth/auth');
+++const client = require('./openre-client');
+++const authority = require('./authority');
+++
+++const router = express.Router();
+++
+++router.get('/status', requireAdmin, (req, res) => {
+++ let slots = [];
+++ let live = [];
+++ try {
+++ slots = db.all("SELECT id, user_id, slug, title, protocol, streaming_method, openre_stream_id FROM managed_streams WHERE ingest_authority = 'openre' ORDER BY id");
+++ live = db.all("SELECT session_id, managed_stream_id, stream_id, state, confirmed_at FROM openre_sessions WHERE state = 'live'");
+++ } catch { /* schema not ready */ }
+++ res.json({
+++ enabled: client.enabled(),
+++ openre_url: client.settings().url || null,
+++ events_secret_set: Boolean(process.env.OPENRE_EVENTS_SECRET),
+++ protocols: [...authority.OPENRE_PROTOCOLS],
+++ slots,
+++ live_sessions: live,
+++ });
+++});
+++
+++router.put('/managed/:id/ingest-authority', requireAdmin, async (req, res) => {
+++ const id = parseInt(req.params.id, 10);
+++ if (!Number.isInteger(id) || id <= 0) return res.status(400).json({ error: 'Invalid managed stream id' });
+++ try {
+++ const r = await authority.setAuthority(id, String((req.body && req.body.authority) || ''), { force: Boolean(req.body && req.body.force) });
+++ if (r.error) return res.status(r.status).json({ error: r.error });
+++ console.log(`[OpenRe] admin ${req.user.username} set slot ${id} ingest authority to ${r.body.ingest_authority}`);
+++ return res.status(r.status).json(r.body);
+++ } catch (err) {
+++ console.error('[OpenRe] switch failed:', err.message);
+++ return res.status(err.status && err.status < 500 ? 409 : 502).json({ error: `OpenRe: ${err.message}` });
+++ }
+++});
+++
+++module.exports = router;
++diff --git a/server/openre/schema.js b/server/openre/schema.js
++new file mode 100644
++index 0000000..c0c0d85
++--- /dev/null
+++++ b/server/openre/schema.js
++@@ -0,0 +1,36 @@
+++'use strict';
+++/**
+++ * OpenRe ingest switch — Live-side schema (roadmap Wave 7, ADR-009). Additive and idempotent:
+++ *
+++ * managed_streams.ingest_authority 'live' (default) | 'openre' who ingests RTMP for this slot
+++ * managed_streams.openre_stream_id the OpenRe stream definition serving the slot (std_…)
+++ * openre_sessions Live's projection of OpenRe ingest sessions it mirrors into
+++ * `streams` (rebuilt from openre.session.* events; the
+++ * revision guard makes out-of-order delivery harmless)
+++ *
+++ * With every slot on 'live' (the default) nothing reads these columns differently from before.
+++ */
+++function ensure(db) {
+++ const cols = db.prepare('PRAGMA table_info(managed_streams)').all().map(c => c.name);
+++ if (cols.length && !cols.includes('ingest_authority')) {
+++ db.exec("ALTER TABLE managed_streams ADD COLUMN ingest_authority TEXT NOT NULL DEFAULT 'live'");
+++ }
+++ if (cols.length && !cols.includes('openre_stream_id')) {
+++ db.exec('ALTER TABLE managed_streams ADD COLUMN openre_stream_id TEXT DEFAULT NULL');
+++ }
+++ db.exec(`CREATE TABLE IF NOT EXISTS openre_sessions (
+++ session_id TEXT PRIMARY KEY,
+++ managed_stream_id INTEGER,
+++ stream_id INTEGER,
+++ state TEXT NOT NULL,
+++ revision INTEGER NOT NULL DEFAULT 0,
+++ started_at DATETIME,
+++ ended_at DATETIME,
+++ confirmed_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+++ updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
+++ )`);
+++ db.exec('CREATE INDEX IF NOT EXISTS idx_openre_sessions_stream ON openre_sessions(stream_id)');
+++ db.exec('CREATE INDEX IF NOT EXISTS idx_openre_sessions_state ON openre_sessions(state)');
+++}
+++
+++module.exports = { ensure };
++diff --git a/server/streaming/restream-routes.js b/server/streaming/restream-routes.js
++index 4b5f8e4..4ac081b 100644
++--- a/server/streaming/restream-routes.js
+++++ b/server/streaming/restream-routes.js
++@@ -14,6 +14,23 @@ const chatRelayService = require('../integrations/chat-relay-service');
++
++ const router = express.Router();
++
+++/**
+++ * A slot ingested by OpenRe.Stream (ingest_authority = 'openre') restreams from OpenRe: its
+++ * destinations are managed there (openre.stream), so Live refuses to edit or run them here rather
+++ * than letting two systems push the same stream. Slots on Live's own ingest are unaffected.
+++ */
+++function refuseIfOpenre(res, managedStreamId) {
+++ const openre = require('../openre/authority');
+++ if (!openre.slotIsOpenre(managedStreamId)) return false;
+++ const slot = openre.slotById(managedStreamId);
+++ res.status(409).json({
+++ error: 'Restreaming for this stream slot is managed on OpenRe.Stream',
+++ managed_by: 'openre',
+++ manage_url: require('../openre/openre-client').manageUrl(slot && slot.openre_stream_id),
+++ });
+++ return true;
+++}
+++
++ const VALID_PLATFORMS = ['youtube', 'twitch', 'kick', 'custom'];
++ const VALID_QUALITY_PRESETS = ['auto', 'low', 'medium', 'high', 'ultra', 'source'];
++ const VALID_ENCODER_PRESETS = ['ultrafast', 'superfast', 'veryfast', 'faster', 'fast', 'medium', 'slow'];
++@@ -198,6 +215,8 @@ router.post('/destinations', requireAuth, (req, res) => {
++ resolvedSlotId = ms.id;
++ }
++
+++ if (resolvedSlotId && refuseIfOpenre(res, resolvedSlotId)) return;
+++
++ // Enforce a reasonable limit per slot (or globally if no slot)
++ const existing = resolvedSlotId
++ ? (db.getRestreamDestinationsByManagedStream(resolvedSlotId) || [])
++@@ -256,6 +275,7 @@ router.put('/destinations/:id', requireAuth, (req, res) => {
++ if (!dest || dest.user_id !== req.user.id) {
++ return res.status(404).json({ error: 'Destination not found' });
++ }
+++ if (refuseIfOpenre(res, dest.managed_stream_id)) return;
++
++ const updates = {};
++ if (req.body.name !== undefined) updates.name = req.body.name?.trim() || dest.name;
++@@ -296,6 +316,7 @@ router.put('/destinations/:id', requireAuth, (req, res) => {
++ if (!ms || ms.user_id !== req.user.id) {
++ return res.status(403).json({ error: 'Not your stream slot' });
++ }
+++ if (refuseIfOpenre(res, ms.id)) return;
++ updates.managed_stream_id = ms.id;
++ } else {
++ updates.managed_stream_id = null;
++@@ -344,6 +365,7 @@ router.delete('/destinations/:id', requireAuth, (req, res) => {
++ if (!dest || dest.user_id !== req.user.id) {
++ return res.status(404).json({ error: 'Destination not found' });
++ }
+++ if (refuseIfOpenre(res, dest.managed_stream_id)) return;
++
++ // Stop any active restream for this destination
++ const liveStreams = getLiveStreamsForDestination(dest, req.user.id);
++@@ -367,6 +389,7 @@ router.post('/destinations/:id/start', requireAuth, async (req, res) => {
++ if (!dest || dest.user_id !== req.user.id) {
++ return res.status(404).json({ error: 'Destination not found' });
++ }
+++ if (refuseIfOpenre(res, dest.managed_stream_id)) return;
++ if (!dest.server_url || !dest.stream_key) {
++ return res.status(400).json({ error: 'Destination is not fully configured (missing server URL or stream key)' });
++ }
++@@ -423,6 +446,7 @@ router.post('/destinations/:id/stop', requireAuth, (req, res) => {
++ if (!dest || dest.user_id !== req.user.id) {
++ return res.status(404).json({ error: 'Destination not found' });
++ }
+++ if (refuseIfOpenre(res, dest.managed_stream_id)) return;
++
++ const liveStreams = getLiveStreamsForDestination(dest, req.user.id);
++ for (const stream of liveStreams) {
++diff --git a/server/streaming/routes.js b/server/streaming/routes.js
++index 70a0857..105be81 100644
++--- a/server/streaming/routes.js
+++++ b/server/streaming/routes.js
++@@ -28,6 +28,8 @@ const { requireAuth, requireStreamer, optionalAuth } = require('../auth/auth');
++ const jsmpegRelay = require('./jsmpeg-relay');
++ const webrtcSFU = require('./webrtc-sfu');
++ const recorder = require('./recorder');
+++const openreAuthority = require('../openre/authority');
+++const openreMirror = require('../openre/mirror');
++ const i18n = require('../i18n/translate');
++ const robotStreamerService = require('../integrations/robotstreamer-service');
++ const chatRelayService = require('../integrations/chat-relay-service');
++@@ -1467,7 +1469,7 @@ router.get('/managed/:managedStreamId/history', requireAuth, (req, res) => {
++
++ // ── Get full managed stream profile (structured fields + broadcast settings blob) ──
++ // Returned to the workspace panel for a single round-trip load.
++-router.get('/managed/:managedStreamId/profile', requireAuth, (req, res) => {
+++router.get('/managed/:managedStreamId/profile', requireAuth, async (req, res) => {
++ const managedStreamId = parseInt(req.params.managedStreamId);
++ if (!Number.isFinite(managedStreamId)) return res.status(400).json({ error: 'Invalid ID' });
++ try {
++@@ -1490,7 +1492,7 @@ router.get('/managed/:managedStreamId/profile', requireAuth, (req, res) => {
++ })();
++ const rtmpUrl = `rtmp://${rtmpHost}:${config.rtmp.port}/live`;
++
++- res.json({
+++ const body = {
++ managed_stream: msPublic,
++ stream_key: streamKey,
++ broadcast_settings: broadcastSettings,
++@@ -1499,7 +1501,16 @@ router.get('/managed/:managedStreamId/profile', requireAuth, (req, res) => {
++ whip_url_warning: whipUrlWarning,
++ rtmp_url: rtmpUrl,
++ restream_destinations: restreamDestinations,
++- });
+++ };
+++ // OpenRe ingests this slot: its RTMP server and key come from OpenRe (the key is only
+++ // ever shown by Regenerate). Slots on Live's own ingest get exactly the response above.
+++ if (openreAuthority.authorityOf(ms) === 'openre') {
+++ const subject = require('../auth/identity-sync').subjectOf(ms.user_id);
+++ Object.assign(body, { stream_key: null }, await openreAuthority.ingestFor(ms, subject).catch((err) => ({
+++ ingest_authority: 'openre', stream_key_managed_by: 'openre', rtmp_url: null, stream_key_hint: `OpenRe is unreachable (${err.message})`,
+++ })));
+++ }
+++ res.json(body);
++ } catch (err) {
++ console.error('[ManagedStreams] Profile error:', err.message);
++ res.status(500).json({ error: 'Could not load profile' });
++@@ -1568,7 +1579,7 @@ router.get('/managed', requireAuth, (req, res) => {
++ try {
++ const managed = db.getManagedStreamsByUserId(req.user.id);
++ const limit = db.getManagedStreamLimit(req.user);
++- res.json({ managed_streams: managed, limit });
+++ res.json({ managed_streams: managed.map(openreAuthority.serializeSlot), limit });
++ } catch (err) {
++ console.error('[ManagedStreams] List error:', err.message);
++ res.status(500).json({ error: 'Failed to list managed streams' });
++@@ -1815,7 +1826,7 @@ router.delete('/managed/:id', requireAuth, (req, res) => {
++ });
++
++ // Regenerate stream key for a managed stream
++-router.post('/managed/:id/regenerate-key', requireAuth, (req, res) => {
+++router.post('/managed/:id/regenerate-key', requireAuth, async (req, res) => {
++ try {
++ const msId = parseInt(req.params.id);
++ const ms = db.getManagedStreamById(msId);
++@@ -1824,6 +1835,17 @@ router.post('/managed/:id/regenerate-key', requireAuth, (req, res) => {
++ return res.status(403).json({ error: 'Not your managed stream' });
++ }
++
+++ // OpenRe ingests this slot: OpenRe rotates its key (the old one stops working at once) and
+++ // the new one is shown here, once.
+++ if (openreAuthority.authorityOf(ms) === 'openre') {
+++ try {
+++ const subject = require('../auth/identity-sync').subjectOf(ms.user_id);
+++ return res.json(await openreAuthority.rotateFor(ms, subject));
+++ } catch (err) {
+++ return res.status(502).json({ error: `Could not rotate the key on OpenRe: ${err.message}` });
+++ }
+++ }
+++
++ const crypto = require('crypto');
++ const newKey = crypto.randomBytes(20).toString('hex');
++ db.run('UPDATE managed_streams SET stream_key = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', [newKey, msId]);
++@@ -2125,7 +2147,7 @@ router.delete('/:id', requireAuth, (req, res) => {
++ });
++
++ // ── Get Streaming Endpoint Info ──────────────────────────────
++-router.get('/:id/endpoint', requireAuth, (req, res) => {
+++router.get('/:id/endpoint', requireAuth, async (req, res) => {
++ try {
++ const stream = db.getStreamById(req.params.id);
++ if (!stream) return res.status(404).json({ error: 'Stream not found' });
++@@ -2175,6 +2197,18 @@ router.get('/:id/endpoint', requireAuth, (req, res) => {
++ whipUrlSource,
++ ...(whipUrlWarning ? { whipUrlWarning } : {}),
++ };
+++ } else if (stream.protocol === 'rtmp' && openreAuthority.slotIsOpenre(stream.managed_stream_id)) {
+++ // OpenRe ingests this slot: its server URL, the hint of its key (never the key).
+++ const slot = openreAuthority.slotById(stream.managed_stream_id);
+++ const ingest = await openreAuthority.ingestFor(slot, require('../auth/identity-sync').subjectOf(slot.user_id)).catch(() => ({}));
+++ endpoint = {
+++ rtmpUrl: ingest.rtmp_url || null,
+++ streamKey: null,
+++ streamKeyHint: ingest.stream_key_hint || 'Press Regenerate to get your OpenRe stream key',
+++ keyManagedBy: 'openre',
+++ flvUrl: `/api/streams/rtmp-proxy/${stream.id}.flv`,
+++ };
+++ return res.json({ endpoint, stream_key: null });
++ } else if (stream.protocol === 'rtmp') {
++ const rtmpHost = config.rtmp.host || (() => {
++ try { return new URL(config.baseUrl).hostname; } catch { return hostname; }
++@@ -2239,6 +2273,10 @@ router.get('/:id/rtmp-status', requireAuth, (req, res) => {
++ if (stream.protocol !== 'rtmp') {
++ return res.status(400).json({ error: 'Not an RTMP stream' });
++ }
+++ const mirrored = openreMirror.sessionForStream(stream.id);
+++ if (mirrored) {
+++ return res.json({ receiving: mirrored.state === 'live', connected_at: mirrored.started_at, managed_by: 'openre' });
+++ }
++ const rtmpKey = stream.managed_stream_key || db.getUserById(stream.user_id)?.stream_key;
++ const rtmpServer = require('./rtmp-server');
++ const status = rtmpServer.getStatus(rtmpKey);
++@@ -2402,17 +2440,27 @@ router.get('/:id/call', optionalAuth, (req, res) => {
++ // ── RTMP FLV Proxy ───────────────────────────────────────────
++ // Proxies HTTP-FLV from the internal NMS server so the browser fetches
++ // from the same HTTPS origin, avoiding CSP / mixed-content issues.
++-router.get('/rtmp-proxy/:streamId.flv', (req, res) => {
+++router.get('/rtmp-proxy/:streamId.flv', async (req, res) => {
++ try {
++ const stream = db.getStreamById(req.params.streamId);
++ if (!stream || !stream.is_live || stream.protocol !== 'rtmp') {
++ return res.status(404).end();
++ }
++- const flvKey = stream.managed_stream_key || db.getUserById(stream.user_id)?.stream_key;
++- if (!flvKey) return res.status(404).end();
++-
++- const nmsPort = config.rtmp.port + 8000;
++- const url = `http://127.0.0.1:${nmsPort}/live/${flvKey}.flv`;
+++ let url;
+++ const mirrored = openreMirror.sessionForStream(stream.id);
+++ if (mirrored) {
+++ // An OpenRe session: pull HTTP-FLV from the OpenRe worker that holds it, as named by
+++ // OpenRe's playback descriptor (a loopback URL on this host; no key in it).
+++ if (mirrored.state !== 'live') return res.status(404).end();
+++ const pb = await require('../openre/openre-client').playback(mirrored.session_id).catch(() => null);
+++ if (!pb || !pb.flv || !/^http:\/\/127\.0\.0\.1:\d+\/live\/ses_[0-9A-Z]+\.flv$/.test(pb.flv.internal_url)) return res.status(502).end();
+++ url = pb.flv.internal_url;
+++ } else {
+++ const flvKey = stream.managed_stream_key || db.getUserById(stream.user_id)?.stream_key;
+++ if (!flvKey) return res.status(404).end();
+++ const nmsPort = config.rtmp.port + 8000;
+++ url = `http://127.0.0.1:${nmsPort}/live/${flvKey}.flv`;
+++ }
++
++ const http = require('http');
++ const upstream = http.get(url, (nmsRes) => {
++diff --git a/server/streaming/rtmp-server.js b/server/streaming/rtmp-server.js
++index fb8ba5c..709da89 100644
++--- a/server/streaming/rtmp-server.js
+++++ b/server/streaming/rtmp-server.js
++@@ -121,6 +121,16 @@ class RTMPServer extends EventEmitter {
++ return;
++ }
++
+++ // OpenRe.Stream ingests this slot (managed_streams.ingest_authority = 'openre'): refuse
+++ // here so one stream can never be ingested twice. Slots on 'live' (the default) are
+++ // untouched. See server/openre/authority.js.
+++ if (require('../openre/authority').refusesLiveIngest({ managedStream, protocol: 'rtmp' })) {
+++ console.log(`[RTMP] Rejected: slot ${managedStream.id} is ingested by OpenRe`);
+++ const session = this.nms.getSession(id);
+++ if (session) session.reject();
+++ return;
+++ }
+++
++ // Create or update stream record
++ // Look for an existing RTMP stream (created via Go Live page) that's waiting for the RTMP client
++ const existingStreams = db.getLiveStreamsByUserId(resolvedUser.id);
++diff --git a/test/openre-switch.test.js b/test/openre-switch.test.js
++new file mode 100644
++index 0000000..d9630e8
++--- /dev/null
+++++ b/test/openre-switch.test.js
++@@ -0,0 +1,216 @@
+++'use strict';
+++
+++// OpenRe ingest switch (roadmap Wave 7, ADR-009). With OPENRE_URL unset, or a slot on 'live' (the
+++// default), nothing changes. With a slot switched to 'openre': Live's RTMP refuses its key, the
+++// Go Live helpers show OpenRe's URL and rotate OpenRe's key, and OpenRe session events (signed
+++// Events deliveries, applied once, in revision order) become ordinary `streams` rows.
+++
+++const assert = require('assert');
+++const fs = require('fs');
+++const os = require('os');
+++const path = require('path');
+++const http = require('http');
+++
+++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-openre-switch-'));
+++process.env.DB_PATH = path.join(tmp, 'live.db');
+++delete process.env.OPENRE_URL;
+++delete process.env.OPENRE_EVENTS_SECRET;
+++
+++const SUBJECT = 'usr_01J0000000000000000000000Q';
+++const SES = (n) => `ses_01J00000000000000000000${String(n).padStart(3, '0')}`.slice(0, 30);
+++const openreCalls = [];
+++let openreSessions = {};
+++let openreStreams = [];
+++
+++const stub = http.createServer((req, res) => {
+++ let body = '';
+++ req.on('data', (c) => { body += c; });
+++ req.on('end', () => {
+++ res.setHeader('Content-Type', 'application/json');
+++ const send = (status, obj) => { res.statusCode = status; res.end(JSON.stringify(obj)); };
+++ if (req.url === '/oauth/token') {
+++ const p = new URLSearchParams(body);
+++ // Live also asks for other audiences here (the go-live fan-out to Network).
+++ return send(200, { access_token: p.get('audience') === 'openvibe.openre' ? 'svc-live' : 'other', token_type: 'Bearer', expires_in: 300 });
+++ }
+++ if (!req.url.startsWith('/api/v1/')) return send(404, {});
+++ assert.strictEqual(req.headers.authorization, 'Bearer svc-live');
+++ openreCalls.push({ method: req.method, url: req.url, subject: req.headers['x-ov-subject'] || null, body: body ? JSON.parse(body) : null });
+++ const stream = (id) => ({ id, ingest: { rtmp: { url: 'rtmp://ingest.openre.stream:1936/live', key_hint: 'WXYZ' } } });
+++ if (req.method === 'GET' && req.url.startsWith('/api/v1/streams?external_ref=')) return send(200, { streams: openreStreams });
+++ if (req.method === 'POST' && req.url === '/api/v1/streams') { openreStreams = [stream('std_01J0000000000000000000000S')]; return send(201, { stream: openreStreams[0], key: { key: 'ork_dropped' } }); }
+++ if (req.method === 'GET' && req.url.startsWith('/api/v1/streams/std_')) return send(200, { stream: stream(req.url.split('/').pop()) });
+++ if (req.method === 'POST' && req.url.endsWith('/keys/rotate')) return send(200, { key: { key: 'ork_newkey_shown_once' }, ingest: { rtmp: { url: 'rtmp://ingest.openre.stream:1936/live' } } });
+++ const m = /^\/api\/v1\/sessions\/(ses_[0-9A-Z]+)(\/playback)?$/.exec(req.url);
+++ if (m && openreSessions[m[1]]) {
+++ if (m[2]) return send(200, { playback: { flv: { internal_url: `http://127.0.0.1:19999/live/${m[1]}.flv` } } });
+++ return send(200, { session: openreSessions[m[1]] });
+++ }
+++ return send(404, { code: 'openre.not_found' });
+++ });
+++});
+++
+++function signed(event, secret) {
+++ const { signDelivery } = require('openvibe-sdk/events');
+++ const raw = Buffer.from(JSON.stringify({ event, seq: 1 }));
+++ return { raw, sig: signDelivery(raw, secret) };
+++}
+++
+++async function deliver(base, event, { secret = 'whsec_test', sig } = {}) {
+++ const s = signed(event, secret);
+++ const res = await fetch(`${base}/internal/openre-events`, { method: 'POST', headers: { 'Content-Type': 'application/json', 'X-OpenVibe-Signature': sig || s.sig }, body: s.raw });
+++ return res.status;
+++}
+++
+++let seq = 0;
+++function sessionEvent(type, sessionId, revision, payload = {}) {
+++ seq++;
+++ return {
+++ event_id: `evt_01J00000000000000000${String(seq).padStart(5, '0')}`,
+++ event_type: `openre.session.${type}`,
+++ version: 1,
+++ source: 'openre',
+++ actor: { type: 'user', id: SUBJECT },
+++ timestamp: new Date().toISOString(),
+++ subject: { type: 'ingest_session', id: sessionId, revision },
+++ payload: { session_id: sessionId, external_refs: [{ service: 'live', type: 'managed_stream', id: '701' }], mirror_to_live: true, started_at: new Date().toISOString(), ...payload },
+++ };
+++}
+++
+++(async () => {
+++ const db = require('../server/db/database');
+++ db.initDb();
+++ const d = db.getDb();
+++ d.prepare("INSERT INTO users (id, username, display_name, password_hash, stream_key) VALUES (601, 'caster', 'Caster', 'x', 'personalkey601')").run();
+++ d.prepare("INSERT INTO users (id, username, display_name, password_hash) VALUES (602, 'nosub', 'NoSub', 'x')").run();
+++ d.prepare(`INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (601, 'network', '91', '${SUBJECT}')`).run();
+++ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key, streaming_method) VALUES (701, 601, 'OBS slot', 'rtmp', 'livekey701aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', 'obs')").run();
+++ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (702, 601, 'Browser slot', 'webrtc', 'livekey702aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')").run();
+++ d.prepare("INSERT INTO managed_streams (id, user_id, title, protocol, stream_key) VALUES (703, 602, 'No subject', 'rtmp', 'livekey703aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')").run();
+++
+++ const authority = require('../server/openre/authority');
+++ const mirror = require('../server/openre/mirror');
+++
+++ // ── Switch off ────────────────────────────────────────────
+++ const cols = d.prepare('PRAGMA table_info(managed_streams)').all().map(c => c.name);
+++ assert.ok(cols.includes('ingest_authority') && cols.includes('openre_stream_id'));
+++ assert.strictEqual(db.getManagedStreamById(701).ingest_authority, 'live', "every slot starts on 'live'");
+++ const slot701 = db.getManagedStreamById(701);
+++ assert.strictEqual(authority.serializeSlot(slot701), slot701, 'the same object: responses unchanged');
+++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: slot701 }), false);
+++ // Even a slot marked 'openre' behaves as before while OpenRe is not configured.
+++ d.prepare("UPDATE managed_streams SET ingest_authority = 'openre' WHERE id = 701").run();
+++ const marked = db.getManagedStreamById(701);
+++ assert.strictEqual(authority.authorityOf(marked), 'live');
+++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: marked }), false);
+++ assert.strictEqual(authority.serializeSlot(marked), marked);
+++ assert.strictEqual(authority.slotIsOpenre(701), false);
+++ assert.strictEqual(mirror.hasLiveSession(1), false);
+++ assert.strictEqual(mirror.ownsStream(1), false);
+++ assert.strictEqual(await mirror.reconcileOnce(), 0);
+++ d.prepare("UPDATE managed_streams SET ingest_authority = 'live' WHERE id = 701").run();
+++
+++ const express = require('express');
+++ const app = express();
+++ app.use(express.json({ limit: '1mb', verify: (req, res, buf) => { req.rawBody = buf; } }));
+++ app.post('/internal/openre-events', mirror.webhookHandler);
+++ const server = await new Promise(r => { const s = app.listen(0, '127.0.0.1', () => r(s)); });
+++ const base = `http://127.0.0.1:${server.address().port}`;
+++ assert.strictEqual(await deliver(base, sessionEvent('started', SES(1), 2)), 503, 'inert without OPENRE_EVENTS_SECRET');
+++
+++ // ── Configured, slot still on 'live' ─────────────────────
+++ await new Promise(r => stub.listen(0, '127.0.0.1', r));
+++ const stubUrl = `http://127.0.0.1:${stub.address().port}`;
+++ process.env.OPENRE_URL = stubUrl;
+++ process.env.OV_NETWORK_INTERNAL_URL = stubUrl;
+++ process.env.OV_OAUTH_CLIENT_SECRET = 'live-secret';
+++ process.env.OPENRE_EVENTS_SECRET = 'whsec_test';
+++ require('../server/openre/openre-client')._reset();
+++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: db.getManagedStreamById(701) }), false);
+++ assert.strictEqual(await deliver(base, sessionEvent('started', SES(2), 2)), 204);
+++ assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM streams').get().n, 0, 'a slot on live is never mirrored');
+++
+++ // ── The switch ───────────────────────────────────────────
+++ assert.strictEqual((await authority.setAuthority(702, 'openre')).status, 409, 'a browser slot needs force');
+++ assert.strictEqual((await authority.setAuthority(703, 'openre')).status, 409, 'no canonical subject');
+++ const liveRow = db.createStream({ user_id: 601, managed_stream_id: 701, title: 'on Live', protocol: 'rtmp' });
+++ assert.strictEqual((await authority.setAuthority(701, 'openre')).status, 409, 'refused while live on Live');
+++ db.endStream(liveRow.lastInsertRowid);
+++ const flip = await authority.setAuthority(701, 'openre');
+++ assert.strictEqual(flip.status, 200, JSON.stringify(flip));
+++ assert.strictEqual(flip.body.openre_stream_id, 'std_01J0000000000000000000000S');
+++ const created = openreCalls.find(c => c.method === 'POST' && c.url === '/api/v1/streams');
+++ assert.strictEqual(created.subject, SUBJECT, 'created for the owner subject');
+++ assert.deepStrictEqual(created.body.external_refs.map(r => `${r.service}:${r.type}:${r.id}`), ['live:managed_stream:701', 'live:user:601']);
+++ assert.strictEqual(created.body.mirror_to_live, true);
+++ const switched = db.getManagedStreamById(701);
+++ assert.strictEqual(switched.ingest_authority, 'openre');
+++ assert.strictEqual(db.getManagedStreamByStreamKey('livekey701aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'), undefined, "Live's old key is rotated away");
+++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched }), true, "Live's RTMP refuses the slot");
+++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched, protocol: 'whip' }), false, 'only RTMP moved');
+++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: null }), false, 'personal keys stay on Live');
+++ const shown = authority.serializeSlot(switched);
+++ assert.strictEqual(shown.stream_key, null);
+++ assert.strictEqual(shown.stream_key_managed_by, 'openre');
+++ const ingest = await authority.ingestFor(switched, SUBJECT);
+++ assert.strictEqual(ingest.rtmp_url, 'rtmp://ingest.openre.stream:1936/live');
+++ assert.match(ingest.stream_key_hint, /…WXYZ/);
+++ const rotated = await authority.rotateFor(switched, SUBJECT);
+++ assert.strictEqual(rotated.stream_key, 'ork_newkey_shown_once');
+++ assert.strictEqual(openreCalls.filter(c => c.url.endsWith('/keys/rotate')).pop().subject, SUBJECT);
+++
+++ // ── Mirror ────────────────────────────────────────────────
+++ assert.strictEqual(await deliver(base, sessionEvent('started', SES(3), 2), { sig: 'sha256=00' }), 401, 'bad signature');
+++ assert.strictEqual(await deliver(base, sessionEvent('started', SES(3), 2, { mirror_to_live: false })), 204);
+++ assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM streams WHERE is_live = 1').get().n, 0, 'no consent, no mirror');
+++
+++ const started = sessionEvent('started', SES(4), 2);
+++ assert.strictEqual(await deliver(base, started), 204);
+++ assert.strictEqual(await deliver(base, started), 204, 'redelivery');
+++ const rows = d.prepare('SELECT * FROM streams WHERE is_live = 1').all();
+++ assert.strictEqual(rows.length, 1, 'exactly one live row');
+++ assert.strictEqual(rows[0].protocol, 'rtmp');
+++ assert.strictEqual(rows[0].managed_stream_id, 701);
+++ assert.strictEqual(rows[0].user_id, 601);
+++ const streamId = rows[0].id;
+++ assert.strictEqual(mirror.sessionForStream(streamId).session_id, SES(4));
+++ assert.strictEqual(mirror.hasLiveSession(streamId), true, 'stale cleanup leaves it alone');
+++ assert.strictEqual(mirror.ownsStream(streamId), true, 'Live does not resume its own restreams for it');
+++
+++ // Reconcile: OpenRe says live → heartbeat refreshed; says ended → row ended.
+++ openreSessions[SES(4)] = { id: SES(4), state: 'live', revision: 2 };
+++ d.prepare("UPDATE streams SET last_heartbeat = datetime('now', '-10 minutes') WHERE id = ?").run(streamId);
+++ await mirror.reconcileOnce();
+++ assert.ok(d.prepare("SELECT last_heartbeat > datetime('now', '-1 minute') AS fresh FROM streams WHERE id = ?").get(streamId).fresh);
+++ openreSessions[SES(4)] = { id: SES(4), state: 'ended', revision: 4, ended_at: new Date().toISOString() };
+++ await mirror.reconcileOnce();
+++ assert.strictEqual(db.getStreamById(streamId).is_live, 0);
+++ // The ended event arriving afterwards changes nothing (revision guard).
+++ assert.strictEqual(await deliver(base, sessionEvent('ended', SES(4), 4)), 204);
+++
+++ // Out of order: ended (rev 4) before started (rev 2) never creates a live row.
+++ assert.strictEqual(await deliver(base, sessionEvent('ended', SES(5), 4)), 204);
+++ assert.strictEqual(await deliver(base, sessionEvent('started', SES(5), 2)), 204);
+++ assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM streams WHERE is_live = 1').get().n, 0);
+++
+++ // A normal started → failed pair.
+++ assert.strictEqual(await deliver(base, sessionEvent('started', SES(6), 2)), 204);
+++ const s6 = d.prepare('SELECT id FROM streams WHERE is_live = 1').get().id;
+++ assert.strictEqual(await deliver(base, sessionEvent('failed', SES(6), 3, { failure_reason: 'worker_lost' })), 204);
+++ assert.strictEqual(db.getStreamById(s6).is_live, 0);
+++
+++ // ── Rollback ─────────────────────────────────────────────
+++ assert.strictEqual((await authority.setAuthority(701, 'live')).status, 200);
+++ const back = db.getManagedStreamById(701);
+++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: back }), false);
+++ assert.strictEqual(authority.serializeSlot(back), back);
+++ // Unsetting OPENRE_URL is the emergency rollback for every switched slot at once.
+++ await authority.setAuthority(701, 'openre');
+++ delete process.env.OPENRE_URL;
+++ assert.strictEqual(authority.refusesLiveIngest({ managedStream: db.getManagedStreamById(701) }), false);
+++
+++ mirror._reset();
+++ server.close();
+++ stub.close();
+++ console.log('✅ openre switch: off = unchanged, switch/rotation, RTMP refusal, signed mirror (once, ordered), reconcile, rollback');
+++})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/docs/service-manifest-proposal.json b/docs/service-manifest-proposal.json
nnew file mode 100644
iindex 0000000..070d0cc
---- /dev/null
++++ b/docs/service-manifest-proposal.json
@@@ -0,0 +1,40 @@
++{
++ "version": "0.1.0",
++ "contractRanges": {
++ "openvibe-contracts": ">=0.13.0 <1.0.0"
++ },
++ "id": "openre",
++ "name": "OpenRe.Stream",
++ "status": "alpha",
++ "repository": "OpenVibers/OpenRe.Stream",
++ "domains": [
++ "openre.stream",
++ "ingest.openre.stream"
++ ],
++ "publicOrigin": "https://openre.stream",
++ "health": "/api/health",
++ "ready": "/api/ready",
++ "capabilities": [
++ "openre.stream.read",
++ "openre.stream.write",
++ "openre.key.rotate",
++ "openre.session.read",
++ "openre.session.end",
++ "openre.output.read",
++ "openre.output.write"
++ ],
++ "eventsProduced": [
++ "openre.session.started",
++ "openre.session.ended",
++ "openre.session.failed",
++ "openre.output.healthy",
++ "openre.output.failed",
++ "openre.recording.requested",
++ "openre.key.rotated"
++ ],
++ "eventsConsumed": [],
++ "namespacesOwned": [
++ "openre.*"
++ ],
++ "notes": "Ingest and restream authority (ADR-009). Alpha: RTMP ingest and RTMP/SRT restream run in transport workers with generations and leases, separate from the API; WHIP, WebRTC/SFU and JSMPEG are still ingested by Live. Runs in parallel with Live; slots move one at a time behind Live's per-slot ingest_authority switch."
++}
ddiff --git a/package-lock.json b/package-lock.json
iindex 9ca58c4..0a0a916 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -14,9 +14,9 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"node-media-server": "2.7.4",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.8.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.13.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.2.2",
-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.0.0"
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.2.1"
},
"engines": {
"node": ">=22.12.0"
@@@ -1115,9 +1115,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.8.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.8.0",
-- "integrity": "sha512-2Msdb+k+fJZf9kQTPS2jdY8+o9FhL9tBuu3FzArHs3JcBcc06ReHpOi2sgnSBgQcSvZSaSsWT2YVbmhjnEplqQ==",
++ "version": "0.13.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.13.0",
++ "integrity": "sha512-MeIGJYHQsbr6ISeEErduUw/PD52X5IBclH2ht/G/t4lSX5X4o2AwcZYLEzOZrcbaXQyplgC7r86RCGFvi1ifUA==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
@@@ -1140,9 +1140,9 @@
}
},
"node_modules/openvibe-shared": {
-- "version": "1.0.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.0.0",
-- "integrity": "sha512-JOSH6LbIVzE9oiVWZmUnjZ+Ab8dRQ/SPk9aqBjbJGXp6kBoJA+tj7x1mp6vCusaxGHp17b2r9gmJQXk1TRL/JA==",
++ "version": "1.2.1",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.2.1",
++ "integrity": "sha512-t8EHgju5/H2VqkM89LzdCKxBogJ8xilxmvZVJWQIYaGPKe7vMe4yrITlJi/Sy98phsffeFRogpaukDuO6QlDUw==",
"license": "MIT",
"dependencies": {
"jsonwebtoken": "^9.0.0"
ddiff --git a/package.json b/package.json
iindex 4ac933b..c37b6ce 100644
---- a/package.json
++++ b/package.json
@@@ -23,9 +23,9 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"node-media-server": "2.7.4",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.8.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.13.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.2.2",
-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.0.0"
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.2.1"
},
"engines": {
"node": ">=22.12.0"
ddiff --git a/scripts/migrate-from-live.js b/scripts/migrate-from-live.js
oold mode 100644
nnew mode 100755
ddiff --git a/server/config.js b/server/config.js
iindex 9ae75c6..6e6a20b 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -57,7 +57,8 @@ function load(env = process.env) {
// ── Secrets ─────────────────────────────────────────────
// 32-byte key (hex or base64) that encrypts destination stream keys and SRT passphrases
// at rest (AES-256-GCM). OPENRE_SECRETS_KEY_PREVIOUS is tried for decryption only, so the
-- // key can be rotated (scripts/reencrypt is not needed: rows are re-encrypted on write).
++ // key can be rotated: old rows keep opening while it is set, and a destination is sealed
++ // with the current key whenever its stream key is saved again.
secretsKey: env.OPENRE_SECRETS_KEY || '',
secretsKeyPrevious: env.OPENRE_SECRETS_KEY_PREVIOUS || '',
@@@ -66,6 +67,9 @@ function load(env = process.env) {
// Public ingest port. Live's in-process ingest owns 1935 until the RTMP cutover
// (README "Port plan"); OpenRe listens on 1936 until then.
port: rtmpPort,
++ // Further public ports served by every worker, e.g. 1935 once Live's own RTMP ingest is
++ // retired (README "Port plan"): URLs handed out with :1936 keep working forever.
++ extraPorts: list(env.OPENRE_RTMP_EXTRA_PORTS, []).map(Number).filter(n => Number.isInteger(n) && n > 0 && n !== rtmpPort),
bindHost: env.OPENRE_RTMP_BIND || '0.0.0.0',
// What streamers paste into OBS: rtmp://<publicHost>[:port]/live
publicHost: env.OPENRE_RTMP_PUBLIC_HOST || (isProduction ? 'ingest.openre.stream' : '127.0.0.1'),
@@@ -111,7 +115,6 @@ function load(env = process.env) {
liveAckTimeoutMs: int(env.OPENRE_LIVE_ACK_TIMEOUT_MS, 20000),
stableMs: int(env.OPENRE_STABLE_MS, 30000),
startDelayMs: int(env.OPENRE_OUTPUT_START_DELAY_MS, 3000),
-- cooldownMinutes: int(env.OPENRE_DEST_COOLDOWN_MINUTES, 30),
},
// ── Events (OpenVibe.Events through the SDK outbox) ─────
ddiff --git a/test/proposals.test.js b/test/proposals.test.js
nnew file mode 100644
iindex 0000000..6dccf21
---- /dev/null
++++ b/test/proposals.test.js
@@@ -0,0 +1,36 @@
++'use strict';
++// The capability and service manifest proposals are valid contracts documents, and they match what
++// the code enforces and emits (so the contracts release that adopts them is exact).
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const contracts = require('openvibe-contracts');
++const { TYPES } = require('../server/events');
++const { suite, ROOT } = require('./helpers');
++
++const t = suite('proposals');
++const dir = path.join(ROOT, 'docs', 'capabilities-proposal');
++const manifest = JSON.parse(fs.readFileSync(path.join(ROOT, 'docs', 'service-manifest-proposal.json'), 'utf8'));
++
++t('every capability proposal validates against capabilities.capability@1 and has three segments', () => {
++ for (const f of fs.readdirSync(dir)) {
++ const cap = JSON.parse(fs.readFileSync(path.join(dir, f), 'utf8'));
++ const v = contracts.validate('capabilities.capability@1', cap);
++ assert.ok(v.valid, `${f}: ${JSON.stringify(v.errors)}`);
++ assert.strictEqual(`${cap.id}.json`, f);
++ assert.strictEqual(cap.id.split('.').length, 3);
++ assert.strictEqual(cap.owner, 'openre');
++ }
++});
++
++t('the service manifest proposal validates and lists exactly the guarded capabilities and emitted events', () => {
++ const v = contracts.validate('registry.service-manifest@1', manifest);
++ assert.ok(v.valid, JSON.stringify(v.errors));
++ const code = fs.readFileSync(path.join(ROOT, 'server', 'api', 'v1.js'), 'utf8');
++ const guarded = [...new Set([...code.matchAll(/guard\('([a-z.]+)'\)/g)].map(m => m[1]))].sort();
++ assert.deepStrictEqual(guarded, [...manifest.capabilities].sort());
++ assert.deepStrictEqual(fs.readdirSync(dir).map(f => f.replace(/\.json$/, '')).sort(), guarded);
++ assert.deepStrictEqual([...manifest.eventsProduced].sort(), Object.values(TYPES).sort());
++});
++
++t.run();
ddiff --git a/workers/rtmp-ingest.js b/workers/rtmp-ingest.js
iindex 483af3e..773d9ae 100644
---- a/workers/rtmp-ingest.js
++++ b/workers/rtmp-ingest.js
@@@ -62,7 +62,8 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
};
/** node-media-server session id → { sessionId, definitionId, endReason } */
const publishers = new Map();
-- let publicServer = null;
++ let publicServers = [];
++ const publicPorts = new Set([config.rtmp.port, ...config.rtmp.extraPorts]);
let playServer = null;
let flvServer = null;
let endpoints = null;
@@@ -97,7 +98,7 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
}
function isPublicSocket(session) {
-- return Boolean(session && session.socket && session.socket.localPort === config.rtmp.port);
++ return Boolean(session && session.socket && publicPorts.has(session.socket.localPort));
}
on('prePublish', (id, streamPath) => {
@@@ -178,10 +179,8 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
}
function closePublic() {
-- if (!publicServer) return;
-- const srv = publicServer;
-- publicServer = null;
-- srv.close();
++ if (!publicServers.length) return;
++ for (const srv of publicServers.splice(0)) srv.close();
log.log('[rtmp] public listener closed; running publishers stay until they leave');
}
@@@ -221,17 +220,20 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
return undefined;
});
const flvPort = await listenInRange(flvServer, config.rtmp.internalPortMin, config.rtmp.internalPortMax, new Set([rtmpPlayPort]));
-- endpoints = { publicPort: config.rtmp.port, rtmpPlayPort, flvPort };
++ endpoints = { publicPort: config.rtmp.port, publicPorts: [...publicPorts], rtmpPlayPort, flvPort };
runtime.register(endpoints);
-- publicServer = newRtmpServer();
-- await new Promise((resolve, reject) => {
-- publicServer.once('error', reject);
-- publicServer.listen({ port: config.rtmp.port, host: config.rtmp.bindHost, reusePort: true }, () => resolve());
-- });
-- publicServer.on('error', (err) => log.error(`[rtmp] public listener: ${err.message}`));
++ for (const port of publicPorts) {
++ const srv = newRtmpServer();
++ await new Promise((resolve, reject) => {
++ srv.once('error', reject);
++ srv.listen({ port, host: config.rtmp.bindHost, reusePort: true }, () => resolve());
++ });
++ srv.on('error', (err) => log.error(`[rtmp] public listener ${port}: ${err.message}`));
++ publicServers.push(srv);
++ }
runtime.ready();
-- log.log(`[rtmp] generation ${runtime.me.generation} ready: publish ${config.rtmp.bindHost}:${config.rtmp.port}, play 127.0.0.1:${rtmpPlayPort}, flv 127.0.0.1:${flvPort}`);
++ log.log(`[rtmp] generation ${runtime.me.generation} ready: publish ${config.rtmp.bindHost}:${[...publicPorts].join(',')}, play 127.0.0.1:${rtmpPlayPort}, flv 127.0.0.1:${flvPort}`);
return endpoints;
}