openvibers/openvibe.live Commits

Restreams end with their source; RobotStreamer state, errors and progress reach the dashboard; SRT destinations

OP OpenVibers committed 17 Sept 2026 · 1ba082d8ff · 21 files, +981 −1014

RobotStreamer passthrough (fixes robots stuck "live" showing black after the streamer stopped):
- the relay now ends on producer-removed, on the stream row going not-live (5s backstop) and
  on explicit stop; restarts re-check the source, back off 3s→30s and give up after 12 —
  before, it restarted flat-out forever and kept its peer to RS open after the source was gone
- a stop that lands mid-setup tears down the half-built peer instead of leaving it on RS
- status(streamId): state/restarts/last error/next retry/uptime/codec/audio + RS's RTCP view
- POST /api/robotstreamer/restream/{start,stop} (owner, live stream): the dashboard's RS
  Start/Stop reach the server (they only touched the browser's unused publisher before)
- RTMP/JSMPEG sources skip the relay (no SFU producer to forward) instead of looping
- the dead ffmpeg "native publisher" and its worker are removed

Every stream-end path stops the forwarders: broadcast-server 60s grace timer (RS, ffmpeg
restreams, chat relays, AI bots), RTMP unpublish (RS + relays + bots), admin force-end (RS,
restreams, relays, bots, signaling room, SFU room, voice channel — it only flipped is_live).

ffmpeg restreams:
- live is acknowledged from `-progress pipe:1` (first block with frames), not guessed after
  15s; no progress in 20s = a real error; progress (fps/bitrate/speed/dropped) in the status
- ffmpeg's last stderr lines are mapped to a sentence the streamer can act on (rejected key,
  unreachable/unresolvable ingest, TLS, connection closed, SRT passphrase, bad option)
- SRT destinations: srt:// URLs go out as MPEG-TS in caller mode with streamid, per-destination
  latency and optional passphrase (stored server-side only, redacted in logs)
- 1x VBV (bufsize = bitrate) on every preset and custom bitrate: ~1s less latency
- status carries platform, transport, liveAt/uptime, next retry, max attempts
- duplicate autoStartForStream and an unused constant removed

Dashboard: each destination row has a detail line (uptime, fps, bitrate, encoder-behind
warning, dropped frames, last error, retry countdown, cooldown), a Retry button on failed
ones, the RobotStreamer row driven by the server's state; the slot form gains an x264 encoder
speed field and SRT latency/passphrase fields that appear for srt:// URLs.

Tests: test/rs-passthrough-lifecycle.test.js, test/restream-manager-lifecycle.test.js (incl.
the progress ACK against a real ffmpeg). Docs: docs/restreaming.md; SECURITY_AUDIT row.

Changes

