openvibers/openvibe.live Commits

RS chat: stop mirroring every message once per racing start

OP OpenVibers committed 20 Aug 2026 · 2a73f9d27e · 4 files, +304 −25

RobotStreamer chat arrived in OpenVibe chat two or three times per message.
startForStream() is called from four lifecycle hooks — stream creation, the WHIP
ingest going live, the boot-time restore sweep, and the integrations route — and
several fire within milliseconds for the same stream. The method is async and
awaits refreshIntegration() BETWEEN reading chatBridges and populating it, so
every concurrent caller passed the "already have a bridge" guard, opened its own
websocket to RobotStreamer, and then chatBridges.set() kept only the last one.
The earlier bridges stayed connected and kept mirroring, so the duplicate count
was simply how many callers had raced.

Share one in-flight promise per stream so the guard actually holds, and add a
last-resort check at registration that discards a duplicate rather than
overwriting the tracked bridge and leaking a websocket that keeps broadcasting.
chat-relay-service has the same four-caller shape but its startBridge is
synchronous, so it cannot race and is left alone.

Also moves two settings into the Settings tab, each as its own collapsible
section matching the existing groups:
  - Control Profile, previously in the Stream tab.
  - Picture-in-picture Camera, previously an inline row, now with its starting
    corner and size. Those are stored as {x,y,w} fractions of the player box so
    the overlay keeps its relative place at any window size; the UI offers a
    corner and a size and converts, since nobody should have to think in
    fractions to put a webcam in the bottom right.

Co-Authored-By: Claude Opus 5 <[email protected]>

Changes

