openvibers/openvibe.live Commits

One go-live per stream slot: a second request at once, or within 15 s while live, gets the same stream (a double go-live made two streams in one second: chat and TTS split between them and RobotStreamer lines arrived twice) (#234)

OP OpenVibers committed 27 days ago · 23a72cf5d1 · 2 files, +154 −55

Changes

ddiff --git a/server/streaming/routes.js b/server/streaming/routes.js
iindex 7c14469e..b8f33bbd 100644
---- a/server/streaming/routes.js
++++ b/server/streaming/routes.js
@@@ -1829,6 +1829,17 @@ router.get('/:id', optionalAuth, async (req, res) => {
});
// ── Start a New Stream (Go Live) ─────────────────────────────
++// One go-live per stream slot at a time. A second request while the first is still being created, or within 15 s of it
++// while that stream is live (a double tap, a retried request), answers with the same stream instead of opening a second
++// one. Two live streams for one slot split the room: on 2026-10-17 viewers' chat and TTS went to one stream while the
++// phone studio sat in the other, and the RobotStreamer relay, started on both, delivered every line twice.
++const GO_LIVE_REPEAT_MS = 15_000;
++const _goLiveInFlight = new Map(); // managed stream id → Promise<{ status, body }>
++function goLiveEndpoint(stream, managedStream) {
++ if (stream.protocol === 'jsmpeg') return jsmpegRelay.createChannel(managedStream.stream_key);
++ if (stream.protocol === 'webrtc') return { roomId: `stream-${stream.id}` };
++ return {};
++}
router.post('/', requireAuth, async (req, res) => {
try {
const managedStreamId = req.body.managed_stream_id ? parseInt(req.body.managed_stream_id) : null;
@@@ -1867,71 +1878,90 @@ router.post('/', requireAuth, async (req, res) => {
return res.status(400).json({ error: 'Invalid stream settings' });
}
-- const channel = await db.ensureChannel(req.user.id);
--
-- // Streamer role promotion deferred — applied on first real feed ingest
-- // (see whip-handler.js, webrtc-sfu producer, jsmpeg relay, rtmp handler)
--
-- // Use managed stream's settings as defaults, allow per-session overrides
-- const streamProtocol = protocol || cleanProtocol(managedStream.protocol) || cleanProtocol(channel.protocol) || 'webrtc';
-- const streamCategory = category || cleanText(managedStream.category, { maxLength: MAX_CATEGORY_LENGTH }) || cleanText(channel.category, { maxLength: MAX_CATEGORY_LENGTH }) || null; // null → the AI classifies the stream from what it shows
++ // The same slot is going live right now: wait for that request and answer with its stream.
++ if (_goLiveInFlight.has(managedStream.id)) {
++ const first = await _goLiveInFlight.get(managedStream.id);
++ return res.status(first.status === 201 ? 200 : first.status).json({ ...first.body, reused: true });
++ }
++ // It went live moments ago and is still live: the same go-live, asked twice.
++ const recent = await db.get('SELECT * FROM streams WHERE managed_stream_id = ? AND is_live = 1 ORDER BY id DESC LIMIT 1', [managedStream.id]);
++ const recentAt = recent && recent.started_at ? Date.parse(String(recent.started_at).replace(' ', 'T') + (/[zZ]|[+-]\d\d:?\d\d$/.test(String(recent.started_at)) ? '' : 'Z')) : NaN;
++ if (recent && recent.user_id === req.user.id && Date.now() - recentAt < GO_LIVE_REPEAT_MS) {
++ const stream = await db.getStreamById(recent.id);
++ return res.status(200).json({ stream, endpoint: goLiveEndpoint(stream, managedStream), reused: true });
++ }
++ const creating = createGoLive(req, managedStream, { title, description, category, protocol, tags, callMode });
++ _goLiveInFlight.set(managedStream.id, creating);
++ try {
++ const done = await creating;
++ res.status(done.status).json(done.body);
++ } finally {
++ _goLiveInFlight.delete(managedStream.id);
++ }
++ } catch (err) {
++ console.error('[Streams] Create error:', err.message);
++ res.status(500).json({ error: 'Failed to create stream' });
++ }
++});
-- const result = await db.createStream({
-- user_id: req.user.id,
-- channel_id: channel.id,
-- managed_stream_id: managedStream.id,
-- title: title || cleanText(managedStream.title, { maxLength: MAX_TITLE_LENGTH }) || `${req.user.display_name}'s Stream`,
-- description: description ?? cleanText(managedStream.description, { maxLength: MAX_DESCRIPTION_LENGTH, allowEmpty: true }) ?? '',
-- category: streamCategory,
-- protocol: streamProtocol,
-- is_nsfw: channel.force_nsfw ? 1 : (hasOwn(req.body, 'is_nsfw') ? cleanBooleanFlag(req.body.is_nsfw) : !!channel.is_nsfw),
-- });
++/** Create the stream row and start its integrations (POST /api/streams). → { status, body } */
++async function createGoLive(req, managedStream, { title, description, category, protocol, tags, callMode }) {
++ const channel = await db.ensureChannel(req.user.id);
++
++ // Streamer role promotion deferred — applied on first real feed ingest
++ // (see whip-handler.js, webrtc-sfu producer, jsmpeg relay, rtmp handler)
++
++ // Use managed stream's settings as defaults, allow per-session overrides
++ const streamProtocol = protocol || cleanProtocol(managedStream.protocol) || cleanProtocol(channel.protocol) || 'webrtc';
++ const streamCategory = category || cleanText(managedStream.category, { maxLength: MAX_CATEGORY_LENGTH }) || cleanText(channel.category, { maxLength: MAX_CATEGORY_LENGTH }) || null; // null → the AI classifies the stream from what it shows
++
++ const result = await db.createStream({
++ user_id: req.user.id,
++ channel_id: channel.id,
++ managed_stream_id: managedStream.id,
++ title: title || cleanText(managedStream.title, { maxLength: MAX_TITLE_LENGTH }) || `${req.user.display_name}'s Stream`,
++ description: description ?? cleanText(managedStream.description, { maxLength: MAX_DESCRIPTION_LENGTH, allowEmpty: true }) ?? '',
++ category: streamCategory,
++ protocol: streamProtocol,
++ is_nsfw: channel.force_nsfw ? 1 : (hasOwn(req.body, 'is_nsfw') ? cleanBooleanFlag(req.body.is_nsfw) : !!channel.is_nsfw),
++ });
-- const streamId = result.lastInsertRowid;
++ const streamId = result.lastInsertRowid;
-- // Initialize heartbeat
-- await db.run('UPDATE streams SET last_heartbeat = ov_now() WHERE id = ?', [streamId]);
++ // Initialize heartbeat
++ await db.run('UPDATE streams SET last_heartbeat = ov_now() WHERE id = ?', [streamId]);
-- if (tags && tags.length > 0) {
-- await db.run('UPDATE streams SET tags = ? WHERE id = ?', [JSON.stringify(tags), streamId]);
-- }
++ if (tags && tags.length > 0) {
++ await db.run('UPDATE streams SET tags = ? WHERE id = ?', [JSON.stringify(tags), streamId]);
++ }
-- // Set call mode if provided — create a stream voice channel
-- if (callMode) {
-- await db.run('UPDATE streams SET call_mode = ? WHERE id = ?', [callMode, streamId]);
-- await callsAuthority.createStreamChannel(streamId, callMode, req.user.id);
-- }
++ // Set call mode if provided — create a stream voice channel
++ if (callMode) {
++ await db.run('UPDATE streams SET call_mode = ? WHERE id = ?', [callMode, streamId]);
++ await callsAuthority.createStreamChannel(streamId, callMode, req.user.id);
++ }
-- let endpoint = {};
-- if (streamProtocol === 'jsmpeg') {
-- endpoint = jsmpegRelay.createChannel(managedStream.stream_key);
-- } else if (streamProtocol === 'webrtc') {
-- endpoint = { roomId: `stream-${streamId}` };
-- }
++ const endpoint = goLiveEndpoint({ id: streamId, protocol: streamProtocol }, managedStream);
-- await db.run(
-- `INSERT INTO cameras (stream_id, camera_index, label, protocol) VALUES (?, 0, 'Main', ?)`,
-- [streamId, streamProtocol]
-- );
++ await db.run(
++ `INSERT INTO cameras (stream_id, camera_index, label, protocol) VALUES (?, 0, 'Main', ?)`,
++ [streamId, streamProtocol]
++ );
-- const stream = await db.getStreamById(streamId);
-- robotStreamerService.startForStream(stream).catch((rsErr) => {
-- console.warn(`[RS] Failed to start integration for stream ${streamId}:`, rsErr.message);
-- });
-- chatRelayService.startForStream(stream).catch((relayErr) => {
-- console.warn(`[ChatRelay] Failed to start relay for stream ${streamId}:`, relayErr.message);
-- });
-- try { require('../integrations/ai-chatbot-service').startForStream(stream); } catch (aiErr) { console.warn('[AI-Bots] start failed:', aiErr.message); }
++ const stream = await db.getStreamById(streamId);
++ robotStreamerService.startForStream(stream).catch((rsErr) => {
++ console.warn(`[RS] Failed to start integration for stream ${streamId}:`, rsErr.message);
++ });
++ chatRelayService.startForStream(stream).catch((relayErr) => {
++ console.warn(`[ChatRelay] Failed to start relay for stream ${streamId}:`, relayErr.message);
++ });
++ try { require('../integrations/ai-chatbot-service').startForStream(stream); } catch (aiErr) { console.warn('[AI-Bots] start failed:', aiErr.message); }
-- // Cross-site on-screen "went live" notification (rate-limited per slot/hour).
-- try { require('./live-events').announceGoLive(stream, req.user); } catch { /* */ }
++ // Cross-site on-screen "went live" notification (rate-limited per slot/hour).
++ try { require('./live-events').announceGoLive(stream, req.user); } catch { /* */ }
-- res.status(201).json({ stream, endpoint });
-- } catch (err) {
-- console.error('[Streams] Create error:', err.message);
-- res.status(500).json({ error: 'Failed to create stream' });
-- }
--});
++ return { status: 201, body: { stream, endpoint } };
++}
// ── Update Stream Info ───────────────────────────────────────
router.put('/:id', requireAuth, async (req, res) => {
ddiff --git a/test/golive-single-stream.test.js b/test/golive-single-stream.test.js
nnew file mode 100644
iindex 00000000..803ba336
---- /dev/null
++++ b/test/golive-single-stream.test.js
@@@ -0,0 +1,69 @@
++'use strict';
++// One go-live per stream slot: two requests at once, or a repeat within 15 s while the stream is live, get the same
++// stream instead of a second one. On 2026-10-17 one go-live made streams 2620 and 2621 in the same second: viewers'
++// chat and TTS went to one while the phone studio sat in the other, and the RobotStreamer relay, started on both,
++// delivered every line twice.
++//
++// node test/golive-single-stream.test.js
++const assert = require('assert');
++const http = require('http');
++process.env.NODE_ENV = 'test';
++console.log = () => {};
++console.warn = () => {};
++
++(async () => {
++ const db = require('../server/db/database');
++ await db.initDb();
++ const raw = db.getDb();
++ // Requests carry their user in x-test-user (the auth middleware is replaced before the routes load).
++ const auth = require('../server/auth/auth');
++ const signIn = async (req) => { const u = await db.getUserById(Number(req.headers['x-test-user'] || 0)); if (u) req.user = u; return u; };
++ auth.requireAuth = async (req, res, next) => ((await signIn(req)) ? next() : res.status(401).json({ error: 'Authentication required' }));
++ auth.optionalAuth = async (req, res, next) => { await signIn(req); next(); };
++ // No outside integrations from a test: RobotStreamer, chat relays, AI viewers, go-live notices.
++ require('../server/integrations/robotstreamer-service').startForStream = async () => {};
++ require('../server/integrations/chat-relay-service').startForStream = async () => {};
++ require('../server/streaming/live-events').announceGoLive = () => {};
++
++ await raw.prepare(`INSERT INTO users (id, username, display_name, email, password_hash, role) OVERRIDING SYSTEM VALUE VALUES (1, 'gooze', 'gooze', 'g@x', 'x', 'streamer')`).run();
++ const channel = await db.ensureChannel(1);
++ const slot = await db.createManagedStream({ user_id: 1, channel_id: channel.id, title: 'Main', protocol: 'webrtc', stream_key: 'k'.repeat(32) });
++ const slotId = Number(slot.lastInsertRowid || slot.id);
++
++ const express = require('express');
++ const app = express();
++ app.use(express.json());
++ app.use('/api/streams', require('../server/streaming/routes'));
++ const server = http.createServer(app).listen(0, '127.0.0.1');
++ await new Promise((r) => server.once('listening', r));
++ const goLive = () => fetch(`http://127.0.0.1:${server.address().port}/api/streams`, {
++ method: 'POST', headers: { 'content-type': 'application/json', 'x-test-user': '1' }, body: JSON.stringify({ managed_stream_id: slotId, title: 'YO' }),
++ }).then(async (r) => ({ status: r.status, body: await r.json() }));
++ const liveRows = async () => (await raw.prepare('SELECT id FROM streams WHERE managed_stream_id = ? AND is_live = 1').all(slotId)).map((r) => Number(r.id));
++
++ try {
++ // Two at once (a double tap): one stream, both answers name it.
++ const [a, b] = await Promise.all([goLive(), goLive()]);
++ assert.deepStrictEqual([a.status, b.status].sort(), [200, 201], JSON.stringify([a, b]));
++ assert.strictEqual(a.body.stream.id, b.body.stream.id);
++ assert.deepStrictEqual(await liveRows(), [Number(a.body.stream.id)], 'one live stream for the slot');
++ assert.deepStrictEqual((a.status === 200 ? a : b).body.endpoint, { roomId: `stream-${a.body.stream.id}` }, 'the repeat gets the same room');
++
++ // Asked again moments later while it is live: the same stream.
++ const again = await goLive();
++ assert.deepStrictEqual([again.status, again.body.stream.id, again.body.reused], [200, a.body.stream.id, true]);
++
++ // Once that stream has ended, going live starts a new one.
++ await raw.prepare('UPDATE streams SET is_live = 0, ended_at = ov_now() WHERE id = ?').run(a.body.stream.id);
++ const next = await goLive();
++ assert.strictEqual(next.status, 201);
++ assert.notStrictEqual(next.body.stream.id, a.body.stream.id);
++ // And a live stream older than 15 s is not taken for a repeat (a restart after a crash gets its own).
++ await raw.prepare("UPDATE streams SET started_at = to_char(now() at time zone 'utc' - interval '60 seconds', 'YYYY-MM-DD HH24:MI:SS') WHERE id = ?").run(next.body.stream.id);
++ assert.strictEqual((await goLive()).status, 201);
++ } finally {
++ server.close();
++ }
++ process.stdout.write('golive-single-stream: all checks passed\n');
++ process.exit(0);
++})().catch((e) => { console.error(e); process.exit(1); });