ddiff --git a/SECURITY_AUDIT.md b/SECURITY_AUDIT.md
iindex 81504818..30e5c8d8 100644
---- a/SECURITY_AUDIT.md
++++ b/SECURITY_AUDIT.md
@@@ -60,6 +60,7 @@ test/authorization were also run against the previous code and failed there, cas
| Low-Medium | AI viewer "clone" copied any user's chat history from every channel (and their AI insight) into a bot. | Streamers clone only from what that person said in their own channel; staff unchanged. | authorization |
| Low-Medium | No limit on chat WebSocket connections per address. | 48 per address (admins exempt); clients back off to 30 s on a 4029 close. | test/chat-connection-cap.test.js |
| Medium | Voice: 1:1 call channels were listed publicly and anyone (anonymous included) could walk into them; `call-user` had no rate limit or block check; anyone could answer any invite; a kicked user could rejoin at once; one host could fill the 8-seat Public channel; TURN was emitted without credentials on a box with no secret; static TURN credentials went to every visitor. | Private calls hidden from strangers and invite-only; call-user 6/min + block check; respond requires the invite; 60 s kick cooldown; 3 sockets per address, duplicate accounts replaced; TURN only with credentials; production coturn now runs `use-auth-secret` and every ICE list (voice and SFU, `server/net/turn.js`) carries a credential that expires after an hour — the old static pair no longer authenticates. | test/turn-credentials.test.js + relay probe on the host |
++| Medium | Restreams outlived their source: the RobotStreamer passthrough kept its peer open (and restarted forever, without checking the stream was still live) after the broadcaster's socket timed out, OBS stopped, or an admin force-ended the stream — the robot stayed "live" on robotstreamer.com showing black. Force-end only flipped `is_live`. The dashboard's RS Stop button never reached the server. | Every stream-end path (grace timer, RTMP unpublish, WHIP end, stale sweep, admin force-end, own DELETE) stops the RS relay, ffmpeg restreams, chat relays and AI bots; the relay also ends itself on `producer-removed` / the stream row going not-live, backs off and gives up after 12 restarts; `POST /api/robotstreamer/restream/{start,stop}` (owner-only, live stream only); SRT passphrases are stored server-side and never returned (`has_srt_passphrase`), and redacted from ffmpeg command lines in logs. | test/rs-passthrough-lifecycle.test.js, test/restream-manager-lifecycle.test.js |
| Low | Slot `control_config_id` accepted other users' profiles; restream viewer counts writable for any destination. | Ownership checks. | manual |
| Low | ONVIF discovery (LAN scan with default-password probing) available to every user. | Admin only, bounded timeout. | manual |
| Low | RobotStreamer API calls carrying the user's RS token disabled TLS verification. | Verification on for api.robotstreamer.com (valid certificate). | manual |
ddiff --git a/docs/README.md b/docs/README.md
iindex cade57c0..ea1810e7 100644
---- a/docs/README.md
++++ b/docs/README.md
@@@ -14,7 +14,8 @@
## Streamer Tools
- [Dashboard](dashboard.md) — Streamer dashboard features: emotes, rewards, chat logs, media management
--- [Restream Branding](restream-branding.md) — Multi-platform restream configuration
++- [Restreaming](restreaming.md) — ffmpeg restreams (RTMP/RTMPS/SRT) and the RobotStreamer passthrough: latency, live acknowledgement, error relay, lifecycle
++- [Restream Branding](restream-branding.md) — Channel branding for the restream targets
## Architecture
- [System Overview](architecture.md) — Server components, protocols, data flow
ddiff --git a/docs/restreaming.md b/docs/restreaming.md
nnew file mode 100644
iindex 00000000..25b95a8d
---- /dev/null
++++ b/docs/restreaming.md
@@@ -0,0 +1,113 @@
++# Restreaming
++
++How a live OpenVibe stream is forwarded to other platforms, what each path does to the picture,
++what the broadcaster sees, and how the forwarders are kept in step with the stream's lifecycle.
++
++## Two forwarders
++
++| Path | Module | Destinations | What happens to the video |
++|---|---|---|---|
++| **ffmpeg restreams** | `server/streaming/restream-manager.js` | Twitch, YouTube, Kick, any RTMP/RTMPS or SRT ingest (`restream_destinations` rows) | RTMP source: **codec copy** (nothing re-encoded). Browser/WHIP/JSMPEG source: re-encoded to H.264/AAC on the server with the destination's quality preset |
++| **RobotStreamer passthrough** | `server/integrations/rs-passthrough-relay.js` | One robot per stream slot (`robotstreamer_integrations`) | **Zero re-encode**: the source's RTP is pulled from our mediasoup SFU and written straight into a werift peer connection joined to RobotStreamer's SFU |
++
++The RobotStreamer path needs an SFU producer, so it serves browser and OBS/WHIP sources. RTMP and
++JSMPEG streams get the chat mirror only (`robotstreamer-service.js` skips the relay for them
++instead of letting it wait for a producer that never comes).
++
++## Latency
++
++- Output muxing is real-time on every path: `-muxdelay 0 -muxpreload 0 -flush_packets 1`.
++- Encoded paths use `-tune zerolatency`, CBR, 2-second closed GOPs and a **1× VBV** (`bufsize =
++ bitrate`), which is about a second less end-to-end than the previous 2× buffer at the same
++ bitrate. Custom bitrates keep the 1× rule.
++- RTMP sources are forwarded with `-c copy`, so their latency is the platform's own.
++- The RobotStreamer relay adds no encoder at all; its latency is the network path to RS plus RS's
++ own WebRTC playout. Keyframe requests from RS are relayed to the source encoder (rate-limited
++ to one per 2 s so a lossy link cannot turn into a keyframe storm).
++- SRT destinations take a `latency` (receiver buffer) per destination; 120 ms is libsrt's default
++ and right for a same-continent ingest, 300–600 ms for intercontinental or Wi-Fi hops.
++
++## SRT destinations
++
++A destination whose server URL starts with `srt://host:port` is sent as MPEG-TS in caller mode
++(`pkt_size=1316`). The "stream key" field becomes the SRT `streamid`; the destination's
++`srt_latency_ms` and optional `srt_passphrase` (10–79 characters, AES per the SRT spec) travel as
++URL options. The passphrase is stored server-side and never returned to the client
++(`has_srt_passphrase` only). Command lines in the log are redacted (`server/utils/redact.js` masks
++`passphrase=`, `streamid=` and the key segment).
++
++## Live acknowledgement and errors (ffmpeg)
++
++ffmpeg is spawned with `-progress pipe:1 -stats_period 1`. The first progress block with
++`frame > 0` (or `out_time > 0` for a codec copy) is the **live** signal — it means the ingest
++accepted the stream and is taking data. Before this the dashboard guessed "Live" after 15 s
++regardless. If no progress arrives within 20 s the run is killed with a "no response from the
++ingest" error so the retry machinery and the dashboard see a real failure instead of a
++"Starting…" that never ends.
++
++The last stderr lines of a failed run are kept raw (`session.lastErrorRaw`, logged) and mapped to
++a sentence a streamer can act on (`RestreamManager.friendlyFfmpegError`): rejected key, unreachable
++or unresolvable ingest, TLS failure, connection closed by the platform, SRT passphrase mismatch,
++bad encoder option. Restarts back off exponentially (5 s → 2 min, 30 attempts); a destination
++that never went live and crashes four times in a row is put in a persisted cooldown
++(circuit breaker) so it is not hammered on every go-live.
++
++`GET /api/restream/status` returns, per live stream and destination: `status`, `platform`,
++`transport` (`rtmp`/`srt`), `liveAt`, `uptimeMs`, `restartAttempts`/`maxRestartAttempts`,
++`nextRestartAt`, `lastError` and the latest `progress` (`fps`, `bitrate_kbps`, `speed`,
++`dropped`, `frame`). It also carries `robotstreamer[streamId]` — the passthrough state below.
++
++## RobotStreamer passthrough lifecycle
++
++The relay must never outlive its source: RobotStreamer keeps a robot "live" for exactly as long
++as our producers exist on its SFU, so a relay left running after the streamer stopped shows RS
++viewers a live robot with a black picture. The session therefore ends on any of:
++
++1. `producer-removed` for the stream's video producer on our SFU (tab closed, room closed);
++2. the stream row going not-live (checked every 5 s as a backstop for end paths that never
++ touch the SFU room — stale-heartbeat sweeps, an ingest handler calling `db.endStream`);
++3. an explicit stop: the broadcaster's **Stop** button (`POST /api/robotstreamer/restream/stop`),
++ the integration being disabled, or any of the stream-end paths below.
++
++Restarts (RS websocket closed, ICE/DTLS to RS lost, the run throwing, a producer being replaced
++on reconnect) first re-check that the source is live, then back off 3 s → 30 s and give up after
++12 consecutive attempts. A run that stayed live for a minute resets the counter. A session that
++gave up stays visible as `failed` with its reason until **Retry**
++(`POST /api/robotstreamer/restream/start`) relaunches it.
++
++`rsPassthroughRelay.status(streamId)` (in `/api/restream/status`) reports `state`
++(`starting | live | restarting | failed`), `restarts`, `last_error` / `last_restart_reason`,
++`next_restart_at`, `uptime_ms`, `video_codec`, `has_audio`, `keyframe_requests`, `ingest_stalls`
++and RS's own RTCP view of us (`rs.loss_pct`, `rs.rtt_ms`, `rs.nacks`, `rs.plis`).
++
++## Every stream-end path stops the forwarders
++
++| End path | Where | Stops |
++|---|---|---|
++| Streamer ends the stream | `DELETE /api/streams/:id` | RS, chat relays, AI bots, restreams (via room close), voice channel |
++| Broadcaster socket gone for 60 s | `broadcast-server.js` grace timer → `_stopForwarders()` | RS, restreams, chat relays, AI bots |
++| OBS stops (RTMP `donePublish`) | `index.js` `rtmpServer.on('unpublish')` | restreams, RS, chat relays, AI bots |
++| WHIP session ends | `whip-handler.js` `endActiveWhipStream` | restreams, RS, chat relays, AI bots, room |
++| No heartbeat for 5 min | `index.js` stale sweep | RS, chat relays, AI bots, restreams, room |
++| Admin force-end | `DELETE /api/admin/streams/:id` | RS, restreams, chat relays, AI bots, room, SFU room, voice channel |
++| Process shutdown | `index.js` SIGTERM handler | `rsPassthroughRelay.stopAll()` closes each RS peer so RS's viewers get `consumerClosed` immediately |
++
++## Dashboard
++
++The live-controls restream panel (`public/js/broadcast.js` `updateRestreamControlPanel`) polls
++`/api/restream/status` every 5 s while live and shows, per destination, a badge (Idle / Starting /
++Live / Error / Failed / Paused) plus a detail line: uptime, fps and bitrate, an encoder-behind
++warning when `speed < 0.95×`, dropped frames, the last error, and a countdown to the next retry.
++The RobotStreamer row is driven by the server's passthrough state when the server relay owns the
++robot (its Start/Stop reach the server; before, the button only touched the browser's unused
++publisher). The stream-slot settings form (`broadcast-workspace.js`) exposes per destination:
++quality preset, video/audio bitrate, fps, x264 encoder speed, and for `srt://` URLs the latency
++and passphrase.
++
++## Tests
++
++- `test/rs-passthrough-lifecycle.test.js` — stop-when-source-ended, backoff, cap, relaunch, status.
++- `test/restream-manager-lifecycle.test.js` — URL building (RTMP/RTMPS/Kick/SRT), per-transport
++ output flags, 1× VBV, friendly errors, and the progress-based live ACK against a real ffmpeg.
++- `test/rs-passthrough-ssrc-filter.test.js`, `test/rs-chat-bridge-single-flight.test.js` — earlier
++ regressions.
ddiff --git a/public/css/features/broadcast.css b/public/css/features/broadcast.css
iindex 22eb0d03..4cde8430 100644
---- a/public/css/features/broadcast.css
++++ b/public/css/features/broadcast.css
@@@ -777,9 +777,32 @@
.bc-restream-row {
display: flex;
align-items: center;
-- gap: 10px;
++ flex-wrap: wrap;
++ gap: 6px 10px;
font-size: 0.82rem;
}
++/* Second line under a destination: uptime, fps/bitrate, the last error, retry countdown. */
++.bc-restream-detail {
++ flex-basis: 100%;
++ padding-left: 30px;
++ font-size: 0.74rem;
++ color: var(--text-secondary);
++ line-height: 1.35;
++ overflow-wrap: anywhere;
++}
++.bc-restream-err { color: var(--danger, #e74c3c); }
++.bc-restream-warn { color: var(--warning, #f39c12); }
++.bc-restream-transport {
++ margin-left: 6px;
++ padding: 1px 5px;
++ border-radius: 6px;
++ font-size: 0.66rem;
++ font-weight: 700;
++ letter-spacing: 0.04em;
++ background: var(--bg-tertiary);
++ color: var(--text-secondary);
++ vertical-align: 1px;
++}
.bc-restream-platform-icon {
width: 20px;
@@@ -793,7 +816,7 @@
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
-- max-width: 120px;
++ max-width: 200px;
}
.bc-restream-status-badge {
ddiff --git a/public/js/broadcast-workspace.js b/public/js/broadcast-workspace.js
iindex af3a5e6d..5ecdf41b 100644
---- a/public/js/broadcast-workspace.js
++++ b/public/js/broadcast-workspace.js
@@@ -2668,8 +2668,8 @@ function _wsRenderRestreamList(container) {
</div>
</div>
<div class="bc-ws-restream-card-info">
-- <span class="muted" style="font-size:0.78rem">${esc(d.server_url || '')} · Key: ****${esc(d.stream_key ? d.stream_key.slice(-4) : '')}</span>
-- <span class="muted" style="font-size:0.78rem">Quality: ${esc(d.quality_preset || 'auto')}</span>
++ <span class="muted" style="font-size:0.78rem">${esc(d.server_url || '')} · ${d.transport === 'srt' ? 'Stream ID' : 'Key'}: ****${esc(d.stream_key ? d.stream_key.slice(-4) : '')}</span>
++ <span class="muted" style="font-size:0.78rem">${d.transport === 'srt' ? `SRT · ${d.srt_latency_ms || 120} ms latency${d.has_srt_passphrase ? ' · encrypted' : ''} · ` : ''}Quality: ${esc(d.quality_preset || 'auto')}${d.custom_encoder_preset ? ` · ${esc(d.custom_encoder_preset)}` : ''}</span>
</div>
${coolBanner}
</div>`;
@@@ -2833,6 +2833,19 @@ async function _wsDeleteRestreamDest(destId) {
}
}
++/** Show the SRT options only for srt:// URLs and label the key field accordingly. */
++function _wsSyncRestreamTransport() {
++ const url = (document.getElementById('ws-rs-server-url')?.value || '').trim();
++ const isSrt = /^srt:\/\//i.test(url);
++ const srt = document.getElementById('ws-rs-srt-fields');
++ if (srt) srt.style.display = isSrt ? '' : 'none';
++ const keyLabel = document.getElementById('ws-rs-stream-key-label');
++ if (keyLabel) keyLabel.textContent = isSrt ? 'Stream ID (streamid)' : 'Stream Key';
++ const hint = document.getElementById('ws-rs-transport-hint');
++ if (hint) hint.textContent = isSrt ? 'SRT: MPEG-TS over UDP in caller mode — lower latency than RTMP and survives packet loss.'
++ : (/^rtmps:\/\//i.test(url) ? 'RTMPS: RTMP over TLS.' : (url ? 'RTMP (FLV).' : ''));
++}
++
function _wsShowRestreamForm(existing) {
const isRsEdit = !!existing?.__rs;
if (isRsEdit) existing = null; // RS config comes from _wsSlotRsIntegration, not a destination row
@@@ -2910,12 +2923,25 @@ function _wsShowRestreamForm(existing) {
</div>
<div class="form-group">
<label>Server URL</label>
-- <input type="text" id="ws-rs-server-url" class="form-input form-input-sm" value="${esc(existing?.server_url || '')}" placeholder="rtmp://...">
++ <input type="text" id="ws-rs-server-url" class="form-input form-input-sm" value="${esc(existing?.server_url || '')}" placeholder="rtmp://… rtmps://… or srt://host:port" oninput="_wsSyncRestreamTransport()">
++ <small class="muted" id="ws-rs-transport-hint"></small>
</div>
<div class="form-group">
-- <label>Stream Key</label>
++ <label id="ws-rs-stream-key-label">Stream Key</label>
<input type="password" id="ws-rs-stream-key" class="form-input form-input-sm" value="${esc(existing?.stream_key || '')}" placeholder="Paste stream key" autocomplete="off">
</div>
++ <div id="ws-rs-srt-fields" class="bc-ws-row" style="display:none;gap:8px">
++ <div class="form-group" style="flex:1;margin:0">
++ <label style="font-size:0.78rem">SRT latency (ms)</label>
++ <input type="number" id="ws-rs-srt-latency" class="form-input form-input-sm" value="${existing?.srt_latency_ms || ''}" placeholder="120" min="20" max="8000">
++ <small class="muted">Receiver buffer. 120 ms same-continent; 300–600 ms across oceans or on Wi-Fi.</small>
++ </div>
++ <div class="form-group" style="flex:1;margin:0">
++ <label style="font-size:0.78rem">SRT passphrase <span class="muted">(optional, 10–79 chars)</span></label>
++ <input type="password" id="ws-rs-srt-passphrase" class="form-input form-input-sm" value="" placeholder="${existing?.has_srt_passphrase ? '•••••••• (saved — leave blank to keep)' : 'None (unencrypted)'}" autocomplete="off" minlength="10" maxlength="79">
++ ${existing?.has_srt_passphrase ? '<label class="bc-toggle-label" style="font-size:0.75rem;margin-top:4px"><input type="checkbox" id="ws-rs-srt-clear-pass"> Remove passphrase</label>' : ''}
++ </div>
++ </div>
<div class="form-group">
<label>Channel URL <span class="muted">(for chat relay)</span></label>
<input type="text" id="ws-rs-channel-url" class="form-input form-input-sm" value="${esc(existing?.channel_url || '')}" placeholder="https://twitch.tv/username">
@@@ -2969,7 +2995,14 @@ function _wsShowRestreamForm(existing) {
<label style="font-size:0.78rem">FPS</label>
<input type="number" id="ws-rs-fps" class="form-input form-input-sm" value="${existing?.custom_fps || ''}" placeholder="Auto" min="15" max="120">
</div>
++ <div class="form-group" style="flex:1;margin:0">
++ <label style="font-size:0.78rem" title="x264 speed/quality trade-off. Faster presets use less CPU (safer with several destinations); slower ones look better at the same bitrate.">Encoder speed</label>
++ <select id="ws-rs-encoder-preset" class="form-input form-input-sm">
++ ${['', 'ultrafast', 'superfast', 'veryfast', 'faster', 'fast', 'medium', 'slow'].map(v => `<option value="${v}" ${(existing?.custom_encoder_preset || '') === v ? 'selected' : ''}>${v ? v : 'Auto (per preset)'}</option>`).join('')}
++ </select>
++ </div>
</div>
++ <p class="muted" style="font-size:0.74rem;margin:6px 0 0">Browser and WHIP sources are re-encoded on the server with these settings; an OBS/RTMP source is forwarded as-is (codec copy), so only the destination sees them.</p>
</details>
</div>
<div class="bc-ws-confirm-actions" style="margin-top:16px">
@@@ -2981,6 +3014,7 @@ function _wsShowRestreamForm(existing) {
</div>`;
document.body.appendChild(overlay);
++ _wsSyncRestreamTransport();
// Apply platform-specific UI on open
_wsRestreamPlatformChanged();
@@@ -3036,8 +3070,18 @@ function _wsShowRestreamForm(existing) {
custom_video_bitrate: parseInt(document.getElementById('ws-rs-video-bitrate').value) || null,
custom_audio_bitrate: parseInt(document.getElementById('ws-rs-audio-bitrate').value) || null,
custom_fps: parseInt(document.getElementById('ws-rs-fps').value) || null,
++ custom_encoder_preset: document.getElementById('ws-rs-encoder-preset')?.value || null,
++ srt_latency_ms: parseInt(document.getElementById('ws-rs-srt-latency')?.value) || null,
managed_stream_id: _wsState.selectedId,
};
++ // Passphrase: blank keeps the saved one; the checkbox removes it; text replaces it.
++ const passEl = document.getElementById('ws-rs-srt-passphrase');
++ if (document.getElementById('ws-rs-srt-clear-pass')?.checked) body.srt_passphrase = '';
++ else if (passEl && passEl.value) body.srt_passphrase = passEl.value;
++ if (body.srt_passphrase && (body.srt_passphrase.length < 10 || body.srt_passphrase.length > 79)) {
++ toast('SRT passphrase must be 10–79 characters', 'error');
++ return;
++ }
if (!body.stream_key) {
toast('Stream key is required', 'error');
ddiff --git a/public/js/broadcast.js b/public/js/broadcast.js
iindex 00f9484f..79e17d90 100644
---- a/public/js/broadcast.js
++++ b/public/js/broadcast.js
@@@ -151,11 +151,32 @@ function stopRsRestream() {
const slot = getRsRestreamSlotStreamId();
if (slot) {
stopRobotStreamerRestream(slot).catch(() => {});
++ // The video actually leaves from the server: tell it to close the robot session, or
++ // RobotStreamer keeps showing the robot live with a black picture.
++ if (rsServerPassthrough()) {
++ api('/robotstreamer/restream/stop', { method: 'POST', body: { stream_id: slot } })
++ .then(() => { _lastRsServerStatuses[slot] = null; setRobotStreamerStatus('RobotStreamer restream stopped.', 'info', 'bc-rsLiveStatus'); updateRestreamControlPanel(); })
++ .catch(err => toast(err?.message || 'Could not stop the RobotStreamer restream', 'error'));
++ }
}
setRsRestreamSlot(null);
updateRsRestreamSlotUI();
}
++/** Ask the server to (re)start the passthrough for a stream and refresh the panel with its answer. */
++async function startRsServerRestream(streamId) {
++ try {
++ const data = await api('/robotstreamer/restream/start', { method: 'POST', body: { stream_id: streamId } });
++ if (data?.status) _lastRsServerStatuses[streamId] = data.status;
++ setRobotStreamerStatus('Restreaming to RobotStreamer via server (raw passthrough, no re-encode).', 'success', 'bc-rsLiveStatus');
++ } catch (err) {
++ setRobotStreamerStatus(err?.message || 'Could not start the RobotStreamer restream', 'error', 'bc-rsLiveStatus');
++ toast(err?.message || 'Could not start the RobotStreamer restream', 'error');
++ }
++ updateRestreamControlPanel();
++ updateRsRestreamSlotUI();
++}
++
/** Start or retry RS restream for the current slot / active stream */
function startOrRetryRsRestream() {
const slotStreamId = getRsRestreamSlotStreamId();
@@@ -169,6 +190,13 @@ function startOrRetryRsRestream() {
}
if (!targetId) return;
++ // Server passthrough: the server forwards whatever the stream's source is (browser, OBS/WHIP),
++ // so there is no local media requirement — just ask it to (re)start.
++ if (rsServerPassthrough()) {
++ setRsRestreamSlot(targetId);
++ startRsServerRestream(targetId);
++ return;
++ }
const ss = getStreamState(targetId);
if (!ss?.localStream) {
if (slotStreamId) {
@@@ -217,9 +245,11 @@ function updateRsRestreamSlotUI() {
const slotId = getRsRestreamSlotStreamId();
const slotSs = slotId ? getStreamState(slotId) : null;
-- const isLive = !!(slotSs?.robotStreamer?.active);
-- const isStarting = !!(slotSs?._rsRestreamStarting);
-- const hasFailed = !isLive && !isStarting && slotSs && !slotSs.robotStreamer && (slotSs._rsConsecutiveShortLived || 0) >= RS_MAX_SHORT_LIVED_RETRIES;
++ const rsSrv = rsServerPassthrough() ? rsServerStatusFor(slotId) : null;
++ const isLive = rsSrv ? rsSrv.state === 'live' : !!(slotSs?.robotStreamer?.active);
++ const isStarting = rsSrv ? (rsSrv.state === 'starting' || rsSrv.state === 'restarting') : !!(slotSs?._rsRestreamStarting);
++ const hasFailed = rsSrv ? rsSrv.state === 'failed'
++ : (!isLive && !isStarting && slotSs && !slotSs.robotStreamer && (slotSs._rsConsecutiveShortLived || 0) >= RS_MAX_SHORT_LIVED_RETRIES);
let html = '';
@@@ -263,6 +293,7 @@ function updateRsRestreamSlotUI() {
html += `<button class="bc-ctrl-btn bc-ctrl-btn-sm" onclick="startOrRetryRsRestream()"><i class="fa-solid fa-play"></i> Start</button>`;
}
html += '</div>';
++ if (rsSrv) html += rsDetailLine(rsSrv);
panel.innerHTML = html;
}
@@@ -3122,16 +3153,110 @@ async function stopRestreamDest(destId) {
* Called periodically while streaming.
*/
let _lastRestreamStatuses = {};
++/** streamId → server-side RobotStreamer passthrough status (null when no relay session). */
++let _lastRsServerStatuses = {};
async function pollRestreamStatus() {
try {
const data = await api('/restream/status');
_lastRestreamStatuses = data.statuses || {};
++ _lastRsServerStatuses = data.robotstreamer || {};
updateRestreamControlPanel();
++ updateRsRestreamSlotUI();
} catch {
// Silent — polling failure is non-critical
}
}
++/** Is the RobotStreamer video path the server-side passthrough (true) or the browser publisher? */
++function rsServerPassthrough() {
++ return !!broadcastState.robotStreamer?.passthrough;
++}
++
++/** Server passthrough status for the RS slot's stream (or the active stream). */
++function rsServerStatusFor(streamId) {
++ const id = streamId || getRsRestreamSlotStreamId() || broadcastState.activeStreamId;
++ if (id && _lastRsServerStatuses[id]) return _lastRsServerStatuses[id];
++ // The server auto-starts the passthrough on go-live (OBS/WHIP sources never assign a
++ // browser-side slot), so fall back to whichever of the user's live streams has a session.
++ for (const v of Object.values(_lastRsServerStatuses)) if (v) return v;
++ return null;
++}
++
++function fmtUptime(ms) {
++ const s = Math.max(0, Math.round(ms / 1000));
++ if (s < 60) return `${s}s`;
++ const m = Math.floor(s / 60), h = Math.floor(m / 60);
++ return h ? `${h}h ${m % 60}m` : `${m}m ${s % 60}s`;
++}
++
++/**
++ * One line of detail under a restream badge: what it is doing, or why it is not.
++ * Built from the server's status so the streamer never has to read the server log.
++ */
++function restreamDetailLine(st) {
++ if (!st) return '';
++ const bits = [];
++ const now = Date.now();
++ if (st.status === 'live') {
++ if (st.uptimeMs) bits.push(`up ${fmtUptime(st.uptimeMs)}`);
++ const p = st.progress;
++ if (p) {
++ if (p.fps) bits.push(`${Math.round(p.fps)} fps`);
++ if (p.bitrate_kbps) bits.push(`${Math.round(p.bitrate_kbps)} kbps`);
++ if (p.speed != null && p.speed < 0.95) bits.push(`<span class="bc-restream-warn">encoder at ${p.speed.toFixed(2)}× — falling behind</span>`);
++ if (p.dropped) bits.push(`${p.dropped} dropped`);
++ }
++ if (st.restartAttempts) bits.push(`${st.restartAttempts} reconnect${st.restartAttempts === 1 ? '' : 's'}`);
++ } else if (st.status === 'starting') {
++ bits.push('connecting to the ingest…');
++ } else if (st.status === 'error') {
++ if (st.lastError) bits.push(`<span class="bc-restream-err">${esc(st.lastError)}</span>`);
++ if (st.nextRestartAt && st.nextRestartAt > now) bits.push(`retry in ${Math.max(1, Math.round((st.nextRestartAt - now) / 1000))}s (${st.restartAttempts}/${st.maxRestartAttempts || '∞'})`);
++ } else if (st.status === 'failed') {
++ bits.push(`<span class="bc-restream-err">${esc(st.lastError || 'gave up after repeated failures')}</span>`);
++ } else if (st.status === 'cooldown') {
++ bits.push(`<span class="bc-restream-err">${esc(st.lastError || 'paused after repeated failures')}</span>`);
++ }
++ return bits.length ? `<div class="bc-restream-detail">${bits.join(' · ')}</div>` : '';
++}
++
++/** Same, for the server-side RobotStreamer passthrough status. */
++function rsDetailLine(rs) {
++ if (!rs) return '';
++ const bits = [];
++ const now = Date.now();
++ if (rs.state === 'live') {
++ bits.push(`up ${fmtUptime(rs.uptime_ms || 0)}`);
++ bits.push(rs.video_codec ? `${String(rs.video_codec).replace('video/', '')} passthrough` : 'passthrough');
++ if (!rs.has_audio) bits.push('<span class="bc-restream-warn">no audio</span>');
++ if (rs.rs) {
++ if (rs.rs.rtt_ms != null) bits.push(`${rs.rs.rtt_ms} ms to RS`);
++ if (rs.rs.loss_pct != null && rs.rs.loss_pct >= 1) bits.push(`<span class="bc-restream-warn">${rs.rs.loss_pct}% loss</span>`);
++ if (rs.rs.stalled_ms > 1500) bits.push(`<span class="bc-restream-warn">source stalled ${Math.round(rs.rs.stalled_ms / 1000)}s</span>`);
++ }
++ if (rs.restarts) bits.push(`${rs.restarts} reconnect${rs.restarts === 1 ? '' : 's'}`);
++ } else if (rs.state === 'starting') {
++ bits.push('connecting to RobotStreamer…');
++ } else if (rs.state === 'restarting') {
++ if (rs.last_restart_reason) bits.push(`<span class="bc-restream-err">${esc(rs.last_restart_reason)}</span>`);
++ if (rs.next_restart_at && rs.next_restart_at > now) bits.push(`retry in ${Math.max(1, Math.round((rs.next_restart_at - now) / 1000))}s (#${rs.restarts})`);
++ } else if (rs.state === 'failed') {
++ bits.push(`<span class="bc-restream-err">${esc(rs.last_error || 'gave up after repeated failures')}</span>`);
++ }
++ return bits.length ? `<div class="bc-restream-detail">${bits.join(' · ')}</div>` : '';
++}
++
++/** Badge class/icon/text for a server-side RS passthrough state. */
++function rsBadgeFor(rs, clientFallback) {
++ if (rs) {
++ if (rs.state === 'live') return { cls: 'bc-restream-status-live', icon: 'fa-solid fa-circle', text: 'Live', active: true };
++ if (rs.state === 'starting') return { cls: 'bc-restream-status-starting', icon: 'fa-solid fa-spinner fa-spin', text: 'Starting…', active: true };
++ if (rs.state === 'restarting') return { cls: 'bc-restream-status-starting', icon: 'fa-solid fa-spinner fa-spin', text: 'Reconnecting…', active: true };
++ if (rs.state === 'failed') return { cls: 'bc-restream-status-failed', icon: 'fa-solid fa-circle-xmark', text: 'Failed', active: false };
++ }
++ return clientFallback;
++}
++
/** Start polling restream status (called when going live) */
function startRestreamStatusPolling() {
stopRestreamStatusPolling();
@@@ -3200,51 +3325,57 @@ function updateRestreamControlPanel() {
badgeText = 'Failed';
}
-- const isActive = statusLabel === 'live' || statusLabel === 'starting';
++ else if (statusLabel === 'cooldown' || (!status && dest.cooldown_ms > 0)) {
++ badgeClass = 'bc-restream-status-failed';
++ badgeIcon = 'fa-solid fa-pause';
++ badgeText = 'Paused';
++ }
-- html += `<div class="bc-restream-row">
++ const isActive = statusLabel === 'live' || statusLabel === 'starting' || statusLabel === 'error';
++ const transport = dest.transport === 'srt' ? '<span class="bc-restream-transport" title="SRT (MPEG-TS)">SRT</span>' : '';
++ const detail = status ? restreamDetailLine(status)
++ : (dest.cooldown_ms > 0 ? `<div class="bc-restream-detail"><span class="bc-restream-err">${esc(dest.last_error || 'paused after repeated failures')}</span> · retry in ~${Math.max(1, Math.ceil(dest.cooldown_ms / 60000))}m</div>` : '');
++
++ html += `<div class="bc-restream-row" data-dest="${dest.id}">
<span class="bc-restream-platform-icon" style="color:${meta.color}"><i class="${meta.icon}"></i></span>
-- <span class="bc-restream-name">${dest.name || meta.name}</span>
++ <span class="bc-restream-name">${esc(dest.name || meta.name)}${transport}</span>
<span class="bc-restream-status-badge ${badgeClass}"><i class="${badgeIcon}"></i> ${badgeText}</span>
<div class="bc-restream-actions">
${isActive
-- ? `<button class="bc-ctrl-btn-sm" onclick="stopRestreamDest(${dest.id})"><i class="fa-solid fa-stop"></i> Stop</button>`
-- : `<button class="bc-ctrl-btn-sm" onclick="startRestreamDest(${dest.id})"><i class="fa-solid fa-play"></i> Start</button>`
++ ? `<button class="bc-ctrl-btn-sm" onclick="stopRestreamDest(${dest.id})" title="Stop pushing to this destination"><i class="fa-solid fa-stop"></i> Stop</button>`
++ : `<button class="bc-ctrl-btn-sm" onclick="startRestreamDest(${dest.id})" title="${statusLabel === 'failed' || dest.cooldown_ms > 0 ? 'Clear the failure and try again' : 'Start pushing to this destination'}"><i class="fa-solid ${statusLabel === 'failed' || dest.cooldown_ms > 0 ? 'fa-rotate-right' : 'fa-play'}"></i> ${statusLabel === 'failed' || dest.cooldown_ms > 0 ? 'Retry' : 'Start'}</button>`
}
</div>
++ ${detail}
</div>`;
}
-- // RS status row (if configured)
++ // RS status row (if configured). With the server passthrough the truth lives on the server
++ // (polled every 5s); the browser publisher's own state is the fallback.
if (hasRs) {
const slotId = getRsRestreamSlotStreamId();
const slotSs = slotId ? getStreamState(slotId) : null;
const isRsLive = !!(slotSs?.robotStreamer?.active);
const isRsStarting = !!(slotSs?._rsRestreamStarting);
++ const rsSrv = rsServerPassthrough() ? rsServerStatusFor(slotId) : null;
-- let rsBadgeClass = 'bc-restream-status-idle';
-- let rsBadgeIcon = 'fa-solid fa-circle';
-- let rsBadgeText = 'Idle';
--
-- if (isRsLive) {
-- rsBadgeClass = 'bc-restream-status-live';
-- rsBadgeText = 'Live';
-- } else if (isRsStarting || slotSs?._rsReconnectTimer) {
-- rsBadgeClass = 'bc-restream-status-starting';
-- rsBadgeIcon = 'fa-solid fa-spinner fa-spin';
-- rsBadgeText = isRsStarting ? 'Starting…' : 'Reconnecting…';
-- }
++ let fallback = { cls: 'bc-restream-status-idle', icon: 'fa-solid fa-circle', text: 'Idle', active: false };
++ if (isRsLive) fallback = { cls: 'bc-restream-status-live', icon: 'fa-solid fa-circle', text: 'Live', active: true };
++ else if (isRsStarting || slotSs?._rsReconnectTimer) fallback = { cls: 'bc-restream-status-starting', icon: 'fa-solid fa-spinner fa-spin', text: isRsStarting ? 'Starting…' : 'Reconnecting…', active: true };
++ const b = rsBadgeFor(rsSrv, fallback);
++ const robot = broadcastState.robotStreamer?.robotId || rsSrv?.robot_id;
-- html += `<div class="bc-restream-row">
++ html += `<div class="bc-restream-row" data-dest="rs">
<span class="bc-restream-platform-icon" style="color:#4a9eff"><i class="fa-solid fa-robot"></i></span>
-- <span class="bc-restream-name">RobotStreamer</span>
-- <span class="bc-restream-status-badge ${rsBadgeClass}"><i class="${rsBadgeIcon}"></i> ${rsBadgeText}</span>
++ <span class="bc-restream-name">RobotStreamer${robot ? ` <span class="bc-restream-transport" title="Robot id">#${esc(String(robot))}</span>` : ''}</span>
++ <span class="bc-restream-status-badge ${b.cls}"><i class="${b.icon}"></i> ${b.text}</span>
<div class="bc-restream-actions">
-- ${isRsLive
-- ? `<button class="bc-ctrl-btn-sm" onclick="stopRsRestream()"><i class="fa-solid fa-stop"></i> Stop</button>`
-- : `<button class="bc-ctrl-btn-sm" onclick="startOrRetryRsRestream()"><i class="fa-solid fa-play"></i> Start</button>`
++ ${b.active
++ ? `<button class="bc-ctrl-btn-sm" onclick="stopRsRestream()" title="Take the robot offline"><i class="fa-solid fa-stop"></i> Stop</button>`
++ : `<button class="bc-ctrl-btn-sm" onclick="startOrRetryRsRestream()" title="${rsSrv?.state === 'failed' ? 'Try again' : 'Send this stream to the robot'}"><i class="fa-solid ${rsSrv?.state === 'failed' ? 'fa-rotate-right' : 'fa-play'}"></i> ${rsSrv?.state === 'failed' ? 'Retry' : 'Start'}</button>`
}
</div>
++ ${rsDetailLine(rsSrv)}
</div>`;
}
ddiff --git a/server/admin/routes.js b/server/admin/routes.js
iindex d1e25c21..08b4d849 100644
---- a/server/admin/routes.js
++++ b/server/admin/routes.js
@@@ -324,6 +324,16 @@ router.delete('/streams/:id', (req, res) => {
}
db.endStream(req.params.id);
++ // A force-end must actually take the stream down everywhere, not just flip is_live:
++ // outbound forwarders (RobotStreamer passthrough, ffmpeg restreams, chat relays), the
++ // signaling room and its viewers, the SFU room and any voice channel.
++ try { require('../integrations/robotstreamer-service').stopForStream(stream.id); } catch (e) { console.warn('[Admin] RS stop failed:', e.message); }
++ try { require('../streaming/restream-manager').stopAllForStream(stream.id); } catch (e) { console.warn('[Admin] restream stop failed:', e.message); }
++ try { require('../integrations/chat-relay-service').stopForStream(stream.id); } catch { /* non-critical */ }
++ try { require('../integrations/ai-chatbot-service').stopForStream(stream.id); } catch { /* non-critical */ }
++ try { require('../streaming/broadcast-server').endStream(stream.id); } catch { /* no room */ }
++ try { require('../streaming/webrtc-sfu').closeRoom(`stream-${stream.id}`); } catch { /* no room */ }
++ try { require('../streaming/call-server').removeStreamChannel(stream.id); } catch { /* no channel */ }
// Accountability: force-ending someone's live stream is a moderation action
// and MUST be logged (previously it left no trace at all).
ddiff --git a/server/config.js b/server/config.js
iindex 1ad6d103..df812687 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -238,7 +238,6 @@ function buildConfig(registryValues) {
// the ffmpeg transcode worker. This is now the ONLY restream path and is ON by
// default; set RS_PASSTHROUGH=0 only as an emergency kill-switch.
passthrough: process.env.RS_PASSTHROUGH !== '0',
-- passthroughRobots: (process.env.RS_PASSTHROUGH_ROBOTS || '').split(',').map(s => s.trim()).filter(Boolean),
},
openvibeBucks: {
// Bit-style currency: integer bucks, 100 bucks = $1.00 streamer cashout
ddiff --git a/server/db/database.js b/server/db/database.js
iindex fbdb8903..21f8f21b 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -805,6 +805,9 @@ function initDb() {
{ name: 'custom_audio_bitrate', def: 'INTEGER DEFAULT NULL' },
{ name: 'custom_fps', def: 'INTEGER DEFAULT NULL' },
{ name: 'custom_encoder_preset', def: 'TEXT DEFAULT NULL' },
++ // SRT destinations (srt:// server URL): receiver latency window and optional encryption.
++ { name: 'srt_latency_ms', def: 'INTEGER DEFAULT NULL' },
++ { name: 'srt_passphrase', def: 'TEXT DEFAULT NULL' },
];
for (const col of newCols) {
if (!cols.includes(col.name)) {
@@@ -4702,13 +4705,14 @@ function getRestreamDestinationById(id) {
function createRestreamDestination(userId, fields) {
const result = run(
`INSERT INTO restream_destinations (user_id, managed_stream_id, platform, name, server_url, stream_key, enabled, auto_start, quality_preset,
-- custom_video_bitrate, custom_audio_bitrate, custom_fps, custom_encoder_preset, channel_url, chat_relay, powerchat_relay, powerchat_count_views)
-- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
++ custom_video_bitrate, custom_audio_bitrate, custom_fps, custom_encoder_preset, srt_latency_ms, srt_passphrase, channel_url, chat_relay, powerchat_relay, powerchat_count_views)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
[userId, fields.managed_stream_id || null, fields.platform, fields.name || null, fields.server_url || null,
fields.stream_key || null, fields.enabled ?? 1, fields.auto_start ?? 0,
fields.quality_preset || 'auto',
fields.custom_video_bitrate ?? null, fields.custom_audio_bitrate ?? null,
fields.custom_fps ?? null, fields.custom_encoder_preset || null,
++ fields.srt_latency_ms ?? null, fields.srt_passphrase || null,
fields.channel_url || null, fields.chat_relay ? 1 : 0, fields.powerchat_relay === 0 ? 0 : 1, fields.powerchat_count_views === 0 ? 0 : 1]
);
return get('SELECT * FROM restream_destinations WHERE id = ?', [result.lastInsertRowid]);
@@@ -4716,7 +4720,7 @@ function createRestreamDestination(userId, fields) {
function updateRestreamDestination(id, fields) {
const allowed = new Set(['name', 'server_url', 'stream_key', 'enabled', 'auto_start', 'quality_preset',
-- 'custom_video_bitrate', 'custom_audio_bitrate', 'custom_fps', 'custom_encoder_preset',
++ 'custom_video_bitrate', 'custom_audio_bitrate', 'custom_fps', 'custom_encoder_preset', 'srt_latency_ms', 'srt_passphrase',
'channel_url', 'chat_relay', 'powerchat_relay', 'powerchat_count_views', 'managed_stream_id', 'connection_id']);
const filtered = Object.entries(fields || {}).filter(([key]) => allowed.has(key));
if (!filtered.length) return getRestreamDestinationById(id);
ddiff --git a/server/index.js b/server/index.js
iindex a0ffc9e8..df083c4a 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -1169,6 +1169,10 @@ async function start() {
});
rtmpServer.on('unpublish', ({ streamId }) => {
restreamManager.stopAllForStream(streamId);
++ // OBS stopped: the RS chat mirror/passthrough, chat relays and AI bots end with the ingest.
++ try { robotStreamerService.stopForStream(streamId); } catch (err) { console.warn('[RS] stop on unpublish failed:', err.message); }
++ try { chatRelayService.stopForStream(streamId); } catch { /* non-critical */ }
++ try { require('./integrations/ai-chatbot-service').stopForStream(streamId); } catch { /* non-critical */ }
});
// 6c. Hook WebRTC SFU events for auto-start restreams
ddiff --git a/server/integrations/robotstreamer-service.js b/server/integrations/robotstreamer-service.js
iindex 0789542b..51456961 100644
---- a/server/integrations/robotstreamer-service.js
++++ b/server/integrations/robotstreamer-service.js
@@@ -512,38 +512,6 @@ class RobotStreamerService {
}, slotId);
}
-- /**
-- * Start the native (server-side) RS video publisher for streams whose
-- * ingest has no browser publisher: WHIP/OBS and RTMP. Browser broadcasts
-- * publish to RS from the client via /ws/robotstreamer-publish instead.
-- */
-- _maybeStartNativePublish(stream, integration, opts = {}) {
-- try {
-- if (!integration?.enabled || !integration.token || !integration.robot_id) return;
-- if (this._activePublish.has(stream.id)) return; // browser publisher already connected
-- const rsNativePublisher = require('./rs-native-publisher');
-- if (rsNativePublisher.isActive(stream.id)) return;
--
-- // Determine the real ingest. When the caller KNOWS the ingest (e.g. we
-- // were triggered from the WHIP ICE-connected handler), trust that over
-- // the configured managed_stream.streaming_method — which can be stale
-- // or misconfigured ('browser') even though OBS is pushing via WHIP.
-- const db2 = require('../db/database');
-- const ms = stream.managed_stream_id ? db2.getManagedStreamById(stream.managed_stream_id) : null;
-- let method = ms?.streaming_method || 'browser';
-- if (opts.ingest === 'whip' || opts.ingest === 'rtmp') method = opts.ingest;
-- if (method !== 'whip' && method !== 'rtmp') {
-- console.log(`[RS] Native publish skipped for stream ${stream.id}: method='${method}' (no browserless ingest)`);
-- return;
-- }
--
-- console.log(`[RS] Starting native video publish for stream ${stream.id} (ingest=${method}, robot=${integration.robot_id})`);
-- rsNativePublisher.start(stream, integration);
-- } catch (err) {
-- console.warn(`[RS] Native publish start failed for stream ${stream.id}:`, err.message);
-- }
-- }
--
/**
* Idempotent per stream, including while a previous call is still in flight.
*
@@@ -579,7 +547,11 @@ class RobotStreamerService {
// RAW passthrough relay (zero re-encode) is the ONLY restream path now: the server
// forwards the source's already-encoded RTP straight to RobotStreamer, replacing both
// the browser's second encode and the old ffmpeg transcode publisher.
-- if (this._passthroughEnabledFor(integration)) {
++ if (stream.protocol === 'rtmp' || stream.protocol === 'jsmpeg') {
++ // No SFU producers to forward: the relay would sit waiting for a video producer,
++ // time out and restart in a loop. Chat mirroring below still applies.
++ console.log(`[RS] Video passthrough skipped for stream ${stream.id}: ${stream.protocol} ingest has no SFU producer (browser/WHIP sources only)`);
++ } else if (this._passthroughEnabledFor(integration)) {
try {
require('./rs-passthrough-relay').start(stream, integration);
console.log(`[RS] Raw passthrough relay started for stream ${stream.id} (robot ${integration.robot_id})`);
@@@ -819,7 +791,6 @@ class RobotStreamerService {
stopForStream(streamId) {
this.stopChatBridge(streamId);
-- try { require('./rs-native-publisher').stop(streamId); } catch { /* ignore */ }
try { require('./rs-passthrough-relay').stop(streamId); } catch { /* ignore */ }
}
ddiff --git a/server/integrations/routes.js b/server/integrations/routes.js
iindex a9851173..b4e94e4b 100644
---- a/server/integrations/routes.js
++++ b/server/integrations/routes.js
@@@ -180,6 +180,48 @@ router.put('/integration', requireAuth, async (req, res) => {
}
});
++// ── Server-side RobotStreamer restream control ───────────────
++// The video passthrough runs on the server, so the dashboard's Start/Stop must reach it: before
++// these routes existed the button only touched the browser's own (unused) publisher, and the
++// only way to stop the robot was to disable the whole integration.
++function ownLiveStream(req, res) {
++ const id = parseInt(req.body?.stream_id, 10);
++ const stream = Number.isFinite(id) ? db.getStreamById(id) : null;
++ if (!stream || stream.user_id !== req.user.id) { res.status(404).json({ error: 'Stream not found' }); return null; }
++ return stream;
++}
++
++router.post('/restream/start', requireAuth, async (req, res) => {
++ try {
++ const stream = ownLiveStream(req, res);
++ if (!stream) return;
++ if (!stream.is_live) return res.status(409).json({ error: 'Stream is not live' });
++ const integration = robotStreamerService.getIntegrationForStream(stream);
++ if (!integration?.enabled || !integration.token || !integration.robot_id) {
++ return res.status(400).json({ error: 'RobotStreamer is not configured (or disabled) for this stream' });
++ }
++ if (!['webrtc', 'whip'].includes(stream.protocol) && !require('../streaming/webrtc-sfu').hasProducers(`stream-${stream.id}`)) {
++ return res.status(400).json({ error: `RobotStreamer passthrough needs a browser or OBS/WHIP source; this stream is ${stream.protocol}` });
++ }
++ await robotStreamerService.startForStream(stream);
++ const relay = require('./rs-passthrough-relay');
++ res.json({ ok: true, status: relay.status(stream.id) });
++ } catch (err) {
++ res.status(500).json({ error: err.message || 'Failed to start RobotStreamer restream' });
++ }
++});
++
++router.post('/restream/stop', requireAuth, (req, res) => {
++ try {
++ const stream = ownLiveStream(req, res);
++ if (!stream) return;
++ require('./rs-passthrough-relay').stop(stream.id);
++ res.json({ ok: true });
++ } catch (err) {
++ res.status(500).json({ error: err.message || 'Failed to stop RobotStreamer restream' });
++ }
++});
++
// Remove a slot-specific RS config (the slot falls back to the account default)
router.delete('/integration', requireAuth, (req, res) => {
try {
ddiff --git a/server/integrations/rs-native-publisher-worker.js b/server/integrations/rs-native-publisher-worker.js
ddeleted file mode 100644
iindex 8320b0c1..00000000
---- a/server/integrations/rs-native-publisher-worker.js
++++ /dev/null
@@@ -1,569 +0,0 @@
--/**
-- * OpenVibe.Live — RobotStreamer native publisher (worker process)
-- *
-- * Publishes a server-ingested stream (WHIP/OBS or RTMP — no browser publisher
-- * available) to the RobotStreamer mediasoup SFU using node WebRTC.
-- *
-- * Media path:
-- * local SFU PlainRtpTransport (SDP) or NMS HTTP-FLV → ffmpeg
-- * → rawvideo I420 pipe → wrtc RTCVideoSource ┐
-- * → s16le PCM pipe → wrtc RTCAudioSource ┴→ mediasoup-client → RS SFU
-- *
-- * Runs as a child process (spawned by rs-native-publisher.js) so the native
-- * wrtc module and the browser-ish globals mediasoup-client needs stay out of
-- * the main server process. Configuration comes via environment variables:
-- *
-- * RS_TOKEN, ROBOT_ID — RobotStreamer credentials (per stream slot)
-- * INPUT_MODE — 'sdp' | 'url'
-- * INPUT_PATH — SDP file path or FLV url
-- * HAS_AUDIO — '1' | '0'
-- * WIDTH, HEIGHT, FPS — output frame geometry (default 1280x720@30)
-- * VIDEO_KBPS, MIN_VIDEO_KBPS
-- *
-- * Adapted from openvibe.games's rs-browser-to-robotstreamer-native publisher.
-- */
--'use strict';
--
--const https = require('node:https');
--const crypto = require('node:crypto');
--const { spawn } = require('node:child_process');
--const WebSocket = require('ws');
--const wrtc = require('@roamhq/wrtc');
--const { Device } = require('mediasoup-client');
--
--global.RTCPeerConnection = wrtc.RTCPeerConnection;
--global.RTCSessionDescription = wrtc.RTCSessionDescription;
--global.RTCIceCandidate = wrtc.RTCIceCandidate;
--global.MediaStreamTrack = wrtc.MediaStreamTrack;
--global.MediaStream = wrtc.MediaStream;
--global.RTCRtpSender = wrtc.RTCRtpSender;
--global.RTCRtpReceiver = wrtc.RTCRtpReceiver;
--global.RTCRtpTransceiver = wrtc.RTCRtpTransceiver;
--
--if (!global.MediaStream) {
-- global.MediaStream = class MediaStream {
-- constructor(tracks = []) {
-- this._tracks = Array.from(tracks);
-- this.id = `stream-${Math.random().toString(16).slice(2)}`;
-- }
-- addTrack(track) { if (!this._tracks.includes(track)) this._tracks.push(track); }
-- removeTrack(track) { this._tracks = this._tracks.filter(t => t !== track); }
-- getTracks() { return this._tracks.slice(); }
-- getAudioTracks() { return this._tracks.filter(t => t.kind === 'audio'); }
-- getVideoTracks() { return this._tracks.filter(t => t.kind === 'video'); }
-- };
--}
--
--// Node 22 exposes globalThis.navigator via a getter-only property —
--// assignment throws, so use defineProperty instead.
--const FAKE_UA = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/131 Safari/537.36';
--Object.defineProperty(global, 'navigator', {
-- value: { userAgent: FAKE_UA, mediaDevices: {} },
-- configurable: true,
-- writable: true,
--});
--
--const {
-- RS_TOKEN,
-- ROBOT_ID,
-- INPUT_MODE,
-- INPUT_PATH,
-- HAS_AUDIO,
--} = process.env;
--
--if (!RS_TOKEN) throw new Error('Missing RS_TOKEN');
--if (!ROBOT_ID) throw new Error('Missing ROBOT_ID');
--if (!INPUT_MODE || !INPUT_PATH) throw new Error('Missing INPUT_MODE/INPUT_PATH');
--
--const width = Number(process.env.WIDTH || 1280);
--const height = Number(process.env.HEIGHT || 720);
--const fps = Number(process.env.FPS || 30);
--const videoKbps = Number(process.env.VIDEO_KBPS || 3500);
--const minVideoKbps = Number(process.env.MIN_VIDEO_KBPS || 1200);
--const hasAudio = String(HAS_AUDIO || '1') === '1';
--const frameSize = Math.floor(width * height * 3 / 2);
--
--let shuttingDown = false;
--let videoPumpTimer = null; // steady encode pacer (cleared on shutdown / ffmpeg exit)
--
--const status = {
-- videoFramesIn: 0,
-- videoFramesPushed: 0,
-- videoFramesDropped: 0,
-- audioFramesPushed: 0,
-- connected: false,
--};
--
--function log(...args) {
-- console.log(new Date().toISOString(), ...args);
--}
--
--function postJson(hostname, path, body) {
-- const payload = JSON.stringify(body);
-- return new Promise((resolve, reject) => {
-- const req = https.request({
-- hostname,
-- port: 443,
-- path,
-- method: 'POST',
-- headers: {
-- 'Content-Type': 'application/json',
-- 'Content-Length': Buffer.byteLength(payload),
-- 'Origin': 'https://robotstreamer.com',
-- 'Referer': 'https://robotstreamer.com/',
-- },
-- timeout: 15000,
-- }, res => {
-- let raw = '';
-- res.on('data', chunk => { raw += chunk; });
-- res.on('end', () => {
-- if (res.statusCode >= 400) {
-- reject(new Error(`HTTP ${res.statusCode}: ${raw.slice(0, 500)}`));
-- return;
-- }
-- try { resolve(JSON.parse(raw)); }
-- catch (err) { reject(new Error(`JSON parse failed: ${err.message}: ${raw.slice(0, 500)}`)); }
-- });
-- });
-- req.on('error', reject);
-- req.on('timeout', () => req.destroy(new Error('API timeout')));
-- req.write(payload);
-- req.end();
-- });
--}
--
--class ProtooPeer {
-- constructor(url) {
-- this.url = url;
-- this.ws = null;
-- this.nextId = Math.floor(Math.random() * 9000000) + 100000;
-- this.pending = new Map();
-- }
--
-- connect() {
-- return new Promise((resolve, reject) => {
-- this.ws = new WebSocket(this.url, ['protoo'], {
-- headers: {
-- Origin: 'https://robotstreamer.com',
-- Referer: `https://robotstreamer.com/robot/${ROBOT_ID}`,
-- 'User-Agent': global.navigator.userAgent,
-- },
-- rejectUnauthorized: false,
-- handshakeTimeout: 15000,
-- perMessageDeflate: false,
-- });
--
-- const timer = setTimeout(() => reject(new Error('SFU websocket timeout')), 15000);
--
-- this.ws.on('open', () => {
-- clearTimeout(timer);
-- log('[sfu] upstream open');
-- resolve();
-- });
-- this.ws.on('message', data => this._onMessage(data.toString()));
-- this.ws.on('error', err => {
-- log('[sfu] websocket error:', err.message);
-- reject(err);
-- });
-- this.ws.on('close', (code, reason) => {
-- log('[sfu] websocket closed:', code, reason.toString());
-- for (const { reject: rej } of this.pending.values()) rej(new Error(`websocket closed ${code}`));
-- this.pending.clear();
-- if (!shuttingDown) {
-- // Upstream dropped us — let the parent restart the whole pipeline.
-- process.exit(2);
-- }
-- });
-- });
-- }
--
-- _onMessage(raw) {
-- let msg;
-- try { msg = JSON.parse(raw); } catch { return; }
--
-- if (msg.response && this.pending.has(msg.id)) {
-- const pending = this.pending.get(msg.id);
-- this.pending.delete(msg.id);
-- if (msg.ok) pending.resolve(msg.data);
-- else pending.reject(new Error(msg.errorReason || msg.errorCode || `protoo request ${msg.id} failed`));
-- return;
-- }
--
-- if (msg.request) {
-- this.ws.send(JSON.stringify({ response: true, id: msg.id, ok: true, data: {} }));
-- }
-- }
--
-- request(method, data = {}) {
-- const id = this.nextId++;
-- return new Promise((resolve, reject) => {
-- this.pending.set(id, { resolve, reject });
-- this.ws.send(JSON.stringify({ request: true, id, method, data }));
-- setTimeout(() => {
-- if (!this.pending.has(id)) return;
-- this.pending.delete(id);
-- reject(new Error(`protoo request timeout: ${method}`));
-- }, 15000);
-- });
-- }
--}
--
--function createDevice() {
-- const names = ['Chrome111', 'Chrome74', undefined];
-- let lastErr;
-- for (const handlerName of names) {
-- try {
-- return handlerName ? new Device({ handlerName }) : new Device();
-- } catch (err) {
-- lastErr = err;
-- }
-- }
-- throw lastErr;
--}
--
--/**
-- * One ffmpeg for both outputs: rawvideo on fd 1 (stdout), s16le PCM on fd 3.
-- * SDP inputs (RTP ports) can only be bound once, so a single process must
-- * demux both tracks.
-- */
--function spawnFfmpeg(videoSource, audioSource) {
-- const inputArgs = INPUT_MODE === 'sdp'
-- ? [
-- '-protocol_whitelist', 'file,rtp,udp',
-- // Big input queue + UDP socket receive buffer so the local SFU→ffmpeg RTP path
-- // (no NACK/retransmit, unlike browser consumers) doesn't drop packets when the
-- // encoder briefly backs up — that packet loss was corrupting frames → choppy.
-- '-thread_queue_size', '8192',
-- '-buffer_size', '16000000',
-- '-analyzeduration', '2000000',
-- '-probesize', '2000000',
-- '-fflags', '+genpts+discardcorrupt+nobuffer+igndts',
-- '-flags', 'low_delay',
-- // The SFU→ffmpeg leg is a single in-order localhost RTP flow, so large reorder
-- // buffering is almost pure added latency. Keep it small (60ms / 16 pkts) — enough
-- // to smooth the rare out-of-order packet without inflating glass-to-glass delay.
-- '-max_delay', '60000',
-- '-reorder_queue_size', '16',
-- '-use_wallclock_as_timestamps', '1',
-- '-err_detect', 'ignore_err',
-- '-avoid_negative_ts', 'make_zero',
-- '-i', INPUT_PATH,
-- ]
-- : [
-- '-fflags', 'nobuffer',
-- '-flags', 'low_delay',
-- '-i', INPUT_PATH,
-- ];
--
-- const args = [
-- '-hide_banner',
-- '-loglevel', 'warning',
-- '-nostdin',
-- ...inputArgs,
-- '-map', '0:v:0',
-- '-vf', `scale=${width}:${height}:force_original_aspect_ratio=decrease,pad=${width}:${height}:(ow-iw)/2:(oh-ih)/2,fps=${fps},format=yuv420p`,
-- '-pix_fmt', 'yuv420p',
-- '-f', 'rawvideo',
-- 'pipe:1',
-- ];
--
-- if (hasAudio) {
-- args.push(
-- '-map', '0:a:0',
-- '-vn',
-- '-af', 'aresample=async=1:first_pts=0:min_hard_comp=0.100,asetnsamples=n=960:p=1',
-- '-ac', '2',
-- '-ar', '48000',
-- '-f', 's16le',
-- 'pipe:3',
-- );
-- }
--
-- log('[ffmpeg] starting:', 'ffmpeg', require('../utils/redact').redactUrl(args.join(' ')));
--
-- const ff = spawn('ffmpeg', args, {
-- stdio: hasAudio ? ['ignore', 'pipe', 'inherit', 'pipe'] : ['ignore', 'pipe', 'inherit'],
-- });
--
-- // ── Video: fixed-size I420 frames → RTCVideoSource, evenly paced ──
-- // Ping-pong two preallocated buffers instead of allocating a fresh ~1.3MB buffer per frame.
-- // At 30fps that was ~40MB/s of allocation → GC pauses that stalled the event loop, backed up
-- // the RTP read, and caused UDP packet loss. wrtc.onFrame copies the I420 data synchronously.
-- //
-- // ffmpeg emits frames to stdout in bursts (a 'data' chunk often carries several frames back
-- // to back). Handing those straight to the encoder makes libwebrtc timestamp them by ARRIVAL,
-- // so a burst becomes a cluster of near-simultaneous frames followed by a gap = visible judder
-- // / "bad framerate" on RobotStreamer even though the source is smooth. Instead we ASSEMBLE
-- // frames here and let a steady 1000/fps pacer hand the newest complete frame to the encoder,
-- // so RS receives evenly-spaced frames (smooth motion). A brief source stall re-sends the last
-- // frame (holds framerate instead of freezing); a burst just skips intermediates (we only ever
-- // expose the latest). The single-threaded event loop guarantees the pacer and this assembler
-- // never overlap, so two buffers suffice — the write target is always != the exposed frame.
-- const frameBufs = [Buffer.allocUnsafe(frameSize), Buffer.allocUnsafe(frameSize)];
-- let bufIdx = 0;
-- let frameBuf = frameBufs[0];
-- let frameOffset = 0;
-- let latestFrame = null; // most-recent complete frame, ready to encode
--
-- ff.stdout.on('data', chunk => {
-- let pos = 0;
-- while (pos < chunk.length) {
-- const need = frameSize - frameOffset;
-- const take = Math.min(need, chunk.length - pos);
-- chunk.copy(frameBuf, frameOffset, pos, pos + take);
-- frameOffset += take;
-- pos += take;
--
-- if (frameOffset === frameSize) {
-- status.videoFramesIn++;
-- latestFrame = frameBuf; // expose the just-completed buffer
-- bufIdx ^= 1; // assemble the next frame in the other buffer
-- frameBuf = frameBufs[bufIdx];
-- frameOffset = 0;
-- }
-- }
-- });
--
-- // Steady encode pacer — feed the newest complete frame at the target cadence so libwebrtc
-- // stamps evenly-spaced timestamps regardless of ffmpeg's bursty stdout delivery.
-- if (videoPumpTimer) clearInterval(videoPumpTimer);
-- videoPumpTimer = setInterval(() => {
-- if (!latestFrame) return;
-- try {
-- videoSource.onFrame({ width, height, data: latestFrame });
-- status.videoFramesPushed++;
-- } catch (err) {
-- log('[video] onFrame failed:', err.message);
-- }
-- }, Math.max(1, Math.round(1000 / fps)));
--
-- // ── Audio: 10ms PCM blocks → RTCAudioSource ─────────────────
-- if (hasAudio && ff.stdio[3]) {
-- const bytesPer10ms = 480 * 2 * 2; // 480 samples × 2ch × 2 bytes
-- let accum = Buffer.alloc(0);
--
-- ff.stdio[3].on('data', chunk => {
-- accum = Buffer.concat([accum, chunk]);
--
-- // Latency cap: if backed up over 80ms, keep the newest 20ms.
-- const maxBytes = bytesPer10ms * 8;
-- if (accum.length > maxBytes) {
-- accum = accum.subarray(accum.length - bytesPer10ms * 2);
-- }
--
-- while (accum.length >= bytesPer10ms) {
-- const block = accum.subarray(0, bytesPer10ms);
-- accum = accum.subarray(bytesPer10ms);
--
-- const samples = new Int16Array(480 * 2);
-- for (let i = 0; i < samples.length; i++) {
-- samples[i] = block.readInt16LE(i * 2);
-- }
--
-- try {
-- audioSource.onData({
-- samples,
-- sampleRate: 48000,
-- bitsPerSample: 16,
-- channelCount: 2,
-- numberOfFrames: 480,
-- });
-- status.audioFramesPushed++;
-- } catch (err) {
-- log('[audio] onData failed:', err.message);
-- }
-- }
-- });
-- }
--
-- ff.on('exit', (code, signal) => {
-- log('[ffmpeg] exited:', { code, signal });
-- if (videoPumpTimer) { clearInterval(videoPumpTimer); videoPumpTimer = null; }
-- if (!shuttingDown) process.exit(3);
-- });
--
-- return ff;
--}
--
--function waitForVideoFrames(minFrames = 5, timeoutMs = 15000) {
-- const start = Date.now();
-- return new Promise(resolve => {
-- const timer = setInterval(() => {
-- if (status.videoFramesIn >= minFrames || Date.now() - start >= timeoutMs) {
-- clearInterval(timer);
-- log('[video] warmup:', { framesIn: status.videoFramesIn, waitedMs: Date.now() - start });
-- resolve();
-- }
-- }, 100);
-- });
--}
--
--async function applySenderParams(sender, kind) {
-- if (!sender || typeof sender.getParameters !== 'function') return;
-- for (let attempt = 1; attempt <= 5; attempt++) {
-- try {
-- const params = sender.getParameters();
-- params.degradationPreference = 'maintain-framerate';
-- params.priority = 'high';
-- if (!params.encodings || !params.encodings.length) params.encodings = [{}];
-- for (const enc of params.encodings) {
-- enc.active = true;
-- enc.priority = 'high';
-- enc.networkPriority = 'high';
-- if (kind === 'video') {
-- enc.maxBitrate = videoKbps * 1000;
-- enc.minBitrate = minVideoKbps * 1000;
-- enc.maxFramerate = fps;
-- enc.scaleResolutionDownBy = 1;
-- }
-- }
-- await sender.setParameters(params);
-- return;
-- } catch (err) {
-- log(`[${kind}] setParameters attempt ${attempt} failed:`, err.message);
-- await new Promise(r => setTimeout(r, 250));
-- }
-- }
--}
--
--async function main() {
-- log(`RobotStreamer page load (robot ${ROBOT_ID})...`);
-- const pageData = await postJson('api.robotstreamer.com', '/v1/robot_page_load', {
-- token: RS_TOKEN,
-- robot_id: ROBOT_ID,
-- referrer: `https://robotstreamer.com/robot/${ROBOT_ID}`,
-- });
--
-- if (!pageData?.rtc_sfu?.host || !pageData?.rtc_sfu?.port) {
-- throw new Error('page_load missing rtc_sfu');
-- }
--
-- const peerId = `p:${crypto.randomBytes(3).toString('hex')}`;
-- const sfuUrl = `wss://${pageData.rtc_sfu.host}:${pageData.rtc_sfu.port}/?roomId=${encodeURIComponent(ROBOT_ID)}&peerId=${encodeURIComponent(peerId)}`;
-- log('Connecting SFU:', sfuUrl);
--
-- const peer = new ProtooPeer(sfuUrl);
-- await peer.connect();
--
-- const routerRtpCapabilities = await peer.request('getRouterRtpCapabilities');
-- const device = createDevice();
-- await device.load({ routerRtpCapabilities });
-- log('mediasoup device loaded');
--
-- const transportInfo = await peer.request('createWebRtcTransport', {
-- producing: true,
-- consuming: false,
-- streamkey: RS_TOKEN,
-- });
--
-- const sendTransport = device.createSendTransport(transportInfo);
--
-- sendTransport.on('connect', ({ dtlsParameters }, callback, errback) => {
-- peer.request('connectWebRtcTransport', {
-- transportId: sendTransport.id,
-- dtlsParameters,
-- }).then(callback).catch(errback);
-- });
--
-- sendTransport.on('produce', async ({ kind, rtpParameters, appData }, callback, errback) => {
-- try {
-- const data = await peer.request('produce', {
-- transportId: sendTransport.id,
-- kind,
-- rtpParameters,
-- appData,
-- });
-- callback({ id: data.id });
-- } catch (err) {
-- errback(err);
-- }
-- });
--
-- await peer.request('join', {
-- device,
-- rtpCapabilities: device.rtpCapabilities,
-- token: RS_TOKEN,
-- });
-- log('Joined RobotStreamer room');
--
-- const videoSource = new wrtc.nonstandard.RTCVideoSource();
-- const audioSource = new wrtc.nonstandard.RTCAudioSource();
-- const videoTrack = videoSource.createTrack();
-- const audioTrack = hasAudio ? audioSource.createTrack() : null;
-- videoTrack.contentHint = 'motion';
--
-- // Start capture BEFORE producing so RS never sees a dead first frame.
-- const ff = spawnFfmpeg(videoSource, audioSource);
-- await waitForVideoFrames(5, 15000);
-- if (status.videoFramesIn === 0) {
-- throw new Error('No video frames from source — ingest not flowing');
-- }
--
-- let videoSender = null;
-- const videoProducer = await sendTransport.produce({
-- track: videoTrack,
-- stopTracks: false,
-- disableTrackOnPause: false,
-- zeroRtpOnPause: false,
-- encodings: [{
-- active: true,
-- maxBitrate: videoKbps * 1000,
-- minBitrate: minVideoKbps * 1000,
-- maxFramerate: fps,
-- scaleResolutionDownBy: 1,
-- priority: 'high',
-- networkPriority: 'high',
-- }],
-- codecOptions: {
-- videoGoogleStartBitrate: videoKbps,
-- videoGoogleMinBitrate: minVideoKbps,
-- videoGoogleMaxBitrate: videoKbps,
-- },
-- onRtpSender: sender => { videoSender = sender; },
-- appData: { source: 'openvibelive-native-relay' },
-- });
-- await applySenderParams(videoSender, 'video');
-- log('Video producer created:', videoProducer.id);
--
-- if (hasAudio) {
-- const audioProducer = await sendTransport.produce({
-- track: audioTrack,
-- codecOptions: {
-- opusStereo: true,
-- opusDtx: false,
-- opusFec: false,
-- opusMaxPlaybackRate: 48000,
-- },
-- appData: { source: 'openvibelive-native-relay', lowLatency: true },
-- });
-- log('Audio producer created:', audioProducer.id);
-- }
--
-- status.connected = true;
-- log('LIVE: publishing to RobotStreamer');
--
-- setInterval(() => {
-- const line = { ...status };
-- status.videoFramesIn = 0;
-- status.videoFramesPushed = 0;
-- log('[stats]', JSON.stringify(line));
-- }, 10000).unref?.();
--
-- const shutdown = () => {
-- shuttingDown = true;
-- if (videoPumpTimer) { clearInterval(videoPumpTimer); videoPumpTimer = null; }
-- try { ff.kill('SIGTERM'); } catch {}
-- try { videoTrack.stop(); } catch {}
-- try { audioTrack?.stop(); } catch {}
-- try { sendTransport.close(); } catch {}
-- process.exit(0);
-- };
--
-- process.on('SIGINT', shutdown);
-- process.on('SIGTERM', shutdown);
--}
--
--main().catch(err => {
-- console.error('[fatal]', err && (err.stack || err.message || String(err)));
-- process.exit(1);
--});
ddiff --git a/server/integrations/rs-native-publisher.js b/server/integrations/rs-native-publisher.js
ddeleted file mode 100644
iindex 0eccd012..00000000
---- a/server/integrations/rs-native-publisher.js
++++ /dev/null
@@@ -1,273 +0,0 @@
--/**
-- * OpenVibe.Live — RobotStreamer native publisher (manager)
-- *
-- * Server-side RS video restreaming for streams that have no browser publisher:
-- * WHIP/OBS ingest (protocol webrtc, streaming_method whip) and RTMP ingest.
-- * Extracts media from the local ingest and spawns rs-native-publisher-worker.js
-- * (node + wrtc + mediasoup-client) to publish it to the RobotStreamer SFU.
-- *
-- * Media sources:
-- * - webrtc (WHIP): PlainRtpTransport consumers on the local mediasoup SFU
-- * → SDP file (same mechanism as the WebRTC→RTMP restream path)
-- * - rtmp: node-media-server HTTP-FLV loopback url
-- *
-- * The worker is restarted with backoff while the stream stays live.
-- */
--'use strict';
--
--const fs = require('fs');
--const os = require('os');
--const path = require('path');
--const { spawn } = require('child_process');
--
--const WORKER_PATH = path.join(__dirname, 'rs-native-publisher-worker.js');
--const MAX_ATTEMPTS = 10;
--const BASE_RESTART_DELAY_MS = 5000;
--const MAX_RESTART_DELAY_MS = 30000;
--
--class RsNativePublisher {
-- constructor() {
-- /** @type {Map<number, object>} streamId → session */
-- this.sessions = new Map();
-- }
--
-- isActive(streamId) {
-- return this.sessions.has(streamId);
-- }
--
-- _allocatePort() {
-- // Distinct range from restream-manager (20000-30000) to avoid clashes.
-- if (!this._nextPort) this._nextPort = 32000;
-- const port = this._nextPort;
-- this._nextPort += 2;
-- if (this._nextPort > 34000) this._nextPort = 32000;
-- return port;
-- }
--
-- /**
-- * Start (or keep) a native RS publish session for a live stream.
-- * @param {object} stream - streams row (must be live)
-- * @param {object} integration - robotstreamer_integrations row (token+robot resolved)
-- */
-- start(stream, integration) {
-- if (!stream?.id || !integration?.token || !integration?.robot_id) return;
--
-- const existing = this.sessions.get(stream.id);
-- if (existing && !existing.stopped) return;
--
-- const session = {
-- streamId: stream.id,
-- userId: stream.user_id,
-- token: integration.token,
-- robotId: String(integration.robot_id),
-- protocol: stream.protocol,
-- stopped: false,
-- attempts: 0,
-- child: null,
-- restartTimer: null,
-- webrtcState: null,
-- };
-- this.sessions.set(stream.id, session);
-- console.log(`[RS Native] Starting publisher for stream ${stream.id} (robot ${session.robotId}, protocol ${stream.protocol})`);
--
-- this._launch(session).catch(err => {
-- console.warn(`[RS Native] Launch failed for stream ${session.streamId}:`, err.message);
-- this._scheduleRestart(session);
-- });
-- }
--
-- stop(streamId) {
-- const session = this.sessions.get(streamId);
-- if (!session) return;
-- session.stopped = true;
-- if (session.restartTimer) { clearTimeout(session.restartTimer); session.restartTimer = null; }
-- if (session.child) { try { session.child.kill('SIGTERM'); } catch {} session.child = null; }
-- this._cleanupMedia(session);
-- this.sessions.delete(streamId);
-- console.log(`[RS Native] Stopped publisher for stream ${streamId}`);
-- }
--
-- stopAll() {
-- for (const streamId of [...this.sessions.keys()]) this.stop(streamId);
-- }
--
-- async _launch(session) {
-- const db = require('../db/database');
-- const stream = db.getStreamById(session.streamId);
-- if (!stream?.is_live) {
-- this.stop(session.streamId);
-- return;
-- }
--
-- let inputMode, inputPath, hasAudio;
--
-- if (session.protocol === 'webrtc') {
-- ({ inputPath, hasAudio } = await this._prepareSfuInput(session));
-- inputMode = 'sdp';
-- } else if (session.protocol === 'rtmp') {
-- const config = require('../config');
-- const user = db.getUserById(session.userId);
-- const streamKey = stream.managed_stream_key || user?.stream_key;
-- if (!streamKey) throw new Error('No stream key for FLV loopback');
-- const flvPort = (config.rtmp?.port || 1935) + 8000;
-- inputMode = 'url';
-- inputPath = `http://127.0.0.1:${flvPort}/live/${streamKey}.flv`;
-- hasAudio = true;
-- } else {
-- throw new Error(`Unsupported protocol for native RS publish: ${session.protocol}`);
-- }
--
-- if (session.stopped) { this._cleanupMedia(session); return; }
--
-- const child = spawn(process.execPath, [WORKER_PATH], {
-- env: {
-- ...process.env,
-- RS_TOKEN: session.token,
-- ROBOT_ID: session.robotId,
-- INPUT_MODE: inputMode,
-- INPUT_PATH: inputPath,
-- HAS_AUDIO: hasAudio ? '1' : '0',
-- },
-- stdio: ['ignore', 'pipe', 'pipe'],
-- });
-- session.child = child;
--
-- const prefix = `[RS Native ${session.streamId}]`;
-- const logChunk = (data) => {
-- for (const line of data.toString().split('\n')) {
-- if (line.trim()) console.log(prefix, line.trim());
-- }
-- };
-- child.stdout.on('data', logChunk);
-- child.stderr.on('data', logChunk);
--
-- child.on('exit', (code, signal) => {
-- if (session.child === child) session.child = null;
-- this._cleanupMedia(session);
-- if (session.stopped) return;
-- console.warn(`${prefix} worker exited (code ${code}, signal ${signal || 'none'})`);
-- this._scheduleRestart(session);
-- });
-- }
--
-- _scheduleRestart(session) {
-- if (session.stopped || session.restartTimer) return;
-- session.attempts++;
-- if (session.attempts > MAX_ATTEMPTS) {
-- console.warn(`[RS Native] Giving up on stream ${session.streamId} after ${MAX_ATTEMPTS} attempts`);
-- this.stop(session.streamId);
-- return;
-- }
-- const delay = Math.min(BASE_RESTART_DELAY_MS * session.attempts, MAX_RESTART_DELAY_MS);
-- console.log(`[RS Native] Restarting publisher for stream ${session.streamId} in ${delay}ms (attempt ${session.attempts})`);
-- session.restartTimer = setTimeout(() => {
-- session.restartTimer = null;
-- const db = require('../db/database');
-- const stream = db.getStreamById(session.streamId);
-- if (!stream?.is_live) {
-- this.stop(session.streamId);
-- return;
-- }
-- this._launch(session).catch(err => {
-- console.warn(`[RS Native] Relaunch failed for stream ${session.streamId}:`, err.message);
-- this._scheduleRestart(session);
-- });
-- }, delay);
-- }
--
-- /**
-- * Create PlainRtpTransport consumers on the local SFU and write an SDP
-- * file describing them (mirrors restream-manager's WebRTC input path).
-- */
-- async _prepareSfuInput(session) {
-- const webrtcSFU = require('../streaming/webrtc-sfu');
-- if (!webrtcSFU.ready) throw new Error('Mediasoup SFU not available');
--
-- const roomId = `stream-${session.streamId}`;
-- const videoProducer = await webrtcSFU.waitForProducer(roomId, 'video', 30000);
-- const audioProducer = webrtcSFU.findProducerByKind(roomId, 'audio');
--
-- const videoRtpPort = this._allocatePort();
-- const videoRtcpPort = videoRtpPort + 1;
-- let audioConsumer = null;
-- let audioRtpPort = null;
--
-- const videoConsumer = await webrtcSFU.createPlainConsumer(
-- roomId, videoProducer.id, '127.0.0.1', videoRtpPort, videoRtcpPort
-- );
-- try {
-- if (audioProducer) {
-- audioRtpPort = this._allocatePort();
-- audioConsumer = await webrtcSFU.createPlainConsumer(
-- roomId, audioProducer.id, '127.0.0.1', audioRtpPort, audioRtpPort + 1
-- );
-- }
-- } catch (err) {
-- webrtcSFU.closePlainConsumer(roomId, videoConsumer.transportId);
-- throw err;
-- }
--
-- const sdpPath = path.join(os.tmpdir(), `openvibe-rs-native-${session.streamId}.sdp`);
-- fs.writeFileSync(sdpPath, this._buildSdp(videoConsumer, audioConsumer, videoRtpPort, audioRtpPort), 'utf8');
--
-- session.webrtcState = {
-- roomId,
-- sdpPath,
-- videoTransportId: videoConsumer.transportId,
-- audioTransportId: audioConsumer?.transportId || null,
-- };
--
-- console.log(`[RS Native] SFU consumers ready for stream ${session.streamId} (video port ${videoRtpPort}${audioConsumer ? `, audio port ${audioRtpPort}` : ', no audio'})`);
-- return { inputPath: sdpPath, hasAudio: !!audioConsumer };
-- }
--
-- _cleanupMedia(session) {
-- if (!session.webrtcState) return;
-- const webrtcSFU = require('../streaming/webrtc-sfu');
-- const { roomId, videoTransportId, audioTransportId, sdpPath } = session.webrtcState;
-- try { if (videoTransportId) webrtcSFU.closePlainConsumer(roomId, videoTransportId); } catch {}
-- try { if (audioTransportId) webrtcSFU.closePlainConsumer(roomId, audioTransportId); } catch {}
-- try { fs.unlinkSync(sdpPath); } catch {}
-- session.webrtcState = null;
-- }
--
-- _buildSdp(videoConsumer, audioConsumer, videoPort, audioPort) {
-- const lines = [
-- 'v=0',
-- 'o=- 0 0 IN IP4 127.0.0.1',
-- 's=OpenVibe.Live RS Native Relay',
-- 'c=IN IP4 127.0.0.1',
-- 't=0 0',
-- ];
--
-- const vPT = videoConsumer.payloadType;
-- const vCodecName = (videoConsumer.mimeType || 'video/VP8').split('/')[1];
-- lines.push(`m=video ${videoPort} RTP/AVP ${vPT}`);
-- lines.push(`a=rtpmap:${vPT} ${vCodecName}/${videoConsumer.clockRate}`);
-- if (videoConsumer.ssrc) lines.push(`a=ssrc:${videoConsumer.ssrc} cname:rs-native-video`);
-- if (videoConsumer.codecParameters) {
-- const fmtp = Object.entries(videoConsumer.codecParameters).map(([k, v]) => `${k}=${v}`).join(';');
-- if (fmtp) lines.push(`a=fmtp:${vPT} ${fmtp}`);
-- }
-- lines.push('a=recvonly');
--
-- if (audioConsumer && audioPort) {
-- const aPT = audioConsumer.payloadType;
-- const aCodecName = (audioConsumer.mimeType || 'audio/opus').split('/')[1];
-- const channels = audioConsumer.channels || 2;
-- lines.push(`m=audio ${audioPort} RTP/AVP ${aPT}`);
-- lines.push(`a=rtpmap:${aPT} ${aCodecName}/${audioConsumer.clockRate}/${channels}`);
-- if (audioConsumer.ssrc) lines.push(`a=ssrc:${audioConsumer.ssrc} cname:rs-native-audio`);
-- if (audioConsumer.codecParameters) {
-- const fmtp = Object.entries(audioConsumer.codecParameters).map(([k, v]) => `${k}=${v}`).join(';');
-- if (fmtp) lines.push(`a=fmtp:${aPT} ${fmtp}`);
-- }
-- lines.push('a=recvonly');
-- }
--
-- lines.push('');
-- return lines.join('\r\n');
-- }
--}
--
--module.exports = new RsNativePublisher();
ddiff --git a/server/integrations/rs-passthrough-relay.js b/server/integrations/rs-passthrough-relay.js
iindex 35c91eb4..df4eb70e 100644
---- a/server/integrations/rs-passthrough-relay.js
++++ b/server/integrations/rs-passthrough-relay.js
@@@ -1,10 +1,9 @@
/**
* OpenVibe.Live — RobotStreamer RAW passthrough relay (zero re-encode).
*
-- * The old native publisher decodes the stream and RE-ENCODES it to RobotStreamer
-- * (ffmpeg + libwebrtc), and browser broadcasters publish to RS from the browser as a
-- * SECOND encode. Both waste CPU and soften the picture. This relay instead forwards
-- * goosely's ALREADY-ENCODED RTP straight through:
++ * Re-encoding the stream for RobotStreamer (the removed ffmpeg "native publisher") or
++ * publishing it a SECOND time from the broadcaster's browser both waste CPU and soften
++ * the picture. This relay instead forwards the source's ALREADY-ENCODED RTP straight through:
*
* our mediasoup SFU (goosely's producer, NACK-protected)
* → DirectTransport consumer (in-process, lossless, encoded RTP)
@@@ -17,8 +16,16 @@
*
* The werift↔mediasoup bridge (hand-rolled SDP answer from RS's transport params) is
* validated end-to-end in scratch/werift-ms-harness.js. RS's SFU is mediasoup, so the
-- * same bridge applies. Gated behind config.robotstreamer.passthrough (default off) with
-- * the transcode publisher kept as fallback.
++ * same bridge applies. This is the only RobotStreamer video path; RS_PASSTHROUGH=0 is an
++ * emergency kill-switch (chat mirroring keeps working without video).
++ *
++ * Sources: anything that produces into our mediasoup SFU — browser broadcasts and OBS/WHIP.
++ * RTMP and JSMPEG ingests have no SFU producers and are not forwarded (robotstreamer-service
++ * skips the relay for them rather than letting it wait for a producer that never comes).
++ *
++ * Lifecycle: the session ends with its source (producer-removed, the stream row going
++ * not-live, or an explicit stop); restarts back off and give up; status(streamId) is what
++ * the dashboard polls.
*/
const WebSocket = require('ws');
const https = require('node:https');
@@@ -31,6 +38,16 @@ try { werift = require('werift'); } catch { /* optional dep — relay disabled i
const RS_API_HOST = 'api.robotstreamer.com';
const UA = 'Mozilla/5.0 (X11; Linux x86_64) OpenVibe.Live-RelayPassthrough';
++// Restart policy. Every restart is visible on RobotStreamer as the video cutting out (see
++// _scheduleRestart), so restarts back off instead of hammering at a flat 4s, and a session that
++// keeps dying gives up rather than looping forever — the old loop never stopped, not even after
++// the OpenVibe stream had ended, which is how a robot stayed "live" (black) on robotstreamer.com.
++const RESTART_BASE_MS = 3000;
++const RESTART_MAX_MS = 30000;
++const MAX_RESTARTS = 12; // consecutive restarts before giving up (a stable run resets it)
++const STABLE_MS = 60000; // a run that stayed live this long resets the restart counter
++const SOURCE_CHECK_MS = 5000; // how often a running session re-checks that its source is live
++
function log(streamId, ...a) { console.log(`[RS Passthrough ${streamId}]`, ...a); }
// ── RS HTTP API (robot_page_load → rtc_sfu host/port) ────────────────────
@@@ -293,22 +310,47 @@ class RsPassthroughRelay {
async start(stream, integration) {
if (!werift) { log(stream.id, 'werift not installed — cannot start passthrough'); return false; }
-- if (this.sessions.has(stream.id)) return true;
-- const session = { streamId: stream.id, robotId: integration.robot_id, token: integration.token, stopped: false, peer: null, pc: null, ingests: [], roomId: `stream-${stream.id}`, restartTimer: null };
++ const existing = this.sessions.get(stream.id);
++ if (existing) {
++ // A session that gave up stays in the map so its failure is visible; an explicit
++ // start (user pressed Retry, integration re-saved) relaunches it from a clean slate.
++ if (existing.state !== 'failed') return true;
++ existing.stopped = false; existing.restarts = 0; existing.lastError = null;
++ this._launch(existing);
++ return true;
++ }
++ const session = {
++ streamId: stream.id, robotId: integration.robot_id, token: integration.token,
++ stopped: false, peer: null, pc: null, ingests: [], roomId: `stream-${stream.id}`,
++ restartTimer: null, sourceCheck: null,
++ state: 'starting', startedAt: 0, liveAt: 0, firstStartedAt: Date.now(),
++ restarts: 0, lastError: null, lastRestartReason: null, nextRestartAt: 0,
++ stats: null, rs: null,
++ };
this.sessions.set(stream.id, session);
++ this._launch(session);
++ return true;
++ }
++
++ /** Run (or re-run) the passthrough for an existing session record. */
++ _launch(session) {
++ session.state = 'starting'; session.startedAt = Date.now(); session.liveAt = 0; session.nextRestartAt = 0;
++ session.stats = null; session.rs = null;
this._run(session).catch(err => {
-- log(stream.id, 'run error:', err.message);
++ // A stop that lands while _run is awaiting (RS handshake, ICE) must not leave the
++ // half-built peer behind: that peer would keep the robot live on RS after we forgot it.
++ if (session.stopped) { this._teardown(session); return; }
++ log(session.streamId, 'run error:', err.message);
this._teardown(session);
-- this._scheduleRestart(session, stream, integration, `_run threw: ${err.message}`);
++ this._scheduleRestart(session, `${err.message}`);
});
-- return true;
}
stop(streamId) {
const s = this.sessions.get(streamId);
if (!s) return;
-- s.stopped = true;
-- if (s.restartTimer) clearTimeout(s.restartTimer);
++ s.stopped = true; s.state = 'stopped';
++ if (s.restartTimer) { clearTimeout(s.restartTimer); s.restartTimer = null; }
this._teardown(s);
this.sessions.delete(streamId);
log(streamId, 'stopped');
@@@ -326,7 +368,7 @@ class RsPassthroughRelay {
for (const id of ids) {
const s = this.sessions.get(id);
if (!s) continue;
-- s.stopped = true;
++ s.stopped = true; s.state = 'stopped';
if (s.restartTimer) { clearTimeout(s.restartTimer); s.restartTimer = null; }
try { this._teardown(s); } catch { /* */ }
this.sessions.delete(id);
@@@ -335,6 +377,46 @@ class RsPassthroughRelay {
return ids.length;
}
++ /**
++ * Is the OpenVibe stream this session forwards still live? The relay must never outlive its
++ * source: RobotStreamer keeps a robot "live" for as long as our producers exist, so a relay
++ * that keeps its peer open (or keeps restarting) after the streamer stopped shows RS viewers
++ * a live robot with a black picture. Injectable for tests.
++ */
++ _sourceLive(streamId) {
++ if (this._sourceLiveOverride) return this._sourceLiveOverride(streamId);
++ try { return !!require('../db/database').getStreamById(streamId)?.is_live; } catch { return true; }
++ }
++
++ /**
++ * What the broadcaster's dashboard shows for this stream, or null when no session exists.
++ * Never includes the RS token.
++ */
++ status(streamId) {
++ const s = this.sessions.get(streamId);
++ if (!s) return null;
++ const now = Date.now();
++ const st = s.stats || {};
++ return {
++ active: !s.stopped && s.state !== 'failed',
++ state: s.state, // starting | live | restarting | failed
++ robot_id: s.robotId,
++ started_at: s.firstStartedAt,
++ live_at: s.liveAt || null,
++ uptime_ms: s.liveAt && s.state === 'live' ? now - s.liveAt : 0,
++ restarts: s.restarts,
++ last_error: s.lastError,
++ last_restart_reason: s.lastRestartReason,
++ next_restart_at: s.nextRestartAt || null,
++ has_audio: !!(st.hasAudio),
++ video_codec: st.codec || null,
++ packets_out: (st.vOut || 0) + (st.aOut || 0),
++ keyframe_requests: st.pli || 0,
++ ingest_stalls: st.ingestStalls || 0,
++ rs: s.rs, // { loss_pct, rtt_ms, nacks, plis } from RS's RTCP
++ };
++ }
++
/**
* Every restart re-produces to RS with a FRESH SSRC, which RobotStreamer surfaces to viewers
* as a new consumer on a new mid — and RS's viewer client reacts to the matching
@@@ -342,25 +424,48 @@ class RsPassthroughRelay {
* restart is directly visible as the stream cutting out. `reason` is therefore not a nicety:
* it is the difference between "RS dropped our websocket", "ICE/DTLS failed" and "_run threw"
* (e.g. mediasoup out of RTC ports), which are three unrelated bugs with one symptom.
++ *
++ * Bails out (stops the session) when the source stream is no longer live, backs off
++ * exponentially, and gives up after MAX_RESTARTS consecutive attempts.
*/
-- _scheduleRestart(session, stream, integration, reason = 'unspecified') {
++ _scheduleRestart(session, reason = 'unspecified') {
if (session.stopped || session.restartTimer) return;
-- this.restartCount = (this.restartCount || 0) + 1;
-- log(session.streamId, `⚠️ passthrough restart #${this.restartCount} scheduled — reason: ${reason} ` +
++ session.lastRestartReason = reason;
++ session.lastError = reason;
++ if (!this._sourceLive(session.streamId)) {
++ log(session.streamId, `source stream is no longer live — stopping passthrough instead of restarting (${reason})`);
++ this.stop(session.streamId);
++ return;
++ }
++ // A run that stayed live for a while proves the path works; only count consecutive failures.
++ if (session.liveAt && Date.now() - session.liveAt > STABLE_MS) session.restarts = 0;
++ session.restarts++;
++ if (session.restarts > MAX_RESTARTS) {
++ session.state = 'failed';
++ session.lastError = `gave up after ${MAX_RESTARTS} restarts — last: ${reason}`;
++ log(session.streamId, `❌ ${session.lastError}`);
++ return;
++ }
++ const delay = Math.min(RESTART_BASE_MS * 2 ** (session.restarts - 1), RESTART_MAX_MS);
++ session.state = 'restarting';
++ session.nextRestartAt = Date.now() + delay;
++ log(session.streamId, `⚠️ passthrough restart #${session.restarts}/${MAX_RESTARTS} in ${delay}ms — reason: ${reason} ` +
`(RS viewers see this as the video cutting out: new SSRC → new consumer → client blanks the MediaStream)`);
session.restartTimer = setTimeout(() => {
session.restartTimer = null;
if (session.stopped) return;
-- this.sessions.delete(session.streamId);
++ if (!this._sourceLive(session.streamId)) { this.stop(session.streamId); return; }
log(session.streamId, 'restarting passthrough…');
-- this.start(stream, integration);
-- }, 4000);
++ this._launch(session);
++ }, delay);
}
_teardown(session) {
if (session.statsTimer) { clearInterval(session.statsTimer); session.statsTimer = null; }
if (session.ingestWatchdog) { clearInterval(session.ingestWatchdog); session.ingestWatchdog = null; }
if (session._audioWatch) { try { session._audioWatch(); } catch { /* */ } session._audioWatch = null; }
++ if (session._producerWatch) { try { session._producerWatch(); } catch { /* */ } session._producerWatch = null; }
++ if (session.sourceCheck) { clearInterval(session.sourceCheck); session.sourceCheck = null; }
const sfu = require('../streaming/webrtc-sfu');
for (const ing of session.ingests) {
try { ing.socket.removeAllListeners('message'); ing.socket.close(); } catch {}
@@@ -387,6 +492,7 @@ class RsPassthroughRelay {
// grace period so a mic that is a moment behind the video producer still makes
// it into the session, and watch for a later one so we can rebuild.
const videoProd = await sfu.waitForProducer(session.roomId, 'video', 30000);
++ if (session.stopped) throw new Error('stopped during setup');
let audioProd = sfu.findProducerByKind(session.roomId, 'audio');
if (!audioProd) {
try { audioProd = await sfu.waitForProducer(session.roomId, 'audio', 4000); }
@@@ -400,32 +506,63 @@ class RsPassthroughRelay {
if (session.stopped || kind !== 'audio' || roomId !== session.roomId) return;
sfu.removeListener?.('producer-added', onAudio);
log(sid, 'audio producer appeared after start — restarting passthrough to include it');
-- const st = { id: sid }; const integ = { robot_id: session.robotId, token: session.token };
this._teardown(session);
-- this._scheduleRestart(session, st, integ, 'audio producer added after a video-only start');
++ this._scheduleRestart(session, 'audio producer added after a video-only start');
};
sfu.on?.('producer-added', onAudio);
session._audioWatch = () => sfu.removeListener?.('producer-added', onAudio);
}
++ // The source going away is the ONE event this relay must react to immediately. When the
++ // broadcaster's SFU transport closes (tab closed, stream ended, room closed) mediasoup
++ // removes the producer; our plain consumer then feeds nothing, but the werift peer to RS
++ // stays connected and RS keeps the robot "live" with a black picture until a human
++ // notices. Tear down at once: if the OpenVibe stream is over, stop for good; if it is a
++ // reconnect (stream still live), come back through the restart path and wait for the new
++ // producer.
++ const onProducerRemoved = ({ roomId, kind }) => {
++ if (session.stopped || roomId !== session.roomId || kind !== 'video') return;
++ sfu.removeListener?.('producer-removed', onProducerRemoved);
++ session._producerWatch = null;
++ log(sid, 'source video producer removed');
++ this._teardown(session);
++ this._scheduleRestart(session, 'source video producer removed');
++ };
++ sfu.on?.('producer-removed', onProducerRemoved);
++ session._producerWatch = () => sfu.removeListener?.('producer-removed', onProducerRemoved);
++
++ // Backstop for end paths that never touch the SFU room (a stale-heartbeat sweep, an
++ // admin force-end, db.endStream from an ingest handler): stop as soon as the stream row
++ // says the source is over.
++ session.sourceCheck = setInterval(() => {
++ if (session.stopped) return;
++ if (!this._sourceLive(sid)) {
++ log(sid, 'source stream ended — stopping passthrough');
++ this.stop(sid);
++ }
++ }, SOURCE_CHECK_MS);
++ session.sourceCheck.unref?.();
++
// 2) Encoded-RTP ingest (PlainTransport → localhost UDP socket).
const videoIn = await openPlainIngest(sfu, session.roomId, videoProd.id);
session.ingests.push(videoIn);
let audioIn = null;
if (audioProd) { audioIn = await openPlainIngest(sfu, session.roomId, audioProd.id); session.ingests.push(audioIn); }
++ if (session.stopped) throw new Error('stopped during setup');
// 3) Connect to RS: discover SFU, open protoo.
const page = await postJson(RS_API_HOST, '/v1/robot_page_load', { token: session.token, robot_id: session.robotId, referrer: `https://robotstreamer.com/robot/${session.robotId}` });
if (!page?.rtc_sfu?.host || !page?.rtc_sfu?.port) throw new Error('robot_page_load missing rtc_sfu');
++ if (session.stopped) throw new Error('stopped during setup');
const peerId = `p:${crypto.randomBytes(3).toString('hex')}`;
const wsUrl = `wss://${page.rtc_sfu.host}:${page.rtc_sfu.port}/?roomId=${encodeURIComponent(session.robotId)}&peerId=${encodeURIComponent(peerId)}`;
const peer = new ProtooPeer(wsUrl, session.robotId, (code) => {
if (session.stopped) return;
-- const st = { id: sid }; const integ = { robot_id: session.robotId, token: session.token };
-- this._teardown(session); this._scheduleRestart(session, st, integ, `RS protoo websocket closed (code ${code})`);
++ this._teardown(session); this._scheduleRestart(session, `RS protoo websocket closed (code ${code})`);
});
session.peer = peer;
await peer.connect();
++ if (session.stopped) throw new Error('stopped during setup');
log(sid, 'RS protoo connected');
const routerRtpCapabilities = await peer.request('getRouterRtpCapabilities');
@@@ -438,6 +575,7 @@ class RsPassthroughRelay {
// 4) Create RS send transport.
const transportInfo = await peer.request('createWebRtcTransport', { producing: true, consuming: false, streamkey: session.token });
++ if (session.stopped) throw new Error('stopped during setup');
// 5) Build werift peer: sendonly transceivers whose codec MATCHES the source (so RS
// decodes the forwarded payload) and that carry the header extensions RS needs
@@@ -465,8 +603,7 @@ class RsPassthroughRelay {
pc.connectionStateChange.subscribe(() => {
log(sid, 'werift conn', pc.connectionState);
if ((pc.connectionState === 'failed' || pc.connectionState === 'disconnected') && !session.stopped) {
-- const st = { id: sid }; const integ = { robot_id: session.robotId, token: session.token };
-- this._teardown(session); this._scheduleRestart(session, st, integ, `werift connectionState=${pc.connectionState} (ICE/DTLS to RS lost)`);
++ this._teardown(session); this._scheduleRestart(session, `werift connectionState=${pc.connectionState} (ICE/DTLS to RS lost)`);
}
});
@@@ -484,12 +621,14 @@ class RsPassthroughRelay {
// 6) Answer from RS transport params → werift connects ICE+DTLS as client.
await pc.setRemoteDescription({ type: 'answer', sdp: buildAnswer(transportInfo, parsed) });
++ if (session.stopped) throw new Error('stopped during setup');
// 7) Hand RS our DTLS fingerprint (role client) + join.
const [fpAlg, fpVal] = parsed.fingerprint.split(' ');
await peer.request('connectWebRtcTransport', { transportId: transportInfo.id, dtlsParameters: { role: 'client', fingerprints: [{ algorithm: fpAlg, value: fpVal }] } });
await peer.request('join', { displayName: 'OpenVibe.Live', device: { flag: 'openvibe-relay', name: 'werift', version: '1' }, rtpCapabilities: routerRtpCapabilities, token: session.token });
log(sid, 'RS transport connected + joined');
++ if (session.stopped) throw new Error('stopped during setup');
// 8) Produce (video, then audio) with rtpParameters matching what werift sends.
const vProd = await peer.request('produce', { transportId: transportInfo.id, kind: 'video', rtpParameters: buildProduceParams(vMedia), appData: { source: 'openvibe-passthrough' } });
@@@ -513,7 +652,9 @@ class RsPassthroughRelay {
pli: 0, lostIn: 0, kf: 0, maxKfGap: 0, inj: 0,
rtxDeep: 0, rtxMiss: 0, rtxTooOld: 0, gopDamaged: 0, lateFill: 0,
ingestStalls: 0, ingestStallMs: 0, vAlien: 0, aAlien: 0,
++ hasAudio: !!audioIn, codec: videoCodec.mimeType,
};
++ session.stats = stats;
const vPt = +vMedia.pts[0];
// ── Deep retransmit cache ────────────────────────────────────────────────────────
@@@ -833,6 +974,12 @@ class RsPassthroughRelay {
const remb = typeof s.receiverEstimatedMaxBitrate === 'number' ? kbps(s.receiverEstimatedMaxBitrate / 1000) + 'k' : '?';
const bwe = typeof s.senderBWE?.availableBitrate === 'number' ? kbps(s.senderBWE.availableBitrate / 1000) + 'k' : '?';
const stalled = _videoStalledSince ? ` STALLED-${Date.now() - _videoStalledSince}ms` : '';
++ session.rs = {
++ loss_pct: typeof s.remoteFractionLost === 'number' ? +((s.remoteFractionLost / 256) * 100).toFixed(1) : null,
++ rtt_ms: typeof s.rtt === 'number' ? Math.round(s.rtt * 1000) : null,
++ nacks: s.nackCount ?? null, plis: s.pliCount ?? null,
++ stalled_ms: _videoStalledSince ? Date.now() - _videoStalledSince : 0,
++ };
log(sid, `flow: vIn ${Math.round((stats.vIn - lastVIn) / 10)}/s vOut ${Math.round((stats.vOut - lastVOut) / 10)}/s ` +
`aIn ${Math.round((stats.aIn - lastAIn) / 10)}/s aOut ${Math.round((stats.aOut - lastAOut) / 10)}/s${stalled} ` +
`| writeErr v=${stats.vErr} a=${stats.aErr} alienSsrc v=${stats.vAlien} a=${stats.aAlien} | stalls=${stats.ingestStalls} (${Math.round(stats.ingestStallMs / 1000)}s total) ` +
@@@ -893,6 +1040,7 @@ class RsPassthroughRelay {
// link and caused a loss burst → the freeze seen right after every (re)connect.
setTimeout(reqKey, 600);
++ session.state = 'live'; session.liveAt = Date.now(); session.lastError = null;
log(sid, '✅ raw passthrough live (zero re-encode)');
}
}
ddiff --git a/server/streaming/broadcast-server.js b/server/streaming/broadcast-server.js
iindex 9d9861c7..73e07953 100644
---- a/server/streaming/broadcast-server.js
++++ b/server/streaming/broadcast-server.js
@@@ -579,6 +579,12 @@ class BroadcastServer extends EventEmitter {
require('./recorder').finalizeStream(client.streamId).catch((err) => {
console.warn(`[Broadcast] Failed to finalize VOD for stale stream ${client.streamId}:`, err.message);
});
++ // Everything that forwards this stream elsewhere must stop with it. This
++ // path (tab closed, laptop lid shut) used to end only the room, so the
++ // RobotStreamer passthrough and any ffmpeg restreams kept their
++ // outbound connections open and the robot / platform stayed "live"
++ // showing nothing.
++ this._stopForwarders(client.streamId);
webrtcSFU.closeRoom(`stream-${client.streamId}`);
} catch (err) {
console.error('[Broadcast] Failed to end stale stream:', err.message);
@@@ -1002,6 +1008,14 @@ class BroadcastServer extends EventEmitter {
console.log(`[Broadcast] Notified pending viewers of SFU producers for stream ${streamId}`);
}
++ /** Stop every outbound forwarder (RS passthrough, ffmpeg restreams, chat relays, AI bots) for a stream. */
++ _stopForwarders(streamId) {
++ try { require('../integrations/robotstreamer-service').stopForStream(streamId); } catch (err) { console.warn('[Broadcast] RS stop failed:', err.message); }
++ try { require('./restream-manager').stopAllForStream(streamId); } catch (err) { console.warn('[Broadcast] restream stop failed:', err.message); }
++ try { require('../integrations/chat-relay-service').stopForStream(streamId); } catch { /* non-critical */ }
++ try { require('../integrations/ai-chatbot-service').stopForStream(streamId); } catch { /* non-critical */ }
++ }
++
/**
* Cleanly end a stream: close broadcaster WS, notify viewers, clear room.
* Called from DELETE /streams/:id and stale heartbeat cleanup.
ddiff --git a/server/streaming/restream-manager.js b/server/streaming/restream-manager.js
iindex c62bcff3..8012ca15 100644
---- a/server/streaming/restream-manager.js
++++ b/server/streaming/restream-manager.js
@@@ -30,7 +30,12 @@ const MAX_RESTART_ATTEMPTS = 30; // 30 attempts (was 10 — too few fo
const FFMPEG_STARTUP_DELAY_RTMP = 3000;
const RAPID_CRASH_THRESHOLD_MS = 5000; // If FFmpeg exits within this, apply extra backoff
const RAPID_CRASH_GIVEUP = 4; // …this many rapid crashes in a row = destination is dead (circuit-break)
--const LIVE_OUTPUT_BUFFER_MS = 1000;
++// ffmpeg is spawned with `-progress pipe:1`, which prints a key=value block to stdout every
++// stats period once the output is open and frames are flowing. That is the live signal: the
++// first block with frame>0 (or out_time>0 for a codec copy) means the ingest accepted the
++// stream and is taking data. Without it we used to guess after 15s.
++const LIVE_ACK_TIMEOUT_MS = 20000; // no progress at all in this long = the ingest never took our data
++const SRT_DEFAULT_LATENCY_MS = 120; // SRT's own default; enough for a same-continent ingest
/**
* Path to a static FFmpeg binary built with OpenSSL (not GnuTLS).
@@@ -66,7 +71,7 @@ const QUALITY_PRESETS = {
label: 'Low (720p 1500k)',
videoBitrate: '1500k',
maxrate: '1800k',
-- bufsize: '3000k',
++ bufsize: '1500k', // 1x bitrate VBV: about a second less latency than the old 2x, still CBR-safe
audioBitrate: '96k',
preset: 'ultrafast', // ultrafast: ~50% less CPU than veryfast — critical for multi-dest servers
scale: '1280:720',
@@@ -77,7 +82,7 @@ const QUALITY_PRESETS = {
label: 'Medium (720p 2500k)',
videoBitrate: '2500k',
maxrate: '3000k',
-- bufsize: '5000k',
++ bufsize: '2500k', // 1x bitrate VBV: about a second less latency than the old 2x, still CBR-safe
audioBitrate: '128k',
preset: 'ultrafast', // ultrafast: stream stability > marginal quality gain from veryfast
scale: '1280:720',
@@@ -88,7 +93,7 @@ const QUALITY_PRESETS = {
label: 'High (720p 4000k)',
videoBitrate: '4000k',
maxrate: '4500k',
-- bufsize: '8000k',
++ bufsize: '4000k', // 1x bitrate VBV: about a second less latency than the old 2x, still CBR-safe
audioBitrate: '160k',
preset: 'superfast', // superfast: good balance for single-dest or beefier servers
scale: '1280:720',
@@@ -99,7 +104,7 @@ const QUALITY_PRESETS = {
label: 'Ultra (1080p 6000k)',
videoBitrate: '6000k',
maxrate: '6500k',
-- bufsize: '12000k',
++ bufsize: '6000k', // 1x bitrate VBV: about a second less latency than the old 2x, still CBR-safe
audioBitrate: '192k',
preset: 'veryfast',
scale: null, // No scaling — pass through native resolution
@@@ -372,9 +377,62 @@ class RestreamManager extends EventEmitter {
url += '/app';
}
++ if (/^srt:\/\//i.test(url)) return this._buildSrtUrl(url, dest);
return `${url}/${dest.stream_key}`;
}
++ /**
++ * SRT destination: the "stream key" is the SRT streamid, and latency/passphrase travel as
++ * URL options. The muxer is MPEG-TS, not FLV, and the caller mode with 1316-byte packets
++ * is what every SRT ingest (OBS-style, Haivision, MediaMTX, SRT Live Server) expects.
++ */
++ _buildSrtUrl(base, dest) {
++ const u = new URL(base);
++ const q = u.searchParams;
++ if (dest.stream_key && !q.has('streamid')) q.set('streamid', dest.stream_key);
++ const latency = Number(dest.srt_latency_ms) > 0 ? Number(dest.srt_latency_ms) : SRT_DEFAULT_LATENCY_MS;
++ q.set('latency', String(latency * 1000)); // libsrt takes microseconds
++ if (dest.srt_passphrase) q.set('passphrase', dest.srt_passphrase);
++ if (!q.has('mode')) q.set('mode', 'caller');
++ if (!q.has('pkt_size')) q.set('pkt_size', '1316');
++ return u.toString();
++ }
++
++ /** Is this destination URL an SRT one (MPEG-TS muxer) rather than RTMP/RTMPS (FLV)? */
++ static isSrtUrl(url) { return /^srt:\/\//i.test(String(url || '')); }
++
++ /**
++ * Muxer + real-time flags for the output. FLV for RTMP(S); MPEG-TS for SRT. Everything that
++ * delays a packet on the way out is off: no mux buffering, flush every packet.
++ */
++ _outputArgs(destUrl) {
++ const common = ['-muxdelay', '0', '-muxpreload', '0', '-flush_packets', '1', '-max_muxing_queue_size', '4096'];
++ if (RestreamManager.isSrtUrl(destUrl)) {
++ return [...common, '-f', 'mpegts', '-mpegts_flags', '+resend_headers', destUrl];
++ }
++ return [...common, '-rtmp_live', 'live', '-f', 'flv', '-flvflags', 'no_duration_filesize', destUrl];
++ }
++
++ /**
++ * Turn ffmpeg's last stderr lines into something a streamer can act on. The raw text is kept
++ * alongside for the log; this is what the dashboard shows.
++ */
++ static friendlyFfmpegError(raw, platform = 'the destination') {
++ const t = String(raw || '');
++ const name = platform === 'custom' ? 'the destination' : platform.charAt(0).toUpperCase() + platform.slice(1);
++ if (/NetStream\.Publish\.BadName|Authentication|Unauthorized|403|Access denied|invalid stream key|bad name|not authorized|Publish rejected/i.test(t)) return `${name} rejected the stream key — check the key (and that it is a live key, not an expired one)`;
++ if (/Name or service not known|Temporary failure in name resolution|getaddrinfo|could not resolve/i.test(t)) return `Could not resolve the ingest host name — check the server URL`;
++ if (/Connection refused|No route to host|Network is unreachable|Connection timed out|timed out|Failed to connect|Cannot open connection/i.test(t)) return `Could not connect to ${name}'s ingest server — it may be down or the URL/port is wrong`;
++ if (/TLS|SSL|gnutls|handshake|certificate/i.test(t)) return `TLS handshake with ${name} failed — try the plain rtmp:// ingest URL or the platform's rtmps:// address`;
++ if (/Broken pipe|Connection reset|End of file|EOF|I\/O error|Input\/output error|Server closed/i.test(t)) return `${name} closed the connection — usually a rejected key, a second stream on the same key, or a platform-side hiccup (auto-retrying)`;
++ if (/srt|SRT/.test(t) && /passphrase|encrypt|crypto|Wrong password/i.test(t)) return `SRT passphrase rejected — the passphrase (or its length, 10–79 characters) does not match the receiver`;
++ if (/SRT.*(Connection setup failure|rejected|timeout)/i.test(t)) return `SRT receiver did not accept the connection — check host, port, streamid and that the receiver is in listener mode`;
++ if (/Unrecognized option|Invalid argument|Option .* not found/i.test(t)) return `Encoder configuration was rejected by ffmpeg (${t.slice(0, 80)})`;
++ if (/No such file|not found/i.test(t) && /ffmpeg/i.test(t)) return 'ffmpeg is not installed on the server';
++ if (!t.trim()) return 'Restream process stopped without a message';
++ return t.slice(0, 160);
++ }
++
/**
* Start RTMP → RTMP restream (codec copy, zero CPU overhead).
*/
@@@ -389,14 +447,7 @@ class RestreamManager extends EventEmitter {
'-i', flvUrl,
'-c', 'copy', // Codec copy — no re-encoding
'-fflags', '+nobuffer+discardcorrupt',
-- '-muxdelay', '0',
-- '-muxpreload', '0',
-- '-flush_packets', '1',
-- '-rtmp_live', 'live',
-- '-rtmp_buffer', '2000', // 2s output buffer — absorbs input jitter
-- '-f', 'flv',
-- '-flvflags', 'no_duration_filesize',
-- destUrl,
++ ...this._outputArgs(destUrl),
];
this._spawnFFmpeg(session, args);
@@@ -418,15 +469,7 @@ class RestreamManager extends EventEmitter {
'-f', 'mpegts',
'-i', 'pipe:0', // Read combined MPEG-TS from stdin
...this._getEncodingArgs(preset, { customOverrides }),
-- '-muxdelay', '0',
-- '-muxpreload', '0',
-- '-flush_packets', '1',
-- '-rtmp_live', 'live',
-- '-rtmp_buffer', '2000',
-- '-max_muxing_queue_size', '4096',
-- '-f', 'flv',
-- '-flvflags', 'no_duration_filesize',
-- destUrl,
++ ...this._outputArgs(destUrl),
];
this._spawnFFmpeg(session, args);
@@@ -583,15 +626,7 @@ class RestreamManager extends EventEmitter {
'-avoid_negative_ts', 'make_zero', // Handle timestamp resets from packet gaps gracefully
'-i', sdpPath,
...this._getEncodingArgs(preset, { hasAudio: !!audioConsumer, customOverrides }),
-- '-muxdelay', '0',
-- '-muxpreload', '0',
-- '-flush_packets', '1',
-- '-rtmp_live', 'live',
-- '-rtmp_buffer', '2000', // 2s output buffer (was 1s) — absorbs input jitter
-- '-max_muxing_queue_size', '4096', // Prevent FLV muxer stalls from interleave gaps
-- '-f', 'flv',
-- '-flvflags', 'no_duration_filesize',
-- destUrl,
++ ...this._outputArgs(destUrl),
];
this._spawnFFmpeg(session, args);
@@@ -672,7 +707,7 @@ class RestreamManager extends EventEmitter {
if (customOverrides.videoBitrate) {
const kbps = parseInt(customOverrides.videoBitrate, 10);
maxrate = `${kbps}k`;
-- bufsize = `${kbps * 2}k`;
++ bufsize = `${kbps}k`; // 1x VBV, same as the presets
}
// Video encoding — CBR with zerolatency for stable real-time RTMP delivery
@@@ -809,13 +844,20 @@ class RestreamManager extends EventEmitter {
const binLabel = useOpenSslBinary ? 'ffmpeg(openssl)' : 'ffmpeg';
console.log(`[Restream] Spawning FFmpeg for ${session.key}: ${binLabel} ${maskedArgs.join(' ')}`);
-- const proc = spawn(ffmpegBin, args, {
++ // `-progress pipe:1` is inserted after the global flags (args start with -hide_banner):
++ // ffmpeg then reports frame/fps/bitrate/speed on stdout, which is both the live ACK and
++ // the health line the dashboard shows.
++ const spawnArgs = args[0] === '-hide_banner' ? ['-hide_banner', '-progress', 'pipe:1', '-stats_period', '1', ...args.slice(1)] : ['-progress', 'pipe:1', ...args];
++ const proc = spawn(ffmpegBin, spawnArgs, {
stdio: ['pipe', 'pipe', 'pipe'],
});
session.process = proc;
session.startedAt = Date.now();
++ session.liveAt = null;
session.liveConfirmedThisRun = false; // reset per ffmpeg run; set true once this run reaches 'live'
++ session.progress = null;
++ session.lastErrorRaw = null;
let stderrBuf = '';
let liveConfirmed = false;
@@@ -833,6 +875,8 @@ class RestreamManager extends EventEmitter {
try { require('../db/database').clearRestreamDestinationCooldown(session.destId); } catch { /* */ }
session.status = 'live';
++ session.liveAt = Date.now();
++ session.lastError = null;
this.emit('status-change', {
streamId: session.streamId, destId: session.destId, status: 'live',
});
@@@ -854,16 +898,36 @@ class RestreamManager extends EventEmitter {
const chunk = data.toString();
stderrBuf += chunk;
if (stderrBuf.length > 4096) stderrBuf = stderrBuf.slice(-4096);
++ });
-- // FFmpeg prints "Output #0" when it opens the output and "frame=" when encoding frames.
-- // Either one confirms the RTMP connection is alive.
-- if (!liveConfirmed && (chunk.includes('Output #0') || chunk.includes('frame='))) {
-- confirmLive();
++ // Progress blocks: one key=value per line, terminated by `progress=continue|end`.
++ let progressBuf = '';
++ let progressSeenAt = 0;
++ proc.stdout.on('data', (data) => {
++ progressBuf += data.toString();
++ if (progressBuf.length > 8192) progressBuf = progressBuf.slice(-8192);
++ let nl;
++ const cur = session.progress || {};
++ while ((nl = progressBuf.indexOf('\n')) >= 0) {
++ const line = progressBuf.slice(0, nl).trim(); progressBuf = progressBuf.slice(nl + 1);
++ const eq = line.indexOf('='); if (eq <= 0) continue;
++ const k = line.slice(0, eq), v = line.slice(eq + 1).trim();
++ if (k === 'frame') cur.frame = +v || 0;
++ else if (k === 'fps') cur.fps = +v || 0;
++ else if (k === 'bitrate') cur.bitrate_kbps = v.endsWith('kbits/s') ? +parseFloat(v) || 0 : null;
++ else if (k === 'speed') cur.speed = v === 'N/A' ? null : +parseFloat(v) || 0;
++ else if (k === 'drop_frames') cur.dropped = +v || 0;
++ else if (k === 'out_time_us' || k === 'out_time_ms') cur.out_ms = Math.round((+v || 0) / 1000);
++ else if (k === 'progress') {
++ progressSeenAt = Date.now();
++ cur.at = progressSeenAt;
++ session.progress = cur;
++ // Frames (or, for a codec copy, output time) advancing means the ingest is taking data.
++ if (!liveConfirmed && ((cur.frame || 0) > 0 || (cur.out_ms || 0) > 0)) confirmLive();
++ }
}
});
-- proc.stdout.on('data', () => {}); // drain stdout
--
proc.on('error', (err) => {
this._disposeTransientSessionState(session, proc);
console.error(`[Restream] FFmpeg spawn error for ${session.key}:`, err.message);
@@@ -891,7 +955,8 @@ class RestreamManager extends EventEmitter {
streamId: session.streamId, destId: session.destId, status: 'idle',
});
} else {
-- session.lastError = lastLines || `FFmpeg exit code ${code}`;
++ session.lastErrorRaw = lastLines || `FFmpeg exit code ${code}`;
++ session.lastError = RestreamManager.friendlyFfmpegError(session.lastErrorRaw, session.destination?.platform);
session.status = 'error';
this.emit('status-change', {
streamId: session.streamId, destId: session.destId,
@@@ -901,14 +966,18 @@ class RestreamManager extends EventEmitter {
}
});
-- // Fallback: if stderr parsing hasn't confirmed live after 15s and FFmpeg is
-- // still running, assume it's connected (some FFmpeg builds suppress output)
++ // No progress at all within the ACK window means the output never opened (an ingest that
++ // accepts the TCP connection and then sits on the handshake, a key that is silently
++ // ignored): kill this run so the restart/back-off machinery — and the dashboard — see a
++ // real error instead of a "Starting…" that never ends.
setTimeout(() => {
-- if (!liveConfirmed && session.process === proc && session.status === 'starting') {
-- console.log(`[Restream] Fallback live confirmation for ${session.key} (no stderr signal after 15s)`);
-- confirmLive();
-- }
-- }, 15000);
++ if (liveConfirmed || session.process !== proc || session.status !== 'starting') return;
++ if (progressSeenAt) { confirmLive(); return; } // progress arrived but no frames yet: treat as connected
++ session.lastErrorRaw = stderrBuf.split('\n').filter(Boolean).slice(-3).join(' | ') || 'no output progress';
++ session.lastError = `No response from the ingest within ${LIVE_ACK_TIMEOUT_MS / 1000}s — ${RestreamManager.friendlyFfmpegError(session.lastErrorRaw, session.destination?.platform)}`;
++ console.warn(`[Restream] ${session.key}: ${session.lastError}`);
++ try { proc.kill('SIGTERM'); } catch { /* */ }
++ }, LIVE_ACK_TIMEOUT_MS).unref?.();
}
/**
@@@ -960,9 +1029,11 @@ class RestreamManager extends EventEmitter {
session.restartAttempts++;
console.log(`[Restream] Scheduling restart for ${session.key} in ${(delay / 1000).toFixed(1)}s (attempt ${session.restartAttempts}/${MAX_RESTART_ATTEMPTS}, ran ${(runtime / 1000).toFixed(1)}s)`);
++ session.nextRestartAt = Date.now() + delay;
session.restartTimer = setTimeout(() => {
session.restartTimer = null;
++ session.nextRestartAt = null;
if (session.status === 'stopped') return;
// Lazy-require db to avoid circular dependency at module load
@@@ -1672,41 +1743,26 @@ class RestreamManager extends EventEmitter {
statuses.push({
destId: session.destId,
++ platform: session.destination?.platform || null,
status: effectiveStatus,
startedAt: session.startedAt,
++ liveAt: session.liveAt || null,
++ uptimeMs: session.liveAt && effectiveStatus === 'live' ? Date.now() - session.liveAt : 0,
restartAttempts: session.restartAttempts,
++ maxRestartAttempts: MAX_RESTART_ATTEMPTS,
++ nextRestartAt: session.nextRestartAt || null,
lastError: effectiveError,
++ transport: RestreamManager.isSrtUrl(session.destination?.server_url) ? 'srt' : 'rtmp',
++ progress: session.progress ? {
++ fps: session.progress.fps ?? null, bitrate_kbps: session.progress.bitrate_kbps ?? null,
++ speed: session.progress.speed ?? null, dropped: session.progress.dropped ?? 0, frame: session.progress.frame ?? 0,
++ } : null,
});
}
}
return statuses;
}
-- /**
-- * Auto-start enabled restreams when a stream goes live.
-- * Called by server/index.js when RTMP publishes or JSMPEG channel is created.
-- * @param {number} streamId
-- * @param {number} userId
-- * @param {object} streamInfo - { protocol, streamKey }
-- */
-- async autoStartForStream(streamId, userId, streamInfo) {
-- const db = require('../db/database');
-- const destinations = db.getRestreamDestinationsByUserId(userId);
-- if (!destinations?.length) return;
--
-- for (const dest of destinations) {
-- if (!dest.enabled || !dest.auto_start) continue;
-- if (!dest.server_url || !dest.stream_key) continue;
--
-- console.log(`[Restream] Auto-starting ${dest.platform} restream for stream ${streamId}`);
-- try {
-- await this.startRestream(streamId, dest, streamInfo);
-- } catch (err) {
-- console.warn(`[Restream] Auto-start failed for dest ${dest.id}:`, err.message);
-- }
-- }
-- }
--
/**
* Resume all enabled restreams for a live stream.
* Unlike autoStartForStream, this does NOT require auto_start=1.
ddiff --git a/server/streaming/restream-routes.js b/server/streaming/restream-routes.js
iindex 5018c0eb..4b5f8e43 100644
---- a/server/streaming/restream-routes.js
++++ b/server/streaming/restream-routes.js
@@@ -69,6 +69,10 @@ function sanitizeDest(d) {
quality_preset: d.quality_preset || 'auto',
channel_url: d.channel_url || '',
chat_relay: !!d.chat_relay,
++ transport: /^srt:\/\//i.test(String(d.server_url || '')) ? 'srt' : 'rtmp',
++ srt_latency_ms: d.srt_latency_ms || null,
++ srt_passphrase: undefined, // never leaves the server
++ has_srt_passphrase: !!d.srt_passphrase,
powerchat_relay: d.powerchat_relay !== 0,
powerchat_count_views: d.powerchat_count_views !== 0,
// Circuit-breaker status — non-zero when the destination is paused after repeated failures.
@@@ -124,6 +128,17 @@ function parseCustomOverrides(body) {
overrides.custom_encoder_preset = VALID_ENCODER_PRESETS.includes(body.custom_encoder_preset)
? body.custom_encoder_preset : null;
}
++ // SRT options. Latency is the receiver's buffer window (libsrt default 120ms); the passphrase
++ // must be 10–79 characters per the SRT spec. An empty string clears the passphrase; leaving the
++ // field out keeps the stored one (the form never sees the stored value).
++ if (body.srt_latency_ms !== undefined) {
++ const v = body.srt_latency_ms === null || body.srt_latency_ms === '' ? null : parseInt(body.srt_latency_ms, 10);
++ overrides.srt_latency_ms = (v !== null && Number.isFinite(v) && v >= 20 && v <= 8000) ? v : null;
++ }
++ if (body.srt_passphrase !== undefined && body.srt_passphrase !== '__keep__') {
++ const v = body.srt_passphrase === null ? '' : String(body.srt_passphrase);
++ overrides.srt_passphrase = v.length >= 10 && v.length <= 79 ? v : null;
++ }
return overrides;
}
@@@ -426,10 +441,14 @@ router.get('/status', requireAuth, (req, res) => {
try {
const liveStreams = db.getLiveStreamsByUserId(req.user.id) || [];
const allStatuses = {};
++ const robotstreamer = {};
for (const stream of liveStreams) {
allStatuses[stream.id] = restreamManager.getStreamStatus(stream.id);
++ // Server-side RobotStreamer passthrough state for the same streams (null = no session).
++ try { robotstreamer[stream.id] = require('../integrations/rs-passthrough-relay').status(stream.id); } catch { robotstreamer[stream.id] = null; }
}
-- res.json({ statuses: allStatuses });
++ res.set('Cache-Control', 'no-store');
++ res.json({ statuses: allStatuses, robotstreamer });
} catch (err) {
console.error('[Restream] Status error:', err.message);
res.status(500).json({ error: 'Failed to get restream status' });
ddiff --git a/server/streaming/routes.js b/server/streaming/routes.js
iindex 67dd1e6b..8b152017 100644
---- a/server/streaming/routes.js
++++ b/server/streaming/routes.js
@@@ -322,7 +322,7 @@ router.get('/channel/:username', optionalAuth, async (req, res) => {
const rsVc = robotStreamerService.getRsViewerCount(ls.user_id, slotId);
const hasBridge = robotStreamerService.chatBridges.has(ls.id);
let hasPublish = robotStreamerService._activePublish?.has(ls.id);
-- try { hasPublish = hasPublish || require('../integrations/rs-native-publisher').isActive(ls.id); } catch { /* ignore */ }
++ try { hasPublish = hasPublish || require('../integrations/rs-passthrough-relay').status(ls.id)?.state === 'live'; } catch { /* ignore */ }
const rsActive = hasBridge || hasPublish;
if (rsActive) {
const integration = db.getRobotStreamerIntegrationForStream(ls.user_id, slotId);
ddiff --git a/test/restream-manager-lifecycle.test.js b/test/restream-manager-lifecycle.test.js
nnew file mode 100644
iindex 00000000..adc962cb
---- /dev/null
++++ b/test/restream-manager-lifecycle.test.js
@@@ -0,0 +1,123 @@
++/**
++ * ffmpeg restreams: destination URLs, per-transport output flags, human-readable failure text,
++ * the progress-based live acknowledgement and the status the dashboard polls.
++ *
++ * node test/restream-manager-lifecycle.test.js
++ */
++'use strict';
++const assert = require('assert');
++const { spawnSync } = require('child_process');
++process.env.NODE_ENV = 'test';
++const quiet = console.log;
++console.log = () => {};
++console.warn = () => {};
++console.error = () => {};
++
++const manager = require('../server/streaming/restream-manager');
++const Manager = manager.constructor;
++
++// ── Destination URLs ─────────────────────────────────────────────────────────────
++assert.strictEqual(manager._buildDestUrl({ id: 1, platform: 'custom', server_url: 'rtmp://a.example/live/', stream_key: 'k1' }), 'rtmp://a.example/live/k1');
++assert.strictEqual(manager._buildDestUrl({ id: 2, platform: 'twitch', server_url: 'rtmp://live.twitch.tv/app', stream_key: 'k2' }), 'rtmps://live.twitch.tv/app/k2', 'twitch is upgraded to rtmps');
++assert.strictEqual(manager._buildDestUrl({ id: 3, platform: 'kick', server_url: 'rtmps://x.kick.com', stream_key: 'k3' }), 'rtmps://x.kick.com/app/k3', 'kick gets /app');
++assert.strictEqual(manager._buildDestUrl({ id: 4, platform: 'custom', server_url: 'http://evil/', stream_key: 'k' }), null, 'non-ingest protocols are refused');
++
++const srt = new URL(manager._buildDestUrl({ id: 5, platform: 'custom', server_url: 'srt://ingest.example:9000', stream_key: 'publish/abc', srt_latency_ms: 250, srt_passphrase: 'correct horse battery' }));
++assert.strictEqual(srt.protocol, 'srt:');
++assert.strictEqual(srt.host, 'ingest.example:9000');
++assert.strictEqual(srt.searchParams.get('streamid'), 'publish/abc', 'stream key becomes the streamid');
++assert.strictEqual(srt.searchParams.get('latency'), '250000', 'latency in microseconds');
++assert.strictEqual(srt.searchParams.get('passphrase'), 'correct horse battery');
++assert.strictEqual(srt.searchParams.get('mode'), 'caller');
++assert.strictEqual(srt.searchParams.get('pkt_size'), '1316');
++const srtDefault = new URL(manager._buildDestUrl({ id: 6, platform: 'custom', server_url: 'srt://h:1?streamid=given', stream_key: 'ignored-when-url-has-one' }));
++assert.strictEqual(srtDefault.searchParams.get('streamid'), 'given');
++assert.strictEqual(srtDefault.searchParams.get('latency'), '120000', 'libsrt default latency');
++assert.ok(!srtDefault.searchParams.has('passphrase'));
++
++// ── Output flags per transport ───────────────────────────────────────────────────
++const flv = manager._outputArgs('rtmps://live.twitch.tv/app/k');
++assert.ok(flv.includes('flv') && flv.includes('-rtmp_live') && flv.includes('-flush_packets'), 'RTMP → FLV, real-time flags');
++const ts = manager._outputArgs('srt://h:1?streamid=x');
++assert.ok(ts.includes('mpegts') && !ts.includes('flv') && !ts.includes('-rtmp_live'), 'SRT → MPEG-TS, no RTMP-only flags');
++assert.strictEqual(ts[ts.length - 1], 'srt://h:1?streamid=x', 'destination is the last argument (the spawn path masks it by position)');
++assert.ok(Manager.isSrtUrl('SRT://x') && !Manager.isSrtUrl('rtmp://x'));
++
++// ── 1x VBV (latency) ─────────────────────────────────────────────────────────────
++const presets = Manager.getQualityPresets();
++for (const [k, p] of Object.entries(presets)) {
++ if (!p.bufsize || !p.videoBitrate) continue;
++ assert.strictEqual(p.bufsize, p.videoBitrate, `${k}: bufsize equals bitrate`);
++}
++const custom = manager._getEncodingArgs(presets.medium, { customOverrides: { videoBitrate: '3000k' } });
++assert.strictEqual(custom[custom.indexOf('-bufsize') + 1], '3000k', 'custom bitrate keeps 1x VBV');
++assert.strictEqual(custom[custom.indexOf('-preset') + 1], presets.medium.preset);
++const withPreset = manager._getEncodingArgs(presets.medium, { customOverrides: { encoderPreset: 'faster' } });
++assert.strictEqual(withPreset[withPreset.indexOf('-preset') + 1], 'faster', 'encoder preset override reaches ffmpeg');
++assert.deepStrictEqual(manager._getCustomOverrides({ custom_encoder_preset: 'fast', custom_fps: 60 }), { fps: 60, encoderPreset: 'fast' });
++assert.deepStrictEqual(manager._getCustomOverrides({ custom_encoder_preset: 'placebo' }), {}, 'unknown preset ignored');
++
++// ── Friendly errors ──────────────────────────────────────────────────────────────
++const fe = Manager.friendlyFfmpegError;
++assert.match(fe('rtmp://x: Server error: NetStream.Publish.BadName', 'twitch'), /Twitch rejected the stream key/);
++assert.match(fe('tcp://x: Connection refused', 'kick'), /Could not connect to Kick/);
++assert.match(fe('Failed to resolve hostname live.example: Name or service not known', 'custom'), /resolve the ingest host/);
++assert.match(fe('gnutls: The TLS connection was non-properly terminated', 'youtube'), /TLS handshake/);
++assert.match(fe('av_interleaved_write_frame(): Broken pipe | End of file', 'twitch'), /closed the connection/);
++assert.match(fe('[srt @ 0x1] Wrong password: passphrase', 'custom'), /SRT passphrase/);
++assert.strictEqual(fe('', 'custom'), 'Restream process stopped without a message');
++assert.ok(fe('x'.repeat(500), 'custom').length <= 160, 'unknown text is truncated');
++
++// ── Status fields ────────────────────────────────────────────────────────────────
++manager.sessions.set('1:9', {
++ key: '1:9', streamId: 1, destId: 9, destination: { platform: 'custom', server_url: 'srt://h:1' },
++ status: 'live', startedAt: Date.now() - 5000, liveAt: Date.now() - 4000, restartAttempts: 2,
++ nextRestartAt: null, lastError: null, progress: { fps: 30, bitrate_kbps: 2500.5, speed: 1.0, dropped: 0, frame: 120 },
++});
++const [st] = manager.getStreamStatus(1);
++assert.strictEqual(st.status, 'live');
++assert.strictEqual(st.transport, 'srt');
++assert.ok(st.uptimeMs >= 3900 && st.uptimeMs < 6000, 'uptime from liveAt');
++assert.strictEqual(st.maxRestartAttempts, 30);
++assert.deepStrictEqual(st.progress, { fps: 30, bitrate_kbps: 2500.5, speed: 1.0, dropped: 0, frame: 120 });
++manager.sessions.delete('1:9');
++
++// ── Progress-based live ACK against a real ffmpeg (skipped when none is installed) ──
++(async () => {
++ const have = spawnSync('ffmpeg', ['-version']).status === 0;
++ if (!have) { console.log = quiet; console.log('restream-manager lifecycle: checks passed (ffmpeg not installed — ACK test skipped)'); process.exit(0); }
++
++ const events = [];
++ const session = {
++ key: 't:1', streamId: 7, destId: 1, destination: { platform: 'custom', server_url: 'rtmp://x' },
++ status: 'starting', process: null, startedAt: null, restartAttempts: 0, restartDelay: 5000,
++ restartTimer: null, stableTimer: null, lastError: null, dataTapCleanup: null, streamInfo: { protocol: 'rtmp' },
++ };
++ manager.sessions.set(session.key, session);
++ manager.on('status-change', (e) => { if (e.destId === 1 && e.streamId === 7) events.push(e.status); });
++ // A 1-second synthetic source into the null muxer: ffmpeg reports progress with frame>0,
++ // which must flip the session to live; exit 0 then makes it idle (not an error/restart).
++ manager._spawnFFmpeg(session, ['-hide_banner', '-loglevel', 'warning', '-f', 'lavfi', '-i', 'testsrc=duration=1:size=64x64:rate=10', '-f', 'null', '-']);
++ const proc = session.process;
++ assert.ok(proc.spawnargs.includes('-progress') && proc.spawnargs.includes('pipe:1'), 'progress goes to stdout');
++ await new Promise((r) => proc.on('close', () => setTimeout(r, 50)));
++ assert.deepStrictEqual(events, ['live', 'idle'], `live acknowledged from progress, clean exit → idle (got ${events})`);
++ assert.ok(session.progress && session.progress.frame >= 10, `progress captured: ${JSON.stringify(session.progress)}`);
++ assert.strictEqual(session.lastError, null);
++ manager.sessions.delete(session.key);
++
++ // Failure path: an output that cannot be opened ends with a friendly error and a scheduled restart.
++ const bad = { ...session, key: 't:2', destId: 2, status: 'starting', process: null, progress: null, destination: { platform: 'custom', server_url: 'rtmp://127.0.0.1:1' } };
++ manager.sessions.set(bad.key, bad);
++ manager._spawnFFmpeg(bad, ['-hide_banner', '-loglevel', 'warning', '-f', 'lavfi', '-i', 'testsrc=duration=1:size=64x64:rate=10', '-c:v', 'libx264', '-preset', 'ultrafast', '-f', 'flv', 'rtmp://127.0.0.1:1/live/k']);
++ await new Promise((r) => bad.process.on('close', () => setTimeout(r, 50)));
++ assert.strictEqual(bad.status, 'error');
++ assert.match(bad.lastError, /Could not connect to the destination/, bad.lastError);
++ assert.ok(bad.restartTimer, 'restart scheduled');
++ assert.ok(bad.nextRestartAt > Date.now(), 'dashboard can count down to the retry');
++ bad.status = 'stopped'; clearTimeout(bad.restartTimer); manager.sessions.delete(bad.key);
++
++ console.log = quiet;
++ console.log('restream-manager lifecycle: all checks passed');
++ process.exit(0);
++})().catch((err) => { console.log = quiet; console.error = process.stderr.write.bind(process.stderr); console.error(err.stack + '\n'); process.exit(1); });
ddiff --git a/test/rs-passthrough-lifecycle.test.js b/test/rs-passthrough-lifecycle.test.js
nnew file mode 100644
iindex 00000000..a89c7bd1
---- /dev/null
++++ b/test/rs-passthrough-lifecycle.test.js
@@@ -0,0 +1,106 @@
++/**
++ * RobotStreamer passthrough lifecycle: the relay must never outlive its source.
++ *
++ * The reported bug: a streamer stops, and robotstreamer.com keeps showing the robot live with a
++ * black picture. The relay kept its werift peer open (RS keeps a robot live as long as our
++ * producers exist) and its restart loop ran flat-out, forever, with no look at whether the
++ * OpenVibe stream was still live. These checks pin the new behaviour:
++ *
++ * - a restart is refused, and the session stopped, when the source stream is not live
++ * - restarts back off and are counted per session; the cap turns the session into `failed`
++ * - a failed session stays visible through status() and an explicit start() relaunches it
++ * - stop() clears every timer and removes the session
++ *
++ * node test/rs-passthrough-lifecycle.test.js
++ */
++'use strict';
++const assert = require('assert');
++process.env.NODE_ENV = 'test';
++const quiet = console.log;
++console.log = () => {};
++console.warn = () => {};
++
++const relay = require('../server/integrations/rs-passthrough-relay');
++const tick = (ms = 5) => new Promise((r) => setTimeout(r, ms));
++
++(async () => {
++ assert.ok(relay.available(), 'werift is installed in this checkout');
++ const stream = { id: 4242 };
++ const integration = { robot_id: 'r-1', token: 't-1' };
++
++ // 1. Source not live → the first failure stops the session instead of restarting it.
++ // (_run throws at once here: the SFU is not initialised in tests, which is a real
++ // failure mode — "mediasoup SFU not ready" — and exercises the same path.)
++ relay._sourceLiveOverride = () => false;
++ assert.strictEqual(await relay.start(stream, integration), true);
++ await tick();
++ assert.strictEqual(relay.isActive(stream.id), false, 'session stopped because the source is not live');
++ assert.strictEqual(relay.status(stream.id), null);
++
++ // 2. Source live → restart is scheduled with backoff and reported through status().
++ relay._sourceLiveOverride = () => true;
++ await relay.start(stream, integration);
++ await tick();
++ let st = relay.status(stream.id);
++ assert.ok(st, 'session exists');
++ assert.strictEqual(st.state, 'restarting');
++ assert.strictEqual(st.restarts, 1);
++ assert.ok(st.next_restart_at > Date.now() + 1000, 'first retry is ~3s out');
++ assert.ok(/SFU not ready/.test(st.last_restart_reason), `reason names the cause: ${st.last_restart_reason}`);
++ assert.strictEqual(st.active, true);
++ const session = relay.sessions.get(stream.id);
++ assert.ok(session.restartTimer, 'restart timer armed');
++
++ // 3. The source ends while a restart is pending → the timer fires into stop().
++ relay._sourceLiveOverride = () => false;
++ clearTimeout(session.restartTimer); session.restartTimer = null;
++ relay._scheduleRestart(session, 'second failure');
++ await tick();
++ assert.strictEqual(relay.isActive(stream.id), false, 'a restart attempt after the source ended stops the session');
++
++ // 4. Cap: too many consecutive restarts → failed, still visible, not active.
++ relay._sourceLiveOverride = () => true;
++ await relay.start(stream, integration);
++ await tick();
++ const s2 = relay.sessions.get(stream.id);
++ clearTimeout(s2.restartTimer); s2.restartTimer = null;
++ s2.restarts = 12;
++ relay._scheduleRestart(s2, 'werift connectionState=failed');
++ st = relay.status(stream.id);
++ assert.strictEqual(st.state, 'failed');
++ assert.strictEqual(st.active, false);
++ assert.ok(/gave up after 12 restarts/.test(st.last_error), st.last_error);
++ assert.strictEqual(s2.restartTimer, null, 'no timer left behind after giving up');
++
++ // 5. Explicit start() on a failed session relaunches it from zero.
++ await relay.start(stream, integration);
++ await tick();
++ st = relay.status(stream.id);
++ assert.strictEqual(st.restarts, 1, 'counter reset by the relaunch');
++ assert.strictEqual(st.state, 'restarting');
++
++ // 6. Backoff grows and is capped.
++ const s3 = relay.sessions.get(stream.id);
++ clearTimeout(s3.restartTimer); s3.restartTimer = null;
++ s3.restarts = 5;
++ relay._scheduleRestart(s3, 'x');
++ const delay6 = s3.nextRestartAt - Date.now();
++ assert.ok(delay6 > 25000 && delay6 <= 30000, `6th retry is capped at 30s (got ${delay6}ms)`);
++
++ // 7. stop() clears everything.
++ relay.stop(stream.id);
++ assert.strictEqual(relay.isActive(stream.id), false);
++ assert.strictEqual(s3.restartTimer, null);
++ assert.strictEqual(s3.stopped, true);
++
++ // 8. status() never leaks the RS token.
++ await relay.start(stream, integration);
++ await tick();
++ assert.ok(!JSON.stringify(relay.status(stream.id)).includes('t-1'), 'token not in status');
++ relay.stop(stream.id);
++
++ delete relay._sourceLiveOverride;
++ console.log = quiet;
++ console.log('rs-passthrough lifecycle: all checks passed');
++ process.exit(0);
++})().catch((err) => { console.log = quiet; console.error(err); process.exit(1); });