ddiff --git a/public/js/broadcast-workspace.js b/public/js/broadcast-workspace.js
iindex 23e90a3b..193e5e68 100644
---- a/public/js/broadcast-workspace.js
++++ b/public/js/broadcast-workspace.js
@@@ -617,14 +617,8 @@ function _wsRenderPanel() {
</details>
</div>
-- <!-- Control profile (Class A structured) -->
-- <div class="form-group">
-- <label><i class="fa-solid fa-gamepad"></i> Control Profile</label>
-- <select id="bc-control-config" class="form-input form-input-sm"
-- onchange="_wsMarkDirty()">
-- <option value="">None (no controls)</option>
-- </select>
-- </div>
++ <!-- Control profile lives under Settings; createNewStream() and the save
++ payload both read #bc-control-config, which is rendered there. -->
<!-- Hidden element required by createNewStream() -->
<select id="bc-managed-stream" style="display:none">
@@@ -723,6 +717,65 @@ function _wsRenderPanel() {
<div class="bc-ws-tab-panel" data-wstabpanel="settings">
<div class="bc-ws-profile-section">
++ <!-- Control Profile -->
++ <details class="bc-ws-slot-settings">
++ <summary><i class="fa-solid fa-gamepad"></i> Control Profile</summary>
++ <div class="bc-ws-slot-settings-inner">
++ <p class="muted" style="font-size:0.82rem;margin-bottom:10px">
++ Lets viewers send commands to hardware attached to this slot. Leave as None
++ for an ordinary stream.
++ </p>
++ <div class="form-group">
++ <select id="bc-control-config" class="form-input form-input-sm" onchange="_wsMarkDirty()">
++ <option value="">None (no controls)</option>
++ </select>
++ </div>
++ </div>
++ </details>
++
++ <!-- Picture-in-picture camera -->
++ <details class="bc-ws-slot-settings">
++ <summary><i class="fa-solid fa-video"></i> Picture-in-picture Camera</summary>
++ <div class="bc-ws-slot-settings-inner">
++ <p class="muted" style="font-size:0.82rem;margin-bottom:10px">
++ Overlay another of your slots on top of this stream. The camera stays a stream
++ in its own right — its own VOD, clips and restreams — so it can be your webcam,
++ a second angle, or a co-host's slot. Viewers can move, resize, mute or hide it
++ themselves; the values below are only where it starts.
++ </p>
++ <div class="form-group">
++ <label>Camera slot</label>
++ <select id="bc-ws-pip-source" class="form-input" onchange="_wsSlotSettingChanged();_wsPipDefaultsToggle()">
++ <option value="">None</option>
++ ${(_wsState.managedStreams || []).filter(o => o.id !== ms.id).map(o =>
++ `<option value="${o.id}"${String(ms.pip_source_msid || '') === String(o.id) ? ' selected' : ''}>${esc(o.title || ('Slot ' + o.id))}</option>`
++ ).join('')}
++ </select>
++ </div>
++ <div id="bc-ws-pip-defaults" style="display:${ms.pip_source_msid ? '' : 'none'}">
++ <div class="bc-ws-row">
++ <div class="form-group" style="flex:1">
++ <label>Starting corner</label>
++ <select id="bc-ws-pip-corner" class="form-input form-input-sm" onchange="_wsSlotSettingChanged()">
++ <option value="br">Bottom Right</option>
++ <option value="bl">Bottom Left</option>
++ <option value="tr">Top Right</option>
++ <option value="tl">Top Left</option>
++ </select>
++ </div>
++ <div class="form-group" style="flex:1">
++ <label>Starting size</label>
++ <select id="bc-ws-pip-size" class="form-input form-input-sm" onchange="_wsSlotSettingChanged()">
++ <option value="0.18">Small</option>
++ <option value="0.25">Medium</option>
++ <option value="0.35">Large</option>
++ </select>
++ </div>
++ </div>
++ </div>
++ </div>
++ </details>
++
<!-- Restream Destinations -->
<details class="bc-ws-slot-settings" open>
<summary><i class="fa-solid fa-tower-broadcast"></i> Restream Destinations</summary>
@@@ -944,23 +997,6 @@ function _wsRenderPanel() {
</label>
</div>
</div>
-- <div class="bc-ws-row">
-- <div class="form-group" style="flex:1">
-- <label>Picture-in-picture camera</label>
-- <select id="bc-ws-pip-source" class="form-input" onchange="_wsSlotSettingChanged()">
-- <option value="">None</option>
-- ${(_wsState.managedStreams || []).filter(o => o.id !== ms.id).map(o =>
-- `<option value="${o.id}"${String(ms.pip_source_msid || '') === String(o.id) ? ' selected' : ''}>${esc(o.title || ('Slot ' + o.id))}</option>`
-- ).join('')}
-- </select>
-- <p class="bc-ws-hint">
-- Viewers of this stream see the chosen slot's live stream as a movable,
-- resizable overlay. It stays a stream in its own right — its own VOD,
-- clips and restreams — so it can be your webcam, a second angle, or
-- someone else's slot entirely.
-- </p>
-- </div>
-- </div>
<div class="bc-ws-row">
<div class="form-group" style="flex:1">
<label>Default VOD Visibility</label>
@@@ -1065,6 +1101,8 @@ function _wsRenderPanel() {
// Async post-render: populate control configs and check device permissions
_wsPopulateControlConfigs(ms.control_config_id);
++ // Corner/size selects are derived from the stored {x,y,w} fractions.
++ _wsPipDefaultsRestore(ms);
if (method === 'browser') {
_wsInitDevicePicker(method, browserMode);
_wsSyncScreenShareHiddenDefaults(browserMode);
@@@ -1888,6 +1926,11 @@ async function _wsSaveAll() {
weather_show_location: ms.weather_show_location || 0,
// '' clears the overlay; the server validates ownership and rejects self-reference.
pip_source_msid: document.getElementById('bc-ws-pip-source')?.value || null,
++ pip_defaults: (() => {
++ const w = Number(document.getElementById('bc-ws-pip-size')?.value) || 0.25;
++ const corner = document.getElementById('bc-ws-pip-corner')?.value || 'br';
++ return { ...(_wsPipCornerToXY(corner, w)), w };
++ })(),
};
const profileBlob = {
@@@ -2180,6 +2223,61 @@ async function _wsRequestCamPermission() {
_wsUpdateCaptureSummary();
}
++
++/* ── Picture-in-picture defaults ──────────────────────────────────────────────
++ The API stores the starting geometry as FRACTIONS of the player box ({x,y,w}),
++ so the overlay lands in the same relative place at any window size. Streamers
++ should not have to think in fractions, so the UI offers a corner and a size and
++ converts. 0.02 is the margin from each edge. */
++
++const _PIP_MARGIN = 0.02;
++
++function _wsPipCornerToXY(corner, w) {
++ const h = w * 9 / 16; // the overlay is 16:9
++ const right = 1 - w - _PIP_MARGIN;
++ const bottom = 1 - h - _PIP_MARGIN;
++ switch (corner) {
++ case 'bl': return { x: _PIP_MARGIN, y: bottom };
++ case 'tr': return { x: right, y: _PIP_MARGIN };
++ case 'tl': return { x: _PIP_MARGIN, y: _PIP_MARGIN };
++ default: return { x: right, y: bottom }; // 'br'
++ }
++}
++
++function _wsPipXYToCorner(x, y) {
++ return (y < 0.5 ? 't' : 'b') + (x < 0.5 ? 'l' : 'r');
++}
++
++/** Show the geometry controls only when a camera slot is actually selected. */
++function _wsPipDefaultsToggle() {
++ const on = !!document.getElementById('bc-ws-pip-source')?.value;
++ _wsShowEl('bc-ws-pip-defaults', on);
++}
++
++/** Restore corner/size from the stored fractions when the panel renders. */
++function _wsPipDefaultsRestore(ms) {
++ let d = {};
++ try { d = typeof ms.pip_defaults === 'string' ? JSON.parse(ms.pip_defaults || '{}') : (ms.pip_defaults || {}); }
++ catch { d = {}; }
++ const sizeEl = document.getElementById('bc-ws-pip-size');
++ const cornerEl = document.getElementById('bc-ws-pip-corner');
++ if (sizeEl) {
++ const w = Number(d.w);
++ // Snap to the nearest offered size rather than showing an option that is not there.
++ const opts = [0.18, 0.25, 0.35];
++ const nearest = Number.isFinite(w)
++ ? opts.reduce((a, b) => (Math.abs(b - w) < Math.abs(a - w) ? b : a))
++ : 0.25;
++ sizeEl.value = String(nearest);
++ }
++ if (cornerEl) {
++ cornerEl.value = (Number.isFinite(Number(d.x)) && Number.isFinite(Number(d.y)))
++ ? _wsPipXYToCorner(Number(d.x), Number(d.y))
++ : 'br';
++ }
++ _wsPipDefaultsToggle();
++}
++
function _wsPopulateInlineDevices(selectId, devices) {
const sel = document.getElementById(selectId);
if (!sel) return;
ddiff --git a/server/integrations/robotstreamer-service.js b/server/integrations/robotstreamer-service.js
iindex c1acdf4a..23fd4ffd 100644
---- a/server/integrations/robotstreamer-service.js
++++ b/server/integrations/robotstreamer-service.js
@@@ -33,6 +33,8 @@ const RS_REFRESH_CACHE_MS = 5 * 60 * 1000;
class RobotStreamerService {
constructor() {
this.chatBridges = new Map();
++ /** streamId -> in-flight startForStream promise (see startForStream). */
++ this._startingStreams = new Map();
/** @type {Map<number, { ws: WebSocket, upstream: WebSocket|null, connectedAt: number }>} streamId → active publish session */
this._activePublish = new Map();
/** @type {Map<string, { count: number, fetchedAt: number }>} `userId:slotId` → cached RS viewer count */
@@@ -541,8 +543,32 @@ class RobotStreamerService {
}
}
++ /**
++ * Idempotent per stream, including while a previous call is still in flight.
++ *
++ * Four different lifecycle hooks call this — stream creation, the WHIP ingest going
++ * live, the boot-time restore sweep, and the integrations route — and several of them
++ * fire within milliseconds of each other for the same stream. startForStream is async
++ * and awaits refreshIntegration() between checking chatBridges and populating it, so
++ * every concurrent caller used to sail past the "already have a bridge" guard, open
++ * its own websocket to RobotStreamer, and then have chatBridges.set() overwrite the
++ * previous entry. Only the last bridge was tracked; the earlier ones stayed connected
++ * and kept mirroring, so viewers saw each RS chat message repeated once per racing
++ * caller. Sharing one in-flight promise makes the guard actually hold.
++ */
async startForStream(stream, opts = {}) {
if (!stream?.id || !stream?.user_id) return;
++ const inflight = this._startingStreams.get(stream.id);
++ if (inflight) return inflight;
++ const p = this._startForStream(stream, opts).finally(() => {
++ if (this._startingStreams.get(stream.id) === p) this._startingStreams.delete(stream.id);
++ });
++ this._startingStreams.set(stream.id, p);
++ return p;
++ }
++
++ async _startForStream(stream, opts = {}) {
++ if (!stream?.id || !stream?.user_id) return;
let integration = this.getIntegrationForStream(stream);
if (!integration?.enabled || !integration.token || !integration.robot_id) {
@@@ -744,6 +770,16 @@ class RobotStreamerService {
}
};
++ // Last-resort guard: if anything still managed to build a second bridge for this
++ // stream, close THIS one rather than overwriting the tracked entry and leaking a
++ // websocket that keeps mirroring into the same chat.
++ const raced = this.chatBridges.get(stream.id);
++ if (raced) {
++ console.warn(`[RS] Chat bridge for stream ${stream.id} already exists — discarding the duplicate`);
++ try { bridge.disconnect(); } catch { /* never connected */ }
++ return raced;
++ }
++
this.chatBridges.set(stream.id, bridge);
connect();
return bridge;
ddiff --git a/test/pip-camera-slot.test.js b/test/pip-camera-slot.test.js
iindex df3c1418..63fe2e65 100644
---- a/test/pip-camera-slot.test.js
++++ b/test/pip-camera-slot.test.js
@@@ -109,6 +109,65 @@ assert.ok(/raw\.muted !== false/.test(pip),
assert.ok(/is-offline/.test(pip), 'a camera going offline must degrade to a badge, not a broken overlay');
console.log('OK I: geometry persists, hide is recoverable, audio defaults to muted, offline degrades gracefully');
++// ── J: workspace placement — Settings owns the control profile and the PiP section ───
++{
++ const ws = read('public/js/broadcast-workspace.js');
++ const at = (name) => ws.indexOf(`data-wstabpanel="${name}"`);
++ const [prof, settings, endpoint] = [at('profile'), at('settings'), at('endpoint')];
++ assert.ok(prof > 0 && settings > prof && endpoint > settings, 'expected profile/settings/endpoint panels in order');
++ const streamPanel = ws.slice(prof, settings);
++ const settingsPanel = ws.slice(settings, endpoint);
++
++ assert.ok(!streamPanel.includes('id="bc-control-config"'),
++ 'the control profile select must no longer render in the Stream tab');
++ assert.ok(settingsPanel.includes('id="bc-control-config"'),
++ 'the control profile select must render in the Settings tab');
++ assert.ok(settingsPanel.includes('id="bc-ws-pip-source"'),
++ 'the PiP camera picker must render in the Settings tab');
++
++ // Both must be their own collapsible sections, matching the other Settings groups.
++ for (const label of ['Control Profile', 'Picture-in-picture Camera']) {
++ // Anchor on the SUMMARY text: an HTML comment carries the same label just above
++ // the <details>, and matching that would look in the wrong place entirely.
++ const idx = settingsPanel.indexOf(`${label}</summary>`);
++ assert.ok(idx > 0, `Settings must have a "${label}" collapsible section`);
++ const before = settingsPanel.slice(Math.max(0, idx - 400), idx);
++ assert.ok(before.includes('<details class="bc-ws-slot-settings">'),
++ `"${label}" must be a collapsible <details> section like the rest of Settings`);
++ }
++ console.log('OK J: control profile and PiP camera are collapsible sections under Settings');
++}
++
++// ── K: PiP starting geometry round-trips through corner/size ─────────────────────────
++{
++ const ws = read('public/js/broadcast-workspace.js');
++ assert.ok(/function _wsPipCornerToXY/.test(ws) && /function _wsPipXYToCorner/.test(ws),
++ 'corner <-> fraction conversion must exist in both directions');
++ assert.ok(/pip_defaults: \(\(\) => \{/.test(ws), 'the save payload must include pip_defaults');
++ assert.ok(/_wsPipDefaultsRestore\(ms\)/.test(ws), 'stored fractions must be restored into the selects');
++
++ // Exercise the conversion the way the UI does: every corner must land in its own
++ // quadrant and stay fully inside the player box.
++ const MARGIN = 0.02;
++ const cornerToXY = (corner, w) => {
++ const h = w * 9 / 16, right = 1 - w - MARGIN, bottom = 1 - h - MARGIN;
++ return corner === 'bl' ? { x: MARGIN, y: bottom }
++ : corner === 'tr' ? { x: right, y: MARGIN }
++ : corner === 'tl' ? { x: MARGIN, y: MARGIN }
++ : { x: right, y: bottom };
++ };
++ const xyToCorner = (x, y) => (y < 0.5 ? 't' : 'b') + (x < 0.5 ? 'l' : 'r');
++ for (const w of [0.18, 0.25, 0.35]) {
++ for (const corner of ['br', 'bl', 'tr', 'tl']) {
++ const { x, y } = cornerToXY(corner, w);
++ assert.strictEqual(xyToCorner(x, y), corner, `${corner} @ w=${w} must round-trip`);
++ assert.ok(x >= 0 && y >= 0 && x + w <= 1 && y + (w * 9 / 16) <= 1,
++ `${corner} @ w=${w} must sit fully inside the player`);
++ }
++ }
++ console.log('OK K: every corner/size default round-trips and stays inside the player box');
++}
++
try { fs.unlinkSync(tmp); } catch { /* */ }
for (const ext of ['-wal', '-shm']) { try { fs.unlinkSync(tmp + ext); } catch { /* */ } }
console.log('✅ PiP camera slot test passed');
ddiff --git a/test/rs-chat-bridge-single-flight.test.js b/test/rs-chat-bridge-single-flight.test.js
nnew file mode 100644
iindex 00000000..c28de38d
---- /dev/null
++++ b/test/rs-chat-bridge-single-flight.test.js
@@@ -0,0 +1,86 @@
++/**
++ * Regression: RobotStreamer chat messages were mirrored into OpenVibe chat once per
++ * racing start, so viewers saw every RS message two or three times in a row.
++ *
++ * startForStream() is called from four lifecycle hooks — stream creation, the WHIP
++ * ingest going live, the boot-time restore sweep, and the integrations route — and
++ * several of them fire within milliseconds for the same stream. The method is async and
++ * awaits refreshIntegration() BETWEEN reading chatBridges and populating it, so every
++ * concurrent caller passed the "already have a bridge" guard, opened its own websocket
++ * to RobotStreamer, and then chatBridges.set() kept only the last one. The earlier
++ * bridges stayed connected and kept mirroring: three racing callers, three copies of
++ * every message.
++ *
++ * The fix is a shared in-flight promise, so the guard actually holds. This exercises
++ * that directly rather than asserting on source text.
++ */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const os = require('os');
++
++process.env.DB_PATH = path.join(os.tmpdir(), `ov-rs-chat-${Date.now()}.db`);
++
++const svc = require('../server/integrations/robotstreamer-service');
++
++// ── A: concurrent starts collapse to a single underlying run ─────────────────────────
++{
++ let runs = 0;
++ const original = svc._startForStream;
++ // Stand in for the real work, with an await inside — the await is the whole point:
++ // it is the window the old code left open between guard and registration.
++ svc._startForStream = async function (stream) {
++ runs++;
++ await new Promise(r => setTimeout(r, 25));
++ return { streamId: stream.id, bridge: true };
++ };
++
++ (async () => {
++ const stream = { id: 4242, user_id: 7 };
++ const results = await Promise.all([
++ svc.startForStream(stream),
++ svc.startForStream(stream),
++ svc.startForStream(stream),
++ svc.startForStream(stream),
++ ]);
++ assert.strictEqual(runs, 1, `four concurrent starts must run the work once, ran ${runs}`);
++ assert.ok(results.every(r => r === results[0]), 'every caller must receive the same bridge');
++ console.log('OK A: four concurrent startForStream calls produce exactly one bridge');
++
++ // ── B: once settled, a later start may run again (stream restart must work) ──
++ const after = await svc.startForStream(stream);
++ assert.strictEqual(runs, 2, 'a start AFTER the first settled must be allowed to run');
++ assert.ok(after, 'the later start still returns a bridge');
++ console.log('OK B: single-flight releases once settled, so a genuine restart still works');
++
++ // ── C: different streams are independent ────────────────────────────────────
++ runs = 0;
++ await Promise.all([
++ svc.startForStream({ id: 1, user_id: 7 }),
++ svc.startForStream({ id: 2, user_id: 7 }),
++ ]);
++ assert.strictEqual(runs, 2, 'separate streams must each get their own bridge');
++ console.log('OK C: the in-flight guard is per stream, not global');
++
++ svc._startForStream = original;
++
++ // ── D: the map does not leak entries ────────────────────────────────────────
++ assert.strictEqual(svc._startingStreams.size, 0,
++ 'in-flight entries must be cleared on settle, or a stream can never be started again');
++ console.log('OK D: in-flight map is drained, so a stream is never permanently blocked');
++
++ // ── E: the registration guard refuses to orphan a live bridge ───────────────
++ const src = fs.readFileSync(path.join(__dirname, '../server/integrations/robotstreamer-service.js'), 'utf8');
++ assert.ok(/already exists — discarding the duplicate/.test(src),
++ 'registration must discard a duplicate rather than overwrite the tracked bridge');
++ const setIdx = src.indexOf('this.chatBridges.set(stream.id, bridge);');
++ const guardIdx = src.indexOf('const raced = this.chatBridges.get(stream.id);');
++ assert.ok(guardIdx > 0 && guardIdx < setIdx, 'the guard must sit before the set()');
++ console.log('OK E: registration discards a duplicate instead of leaking the previous websocket');
++
++ try { fs.unlinkSync(process.env.DB_PATH); } catch { /* */ }
++ for (const ext of ['-wal', '-shm']) { try { fs.unlinkSync(process.env.DB_PATH + ext); } catch { /* */ } }
++ console.log('✅ RS chat bridge single-flight test passed');
++ process.exit(0);
++ })().catch((e) => { console.error('FAIL', e.message); process.exit(1); });
++}