openvibers/openvibe.live Commits
Restream control is slot-only: a stream on a slot auto-starts and resumes only that slot's destinations, and a legacy stream with no slot only the owner's unbound ones. _getDestinationsForStream merged every destination the account owned into a slotted stream, so a streamer with auto-start destinations on one slot (user 1: slot 1 of slots 1/60/85) would start them when going live on another. The display paths follow the same rule: GET /api/restream/destinations without a slot lists only unbound rows (?all=1 is the explicit whole-account listing), GET /api/restream/viewer-counts takes ?managed_stream_id or counts the slots live now, getExternalViewerCountsForUser(user, null) counts only unbound destinations, a manual start of an unbound destination refuses a slotted stream, and the broadcast restream panel loads and polls its own slot. test/restream-slot-scope.test.js covers auto-start and resume per slot, the slot-less case and the routes
Changes
@@@ -1092,10 +1092,10 @@ https://www.101soundboards.com/sounds/12345 150p 2</code></pre><thead><tr><th>Method</th><th>Path</th><th>Notes</th></tr></thead><tbody><tr><td>GET</td><td>/api/restream/presets</td><td>Platform + quality/encoder preset metadata</td></tr>--<tr><td>GET / POST</td><td>/api/restream/destinations</td><td>List / create a destination (<code>platform, stream_key, quality_preset, managed_stream_id?, chat_relay?</code>)</td></tr>++<tr><td>GET / POST</td><td>/api/restream/destinations</td><td>List / create a destination (<code>platform, stream_key, quality_preset, managed_stream_id?, chat_relay?</code>). Listing is per slot: <code>?managed_stream_id=N</code> lists that slot, no slot lists only unbound destinations, <code>?all=1</code> lists the whole account. A stream restreams only to its own slot’s destinations</td></tr><tr><td>PUT / DELETE</td><td>/api/restream/destinations/:id</td><td>Update / delete (stops any active relay)</td></tr><tr><td>POST</td><td>/api/restream/destinations/:id/start · /stop</td><td>Start / stop the FFmpeg relay to a destination</td></tr>--<tr><td>GET</td><td>/api/restream/status · /viewer-counts</td><td>Per-stream relay status / cached external viewer counts</td></tr>++<tr><td>GET</td><td>/api/restream/status · /viewer-counts</td><td>Per-stream relay status / cached external viewer counts (<code>?managed_stream_id=N</code> for one slot; default: the slots live now)</td></tr><tr><td>GET</td><td>/api/restream/oauth/:platform/start · /status</td><td>Link a platform account to auto-fill ingest/key (Twitch/YouTube; Kick links identity only)</td></tr></tbody></table>@@@ -3112,15 +3112,22 @@ const RESTREAM_PLATFORM_META = {let _restreamDestinations = [];let _restreamStatusPollInterval = null;++/** The slot the restream panel is showing: the active stream's, else the one picked in the form. */++function _restreamPanelSlotId() {++ const activeSS = getActiveStreamState();++ const selectedManagedStream = document.getElementById('bc-managed-stream')?.value;++ return activeSS?.streamData?.managed_stream_id || (selectedManagedStream ? parseInt(selectedManagedStream, 10) : null);++}++/** Load restream destinations from server */async function loadRestreamDestinations() {try {-- const activeSS = getActiveStreamState();-- const selectedManagedStream = document.getElementById('bc-managed-stream')?.value;-- const managedStreamId = activeSS?.streamData?.managed_stream_id || (selectedManagedStream ? parseInt(selectedManagedStream, 10) : null);++ // Slot-only, like the server's restream control: a slot lists its own destinations;++ // no slot lists only the unbound (legacy) ones — never the whole account's.++ const managedStreamId = _restreamPanelSlotId();const url = managedStreamId ? `/restream/destinations?managed_stream_id=${managedStreamId}` : '/restream/destinations';const data = await api(url);-- _restreamDestinations = data.destinations || [];++ _restreamDestinations = (data.destinations || []).filter(d => (d.managed_stream_id || null) === (managedStreamId || null));} catch (err) {console.warn('[Restream] Failed to load destinations:', err.message);}@@@ -4644,9 +4651,11 @@ function _stopRestreamViewerPoll() {async function _pollRestreamViewerCounts() {if (!_restreamDestinations || _restreamDestinations.length === 0) return;-- // Fetch all platform viewer counts from server (Kick + Twitch + others)++ // Platform viewer counts (Kick + Twitch + others) for the slot this panel shows; with no slot,++ // the server counts the slots that are live now.try {-- const data = await api('/restream/viewer-counts');++ const slotId = _restreamPanelSlotId();++ const data = await api(slotId ? `/restream/viewer-counts?managed_stream_id=${slotId}` : '/restream/viewer-counts');_platformViewerCount = data?.total || 0;_platformViewerBreakdown = data?.breakdown || [];} catch {@@@ -4470,6 +4470,19 @@ function getRestreamDestinationsByManagedStream(managedStreamId) {return all('SELECT * FROM restream_destinations WHERE managed_stream_id = ? ORDER BY created_at', [managedStreamId]);}++/**++ * The destinations a stream on this slot may use, and nothing else: a slot gets only the++ * destinations bound to it; no slot (a legacy session) gets only the owner's unbound rows.++ * There is no fallback to every destination the account owns: that fallback started one++ * slot's auto-start destinations when the streamer went live on another slot.++ */++function getRestreamDestinationsForSlot(userId, managedStreamId) {++ if (managedStreamId) {++ return all('SELECT * FROM restream_destinations WHERE user_id = ? AND managed_stream_id = ? ORDER BY created_at', [userId, managedStreamId]);++ }++ return all('SELECT * FROM restream_destinations WHERE user_id = ? AND managed_stream_id IS NULL ORDER BY created_at', [userId]);++}++// ── Platform OAuth connection helpers ────────────────────────// ── Per-streamer channel points ("OpenCoins") ──@@@ -7518,7 +7531,7 @@ module.exports = {upsertSubscription, getSubscriptionByProviderRef, getActiveSubscription, isActiveSubscriber,getSubscriptionsByStreamer, getSubscriptionsBySubscriber, getActiveSubscriberCount, setSubscriptionStatus,getSubscriptionsDueRenewal,-- getRestreamDestinationsByManagedStream,++ getRestreamDestinationsByManagedStream, getRestreamDestinationsForSlot,// ChatsaveChatMessage, searchChatMessages, getUserChatHistory, getChatSamplesInChannel,// Chat AI summaries@@@ -1205,23 +1205,19 @@ class RestreamManager extends EventEmitter {}/**-- * Get total external viewer count for a user across all active restream destinations.-- * Returns { total, breakdown: [{ platform, name, count }] }-- * count is null when the platform is live but viewer count is unavailable.-- */-- /**-- * External (Twitch/Kick/YouTube) viewer counts for a user. When managedStreamId is-- * provided, only destinations bound to that stream slot are counted — so a channel-- * running multiple slots doesn't leak one slot's platform viewers onto another.++ * External (Twitch/Kick/YouTube) viewer counts for one of a user's stream slots:++ * { total, breakdown: [{ platform, name, count, destId }] }; count is null when the++ * platform is live but its viewer count is unavailable. Slot-only, like control: a slot++ * counts only its own destinations and no slot (a legacy session) counts only the++ * unbound ones, so one slot's platform viewers never show on another.*/getExternalViewerCountsForUser(userId, managedStreamId = null) {const db = require('../db/database');-- const dests = db.getRestreamDestinationsByUserId(userId) || [];++ const dests = db.getRestreamDestinationsForSlot(userId, managedStreamId || null) || [];const breakdown = [];let total = 0;for (const d of dests) {if (!d.enabled) continue;-- if (managedStreamId != null && d.managed_stream_id !== managedStreamId) continue;const count = this.getCachedViewerCount(d.id);const platformLive = this.isPlatformLive(d.id);if (count != null && count > 0) {@@@ -1813,21 +1809,16 @@ class RestreamManager extends EventEmitter {return destination;}++ /**++ * Slot-only: a stream on a slot restreams to that slot's destinations; a legacy stream with++ * no slot restreams only to the owner's unbound destinations. Merging in every destination++ * the account owns meant going live on slot B auto-started slot A's destinations.++ */_getDestinationsForStream(streamId, userId) {const db = require('../db/database');const stream = db.getStreamById(streamId);if (!stream) return [];---- const globalDests = db.getRestreamDestinationsByUserId(userId) || [];-- const slotDests = stream.managed_stream_id-- ? db.getRestreamDestinationsByManagedStream(stream.managed_stream_id) || []-- : [];---- // Preserve user-global destinations while allowing slot-specific overrides.-- const merged = new Map();-- for (const dest of globalDests) merged.set(dest.id, dest);-- for (const dest of slotDests) merged.set(dest.id, dest);-- return Array.from(merged.values());++ return db.getRestreamDestinationsForSlot(stream.user_id || userId, stream.managed_stream_id || null) || [];}async autoStartForStream(streamId, userId, streamInfo) {@@@ -106,7 +106,8 @@ function getLiveStreamForDestination(dest, userId, requestedStreamId) {return liveStreams.find(s => s.managed_stream_id === dest.managed_stream_id) || null;}if (requestedStreamId) {-- return liveStreams.find(s => s.id === parseInt(requestedStreamId, 10)) || null;++ // Slot-only: an unbound destination may only feed a stream that has no slot either.++ return liveStreams.find(s => s.id === parseInt(requestedStreamId, 10) && !s.managed_stream_id) || null;}// Destination has no managed_stream_id (legacy unbound row — DB backfill not yet run).// Return null so start/stop returns a clear error rather than silently picking any stream.@@@ -170,7 +171,9 @@ router.get('/presets', requireAuth, (req, res) => {});// ── GET /destinations — list user's restream destinations ────--// ?managed_stream_id=N — filter by stream slot++// ?managed_stream_id=N — that stream slot's destinations only.++// No slot — only the unbound (legacy, slot-less) destinations: the same set a stream with no++// slot would restream to. ?all=1 lists every destination the account owns, for management.router.get('/destinations', requireAuth, (req, res) => {try {const managedStreamId = req.query.managed_stream_id ? parseInt(req.query.managed_stream_id) : null;@@@ -181,9 +184,11 @@ router.get('/destinations', requireAuth, (req, res) => {if (!ms || ms.user_id !== req.user.id) {return res.status(403).json({ error: 'Not your stream slot' });}-- dests = db.getRestreamDestinationsByManagedStream(managedStreamId) || [];-- } else {++ dests = db.getRestreamDestinationsForSlot(req.user.id, managedStreamId) || [];++ } else if (req.query.all === '1' || req.query.all === 'true') {dests = db.getRestreamDestinationsByUserId(req.user.id) || [];++ } else {++ dests = db.getRestreamDestinationsForSlot(req.user.id, null) || [];}res.json({ destinations: dests.map(sanitizeDest) });} catch (err) {@@@ -499,10 +504,27 @@ router.post('/viewer-counts', requireAuth, (req, res) => {}});--// ── GET /viewer-counts — get all cached platform viewer counts for the broadcaster++// ── GET /viewer-counts — cached platform viewer counts for the broadcaster++// ?managed_stream_id=N — that slot's destinations only. Without it: the slots the broadcaster++// is live on right now (each live stream counts its own slot; a slot-less stream counts the++// unbound destinations), never every destination the account owns.router.get('/viewer-counts', requireAuth, (req, res) => {try {-- const ext = restreamManager.getExternalViewerCountsForUser(req.user.id);++ const managedStreamId = req.query.managed_stream_id ? parseInt(req.query.managed_stream_id, 10) : null;++ let slots;++ if (managedStreamId) {++ const ms = db.getManagedStreamById(managedStreamId);++ if (!ms || ms.user_id !== req.user.id) return res.status(403).json({ error: 'Not your stream slot' });++ slots = [managedStreamId];++ } else {++ slots = [...new Set((db.getLiveStreamsByUserId(req.user.id) || []).map((s) => s.managed_stream_id || null))];++ }++ const ext = { total: 0, breakdown: [] };++ for (const slotId of slots) {++ const part = restreamManager.getExternalViewerCountsForUser(req.user.id, slotId);++ ext.total += part.total;++ ext.breakdown.push(...part.breakdown);++ }res.json({ total: ext.total, breakdown: ext.breakdown });} catch (err) {res.status(500).json({ error: 'Failed to get viewer counts' });nnew file mode 100644@@@ -0,0 +1,171 @@++/**++ * Restream control is slot-only: going live on a slot auto-starts and resumes that slot's++ * destinations and nothing else; a legacy stream with no slot uses only unbound destinations.++ *++ * The production case: user 1 has slots 1, 60 and 85 with auto-start destinations on slot 1.++ * The old _getDestinationsForStream merged every destination the account owned into a slotted++ * stream, so going live on slot 60 or 85 started slot 1's Twitch and YouTube relays too.++ * The display paths (destination list, broadcaster viewer counts) follow the same rule.++ *++ * node test/restream-slot-scope.test.js++ */++'use strict';++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-restream-slot-'));++process.env.DB_PATH = path.join(tmp, 'live.db');++process.env.NODE_ENV = 'test';++const quiet = console.log;++console.log = () => {};++console.warn = () => {};++++const db = require('../server/db/database');++db.initDb();++const raw = db.getDb();++++const auth = require('../server/auth/auth');++const signIn = (req) => {++ const id = Number(req.headers['x-test-user'] || 0);++ const u = id ? db.getUserById(id) : null;++ if (u) { req.user = u; req.authSource = 'network'; }++ return u;++};++auth.requireAuth = (req, res, next) => (signIn(req) ? next() : res.status(401).json({ error: 'Authentication required' }));++auth.optionalAuth = (req, res, next) => { signIn(req); next(); };++++raw.prepare(`INSERT INTO users (id, username, display_name, email, password_hash, role) VALUES (1, 'alex', 'alex', 'a@x', 'x', 'streamer')`).run();++raw.prepare(`INSERT INTO users (id, username, display_name, email, password_hash, role) VALUES (2, 'other', 'other', 'o@x', 'x', 'streamer')`).run();++db.ensureChannel(1);++db.ensureChannel(2);++const chan = db.getChannelByUserId(1);++++const slot = (slug, userId = 1) => Number(db.createManagedStream({ user_id: userId, channel_id: chan.id, slug, title: slug, stream_key: `key-${slug}` }).lastInsertRowid);++const slotA = slot('main'); // "slot 1": the auto-start destinations live here++const slotB = slot('second'); // "slot 60"++const slotC = slot('third'); // "slot 85": no destinations at all++const otherSlot = slot('theirs', 2);++++const dest = (fields) => db.createRestreamDestination(fields.user_id || 1, {++ server_url: 'rtmp://ingest.example/live', stream_key: `sk-${fields.name}`, enabled: 1, ...fields,++}).id;++const aTwitch = dest({ name: 'a-twitch', platform: 'twitch', managed_stream_id: slotA, auto_start: 1 });++const aYoutube = dest({ name: 'a-youtube', platform: 'youtube', managed_stream_id: slotA, auto_start: 1 });++const bKick = dest({ name: 'b-kick', platform: 'kick', managed_stream_id: slotB, auto_start: 1 });++const bManual = dest({ name: 'b-manual', platform: 'custom', managed_stream_id: slotB, auto_start: 0 });++const unbound = dest({ name: 'legacy', platform: 'custom', auto_start: 1 });++const unboundManual = dest({ name: 'legacy-manual', platform: 'custom', auto_start: 0 });++dest({ name: 'theirs', platform: 'twitch', managed_stream_id: otherSlot, auto_start: 1, user_id: 2 });++++const manager = require('../server/streaming/restream-manager');++let started = [];++manager.startRestream = async (streamId, d) => { started.push([streamId, d.id]); return { status: 'starting' }; };++++const live = (managedStreamId, userId = 1) => Number(db.createStream({ user_id: userId, channel_id: chan.id, managed_stream_id: managedStreamId, title: 't', protocol: 'rtmp' }).lastInsertRowid);++const ids = (list) => list.map(([, destId]) => destId).sort((a, b) => a - b);++++let server;++(async () => {++ // ── Control: auto-start ──────────────────────────────────────────────────────────++ const onB = live(slotB);++ started = [];++ await manager.autoStartForStream(onB, 1, { protocol: 'rtmp' });++ assert.deepStrictEqual(ids(started), [bKick], 'going live on slot B auto-starts only slot B auto-start destinations (not slot A, not unbound)');++++ const onC = live(slotC);++ started = [];++ await manager.autoStartForStream(onC, 1, { protocol: 'rtmp' });++ assert.deepStrictEqual(started, [], 'a slot with no destinations starts nothing, even though slot A has auto-start ones');++++ const onA = live(slotA);++ started = [];++ await manager.autoStartForStream(onA, 1, { protocol: 'rtmp' });++ assert.deepStrictEqual(ids(started), [aTwitch, aYoutube].sort((a, b) => a - b), 'slot A starts its own two');++++ const legacy = live(null);++ started = [];++ await manager.autoStartForStream(legacy, 1, { protocol: 'rtmp' });++ assert.deepStrictEqual(ids(started), [unbound], 'a legacy stream with no slot starts only unbound auto-start destinations');++++ // ── Control: resume (every enabled destination, auto-start or not) ──────────────++ started = [];++ await manager.resumeForStream(onB, 1, { protocol: 'rtmp' });++ assert.deepStrictEqual(ids(started), [bKick, bManual].sort((a, b) => a - b), 'resume on slot B covers only slot B');++ started = [];++ await manager.resumeForStream(onC, 1, { protocol: 'rtmp' });++ assert.deepStrictEqual(started, [], 'resume on an empty slot resumes nothing');++ started = [];++ await manager.resumeForStream(legacy, 1, { protocol: 'rtmp' });++ assert.deepStrictEqual(ids(started), [unbound, unboundManual].sort((a, b) => a - b), 'resume on a slot-less stream covers only unbound destinations');++++ // The owner of the stream decides, not the caller's userId argument.++ started = [];++ await manager.autoStartForStream(onB, 2, { protocol: 'rtmp' });++ assert.deepStrictEqual(ids(started), [bKick], 'a wrong userId cannot pull in another account\'s destinations');++++ // ── Display: external viewer counts ──────────────────────────────────────────────++ manager.setViewerCount(aTwitch, 40, true);++ manager.setViewerCount(bKick, 7, true);++ manager.setViewerCount(unbound, 3, true);++ assert.strictEqual(manager.getExternalViewerCountsForUser(1, slotB).total, 7, 'slot B counts only its own platform viewers');++ assert.strictEqual(manager.getExternalViewerCountsForUser(1, slotA).total, 40);++ assert.strictEqual(manager.getExternalViewerCountsForUser(1, null).total, 3, 'no slot counts only unbound destinations');++++ // ── Display: routes ─────────────────────────────────────────────────────────────++ const express = require('express');++ const app = express();++ app.use(express.json());++ app.use('/api/restream', require('../server/streaming/restream-routes'));++ server = http.createServer(app).listen(0);++ const call = (method, p, user, body) => new Promise((resolve, reject) => {++ const data = body ? JSON.stringify(body) : null;++ const req = http.request({ port: server.address().port, path: p, method, headers: { 'content-type': 'application/json', 'x-test-user': String(user) } }, (res) => {++ let text = '';++ res.on('data', (c) => { text += c; });++ res.on('end', () => { let json = null; try { json = JSON.parse(text); } catch { /* */ } resolve({ status: res.statusCode, json }); });++ });++ req.on('error', reject);++ if (data) req.write(data);++ req.end();++ });++ const listed = (r) => r.json.destinations.map((d) => d.id).sort((a, b) => a - b);++++ let r = await call('GET', `/api/restream/destinations?managed_stream_id=${slotB}`, 1);++ assert.strictEqual(r.status, 200);++ assert.deepStrictEqual(listed(r), [bKick, bManual].sort((a, b) => a - b), 'slot listing is that slot only');++ r = await call('GET', '/api/restream/destinations', 1);++ assert.deepStrictEqual(listed(r), [unbound, unboundManual].sort((a, b) => a - b), 'no slot lists only unbound destinations, not the whole account');++ r = await call('GET', '/api/restream/destinations?all=1', 1);++ assert.strictEqual(r.json.destinations.length, 6, '?all=1 is the explicit whole-account listing');++ assert.ok(r.json.destinations.every((d) => d.user_id === 1), 'and still only the caller\'s own');++ r = await call('GET', `/api/restream/destinations?managed_stream_id=${otherSlot}`, 1);++ assert.strictEqual(r.status, 403, 'another account\'s slot is refused');++++ // Viewer counts: explicit slot, or the slots live right now (A, B, C and slot-less).++ r = await call('GET', `/api/restream/viewer-counts?managed_stream_id=${slotB}`, 1);++ assert.strictEqual(r.json.total, 7);++ r = await call('GET', '/api/restream/viewer-counts', 1);++ assert.strictEqual(r.json.total, 50, 'the default counts each live stream\'s own slot once');++ db.endStream(onA);++ r = await call('GET', '/api/restream/viewer-counts', 1);++ assert.strictEqual(r.json.total, 10, 'slot A is not live any more, so its Twitch viewers are not counted');++ r = await call('GET', `/api/restream/viewer-counts?managed_stream_id=${otherSlot}`, 1);++ assert.strictEqual(r.status, 403);++++ // Manual start of an unbound destination refuses a slotted stream.++ r = await call('POST', `/api/restream/destinations/${unbound}/start`, 1, { streamId: onB });++ assert.strictEqual(r.status, 400, 'an unbound destination cannot be started on a stream that has a slot');++++ server.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ quiet('restream-slot-scope: ok');++ process.exit(0);++})().catch((err) => {++ console.error = quiet;++ quiet(err);++ try { server && server.close(); } catch { /* */ }++ process.exit(1);++});
@@ -1092,10 +1092,10 @@ https://www.101soundboards.com/sounds/12345 150p 2</code></pre>
@@ -1092,10 +1092,10 @@ https://www.101soundboards.com/sounds/12345 150p 2</code></pre>
<thead><tr><th>Method</th><th>Path</th><th>Notes</th></tr></thead>
<thead><tr><th>Method</th><th>Path</th><th>Notes</th></tr></thead>
<tbody>
<tbody>
<tr><td>GET</td><td>/api/restream/presets</td><td>Platform + quality/encoder preset metadata</td></tr>
<tr><td>GET</td><td>/api/restream/presets</td><td>Platform + quality/encoder preset metadata</td></tr>
-<tr><td>GET / POST</td><td>/api/restream/destinations</td><td>List / create a destination (<code>platform, stream_key, quality_preset, managed_stream_id?, chat_relay?</code>)</td></tr>
+<tr><td>GET / POST</td><td>/api/restream/destinations</td><td>List / create a destination (<code>platform, stream_key, quality_preset, managed_stream_id?, chat_relay?</code>). Listing is per slot: <code>?managed_stream_id=N</code> lists that slot, no slot lists only unbound destinations, <code>?all=1</code> lists the whole account. A stream restreams only to its own slot’s destinations</td></tr>
<tr><td>PUT / DELETE</td><td>/api/restream/destinations/:id</td><td>Update / delete (stops any active relay)</td></tr>
<tr><td>PUT / DELETE</td><td>/api/restream/destinations/:id</td><td>Update / delete (stops any active relay)</td></tr>
<tr><td>POST</td><td>/api/restream/destinations/:id/start · /stop</td><td>Start / stop the FFmpeg relay to a destination</td></tr>
<tr><td>POST</td><td>/api/restream/destinations/:id/start · /stop</td><td>Start / stop the FFmpeg relay to a destination</td></tr>
-<tr><td>GET</td><td>/api/restream/status · /viewer-counts</td><td>Per-stream relay status / cached external viewer counts</td></tr>
+<tr><td>GET</td><td>/api/restream/status · /viewer-counts</td><td>Per-stream relay status / cached external viewer counts (<code>?managed_stream_id=N</code> for one slot; default: the slots live now)</td></tr>
<tr><td>GET</td><td>/api/restream/oauth/:platform/start · /status</td><td>Link a platform account to auto-fill ingest/key (Twitch/YouTube; Kick links identity only)</td></tr>
<tr><td>GET</td><td>/api/restream/oauth/:platform/start · /status</td><td>Link a platform account to auto-fill ingest/key (Twitch/YouTube; Kick links identity only)</td></tr>
</tbody>
</tbody>
</table>
</table>
@@ -3112,15 +3112,22 @@ const RESTREAM_PLATFORM_META = {
@@ -3112,15 +3112,22 @@ const RESTREAM_PLATFORM_META = {
let _restreamDestinations = [];
let _restreamDestinations = [];
let _restreamStatusPollInterval = null;
let _restreamStatusPollInterval = null;
+/** The slot the restream panel is showing: the active stream's, else the one picked in the form. */
+function _restreamPanelSlotId() {
+ const activeSS = getActiveStreamState();
+ const selectedManagedStream = document.getElementById('bc-managed-stream')?.value;
+ return activeSS?.streamData?.managed_stream_id || (selectedManagedStream ? parseInt(selectedManagedStream, 10) : null);
+}
+
/** Load restream destinations from server */
/** Load restream destinations from server */
async function loadRestreamDestinations() {
async function loadRestreamDestinations() {
try {
try {
- const activeSS = getActiveStreamState();
- const selectedManagedStream = document.getElementById('bc-managed-stream')?.value;
- const managedStreamId = activeSS?.streamData?.managed_stream_id || (selectedManagedStream ? parseInt(selectedManagedStream, 10) : null);
+ // Slot-only, like the server's restream control: a slot lists its own destinations;
+ // no slot lists only the unbound (legacy) ones — never the whole account's.
+ const managedStreamId = _restreamPanelSlotId();
const url = managedStreamId ? `/restream/destinations?managed_stream_id=${managedStreamId}` : '/restream/destinations';
const url = managedStreamId ? `/restream/destinations?managed_stream_id=${managedStreamId}` : '/restream/destinations';
const data = await api(url);
const data = await api(url);
- _restreamDestinations = data.destinations || [];
+ _restreamDestinations = (data.destinations || []).filter(d => (d.managed_stream_id || null) === (managedStreamId || null));
} catch (err) {
} catch (err) {
console.warn('[Restream] Failed to load destinations:', err.message);
console.warn('[Restream] Failed to load destinations:', err.message);
}
}
@@ -4644,9 +4651,11 @@ function _stopRestreamViewerPoll() {
@@ -4644,9 +4651,11 @@ function _stopRestreamViewerPoll() {
async function _pollRestreamViewerCounts() {
async function _pollRestreamViewerCounts() {
if (!_restreamDestinations || _restreamDestinations.length === 0) return;
if (!_restreamDestinations || _restreamDestinations.length === 0) return;
- // Fetch all platform viewer counts from server (Kick + Twitch + others)
+ // Platform viewer counts (Kick + Twitch + others) for the slot this panel shows; with no slot,
+ // the server counts the slots that are live now.
try {
try {
- const data = await api('/restream/viewer-counts');
+ const slotId = _restreamPanelSlotId();
+ const data = await api(slotId ? `/restream/viewer-counts?managed_stream_id=${slotId}` : '/restream/viewer-counts');
_platformViewerCount = data?.total || 0;
_platformViewerCount = data?.total || 0;
_platformViewerBreakdown = data?.breakdown || [];
_platformViewerBreakdown = data?.breakdown || [];
} catch {
} catch {
@@ -4470,6 +4470,19 @@ function getRestreamDestinationsByManagedStream(managedStreamId) {
@@ -4470,6 +4470,19 @@ function getRestreamDestinationsByManagedStream(managedStreamId) {
return all('SELECT * FROM restream_destinations WHERE managed_stream_id = ? ORDER BY created_at', [managedStreamId]);
return all('SELECT * FROM restream_destinations WHERE managed_stream_id = ? ORDER BY created_at', [managedStreamId]);
}
}
+/**
+ * The destinations a stream on this slot may use, and nothing else: a slot gets only the
+ * destinations bound to it; no slot (a legacy session) gets only the owner's unbound rows.
+ * There is no fallback to every destination the account owns: that fallback started one
+ * slot's auto-start destinations when the streamer went live on another slot.
+ */
+function getRestreamDestinationsForSlot(userId, managedStreamId) {
+ if (managedStreamId) {
+ return all('SELECT * FROM restream_destinations WHERE user_id = ? AND managed_stream_id = ? ORDER BY created_at', [userId, managedStreamId]);
+ }
+ return all('SELECT * FROM restream_destinations WHERE user_id = ? AND managed_stream_id IS NULL ORDER BY created_at', [userId]);
+}
+
// ── Platform OAuth connection helpers ────────────────────────
// ── Platform OAuth connection helpers ────────────────────────
// ── Per-streamer channel points ("OpenCoins") ──
// ── Per-streamer channel points ("OpenCoins") ──
@@ -7518,7 +7531,7 @@ module.exports = {
@@ -7518,7 +7531,7 @@ module.exports = {
upsertSubscription, getSubscriptionByProviderRef, getActiveSubscription, isActiveSubscriber,
upsertSubscription, getSubscriptionByProviderRef, getActiveSubscription, isActiveSubscriber,
getSubscriptionsByStreamer, getSubscriptionsBySubscriber, getActiveSubscriberCount, setSubscriptionStatus,
getSubscriptionsByStreamer, getSubscriptionsBySubscriber, getActiveSubscriberCount, setSubscriptionStatus,
getSubscriptionsDueRenewal,
getSubscriptionsDueRenewal,
- getRestreamDestinationsByManagedStream,
+ getRestreamDestinationsByManagedStream, getRestreamDestinationsForSlot,
// Chat
// Chat
saveChatMessage, searchChatMessages, getUserChatHistory, getChatSamplesInChannel,
saveChatMessage, searchChatMessages, getUserChatHistory, getChatSamplesInChannel,
// Chat AI summaries
// Chat AI summaries
@@ -1205,23 +1205,19 @@ class RestreamManager extends EventEmitter {
@@ -1205,23 +1205,19 @@ class RestreamManager extends EventEmitter {
}
}
/**
/**
- * Get total external viewer count for a user across all active restream destinations.
- * Returns { total, breakdown: [{ platform, name, count }] }
- * count is null when the platform is live but viewer count is unavailable.
- */
- /**
- * External (Twitch/Kick/YouTube) viewer counts for a user. When managedStreamId is
- * provided, only destinations bound to that stream slot are counted — so a channel
- * running multiple slots doesn't leak one slot's platform viewers onto another.
+ * External (Twitch/Kick/YouTube) viewer counts for one of a user's stream slots:
+ * { total, breakdown: [{ platform, name, count, destId }] }; count is null when the
+ * platform is live but its viewer count is unavailable. Slot-only, like control: a slot
+ * counts only its own destinations and no slot (a legacy session) counts only the
+ * unbound ones, so one slot's platform viewers never show on another.
*/
*/
getExternalViewerCountsForUser(userId, managedStreamId = null) {
getExternalViewerCountsForUser(userId, managedStreamId = null) {
const db = require('../db/database');
const db = require('../db/database');
- const dests = db.getRestreamDestinationsByUserId(userId) || [];
+ const dests = db.getRestreamDestinationsForSlot(userId, managedStreamId || null) || [];
const breakdown = [];
const breakdown = [];
let total = 0;
let total = 0;
for (const d of dests) {
for (const d of dests) {
if (!d.enabled) continue;
if (!d.enabled) continue;
- if (managedStreamId != null && d.managed_stream_id !== managedStreamId) continue;
const count = this.getCachedViewerCount(d.id);
const count = this.getCachedViewerCount(d.id);
const platformLive = this.isPlatformLive(d.id);
const platformLive = this.isPlatformLive(d.id);
if (count != null && count > 0) {
if (count != null && count > 0) {
@@ -1813,21 +1809,16 @@ class RestreamManager extends EventEmitter {
@@ -1813,21 +1809,16 @@ class RestreamManager extends EventEmitter {
return destination;
return destination;
}
}
+ /**
+ * Slot-only: a stream on a slot restreams to that slot's destinations; a legacy stream with
+ * no slot restreams only to the owner's unbound destinations. Merging in every destination
+ * the account owns meant going live on slot B auto-started slot A's destinations.
+ */
_getDestinationsForStream(streamId, userId) {
_getDestinationsForStream(streamId, userId) {
const db = require('../db/database');
const db = require('../db/database');
const stream = db.getStreamById(streamId);
const stream = db.getStreamById(streamId);
if (!stream) return [];
if (!stream) return [];
-
- const globalDests = db.getRestreamDestinationsByUserId(userId) || [];
- const slotDests = stream.managed_stream_id
- ? db.getRestreamDestinationsByManagedStream(stream.managed_stream_id) || []
- : [];
-
- // Preserve user-global destinations while allowing slot-specific overrides.
- const merged = new Map();
- for (const dest of globalDests) merged.set(dest.id, dest);
- for (const dest of slotDests) merged.set(dest.id, dest);
- return Array.from(merged.values());
+ return db.getRestreamDestinationsForSlot(stream.user_id || userId, stream.managed_stream_id || null) || [];
}
}
async autoStartForStream(streamId, userId, streamInfo) {
async autoStartForStream(streamId, userId, streamInfo) {
@@ -106,7 +106,8 @@ function getLiveStreamForDestination(dest, userId, requestedStreamId) {
@@ -106,7 +106,8 @@ function getLiveStreamForDestination(dest, userId, requestedStreamId) {
return liveStreams.find(s => s.managed_stream_id === dest.managed_stream_id) || null;
return liveStreams.find(s => s.managed_stream_id === dest.managed_stream_id) || null;
}
}
if (requestedStreamId) {
if (requestedStreamId) {
- return liveStreams.find(s => s.id === parseInt(requestedStreamId, 10)) || null;
+ // Slot-only: an unbound destination may only feed a stream that has no slot either.
+ return liveStreams.find(s => s.id === parseInt(requestedStreamId, 10) && !s.managed_stream_id) || null;
}
}
// Destination has no managed_stream_id (legacy unbound row — DB backfill not yet run).
// Destination has no managed_stream_id (legacy unbound row — DB backfill not yet run).
// Return null so start/stop returns a clear error rather than silently picking any stream.
// Return null so start/stop returns a clear error rather than silently picking any stream.
@@ -170,7 +171,9 @@ router.get('/presets', requireAuth, (req, res) => {
@@ -170,7 +171,9 @@ router.get('/presets', requireAuth, (req, res) => {
});
});
// ── GET /destinations — list user's restream destinations ────
// ── GET /destinations — list user's restream destinations ────
-// ?managed_stream_id=N — filter by stream slot
+// ?managed_stream_id=N — that stream slot's destinations only.
+// No slot — only the unbound (legacy, slot-less) destinations: the same set a stream with no
+// slot would restream to. ?all=1 lists every destination the account owns, for management.
router.get('/destinations', requireAuth, (req, res) => {
router.get('/destinations', requireAuth, (req, res) => {
try {
try {
const managedStreamId = req.query.managed_stream_id ? parseInt(req.query.managed_stream_id) : null;
const managedStreamId = req.query.managed_stream_id ? parseInt(req.query.managed_stream_id) : null;
@@ -181,9 +184,11 @@ router.get('/destinations', requireAuth, (req, res) => {
@@ -181,9 +184,11 @@ router.get('/destinations', requireAuth, (req, res) => {
if (!ms || ms.user_id !== req.user.id) {
if (!ms || ms.user_id !== req.user.id) {
return res.status(403).json({ error: 'Not your stream slot' });
return res.status(403).json({ error: 'Not your stream slot' });
}
}
- dests = db.getRestreamDestinationsByManagedStream(managedStreamId) || [];
- } else {
+ dests = db.getRestreamDestinationsForSlot(req.user.id, managedStreamId) || [];
+ } else if (req.query.all === '1' || req.query.all === 'true') {
dests = db.getRestreamDestinationsByUserId(req.user.id) || [];
dests = db.getRestreamDestinationsByUserId(req.user.id) || [];
+ } else {
+ dests = db.getRestreamDestinationsForSlot(req.user.id, null) || [];
}
}
res.json({ destinations: dests.map(sanitizeDest) });
res.json({ destinations: dests.map(sanitizeDest) });
} catch (err) {
} catch (err) {
@@ -499,10 +504,27 @@ router.post('/viewer-counts', requireAuth, (req, res) => {
@@ -499,10 +504,27 @@ router.post('/viewer-counts', requireAuth, (req, res) => {
}
}
});
});
-// ── GET /viewer-counts — get all cached platform viewer counts for the broadcaster
+// ── GET /viewer-counts — cached platform viewer counts for the broadcaster
+// ?managed_stream_id=N — that slot's destinations only. Without it: the slots the broadcaster
+// is live on right now (each live stream counts its own slot; a slot-less stream counts the
+// unbound destinations), never every destination the account owns.
router.get('/viewer-counts', requireAuth, (req, res) => {
router.get('/viewer-counts', requireAuth, (req, res) => {
try {
try {
- const ext = restreamManager.getExternalViewerCountsForUser(req.user.id);
+ const managedStreamId = req.query.managed_stream_id ? parseInt(req.query.managed_stream_id, 10) : null;
+ let slots;
+ if (managedStreamId) {
+ const ms = db.getManagedStreamById(managedStreamId);
+ if (!ms || ms.user_id !== req.user.id) return res.status(403).json({ error: 'Not your stream slot' });
+ slots = [managedStreamId];
+ } else {
+ slots = [...new Set((db.getLiveStreamsByUserId(req.user.id) || []).map((s) => s.managed_stream_id || null))];
+ }
+ const ext = { total: 0, breakdown: [] };
+ for (const slotId of slots) {
+ const part = restreamManager.getExternalViewerCountsForUser(req.user.id, slotId);
+ ext.total += part.total;
+ ext.breakdown.push(...part.breakdown);
+ }
res.json({ total: ext.total, breakdown: ext.breakdown });
res.json({ total: ext.total, breakdown: ext.breakdown });
} catch (err) {
} catch (err) {
res.status(500).json({ error: 'Failed to get viewer counts' });
res.status(500).json({ error: 'Failed to get viewer counts' });
new file mode 100644
new file mode 100644
@@ -0,0 +1,171 @@
@@ -0,0 +1,171 @@
+/**
+ * Restream control is slot-only: going live on a slot auto-starts and resumes that slot's
+ * destinations and nothing else; a legacy stream with no slot uses only unbound destinations.
+ *
+ * The production case: user 1 has slots 1, 60 and 85 with auto-start destinations on slot 1.
+ * The old _getDestinationsForStream merged every destination the account owned into a slotted
+ * stream, so going live on slot 60 or 85 started slot 1's Twitch and YouTube relays too.
+ * The display paths (destination list, broadcaster viewer counts) follow the same rule.
+ *
+ * node test/restream-slot-scope.test.js
+ */
+'use strict';
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-restream-slot-'));
+process.env.DB_PATH = path.join(tmp, 'live.db');
+process.env.NODE_ENV = 'test';
+const quiet = console.log;
+console.log = () => {};
+console.warn = () => {};
+
+const db = require('../server/db/database');
+db.initDb();
+const raw = db.getDb();
+
+const auth = require('../server/auth/auth');
+const signIn = (req) => {
+ const id = Number(req.headers['x-test-user'] || 0);
+ const u = id ? db.getUserById(id) : null;
+ if (u) { req.user = u; req.authSource = 'network'; }
+ return u;
+};
+auth.requireAuth = (req, res, next) => (signIn(req) ? next() : res.status(401).json({ error: 'Authentication required' }));
+auth.optionalAuth = (req, res, next) => { signIn(req); next(); };
+
+raw.prepare(`INSERT INTO users (id, username, display_name, email, password_hash, role) VALUES (1, 'alex', 'alex', 'a@x', 'x', 'streamer')`).run();
+raw.prepare(`INSERT INTO users (id, username, display_name, email, password_hash, role) VALUES (2, 'other', 'other', 'o@x', 'x', 'streamer')`).run();
+db.ensureChannel(1);
+db.ensureChannel(2);
+const chan = db.getChannelByUserId(1);
+
+const slot = (slug, userId = 1) => Number(db.createManagedStream({ user_id: userId, channel_id: chan.id, slug, title: slug, stream_key: `key-${slug}` }).lastInsertRowid);
+const slotA = slot('main'); // "slot 1": the auto-start destinations live here
+const slotB = slot('second'); // "slot 60"
+const slotC = slot('third'); // "slot 85": no destinations at all
+const otherSlot = slot('theirs', 2);
+
+const dest = (fields) => db.createRestreamDestination(fields.user_id || 1, {
+ server_url: 'rtmp://ingest.example/live', stream_key: `sk-${fields.name}`, enabled: 1, ...fields,
+}).id;
+const aTwitch = dest({ name: 'a-twitch', platform: 'twitch', managed_stream_id: slotA, auto_start: 1 });
+const aYoutube = dest({ name: 'a-youtube', platform: 'youtube', managed_stream_id: slotA, auto_start: 1 });
+const bKick = dest({ name: 'b-kick', platform: 'kick', managed_stream_id: slotB, auto_start: 1 });
+const bManual = dest({ name: 'b-manual', platform: 'custom', managed_stream_id: slotB, auto_start: 0 });
+const unbound = dest({ name: 'legacy', platform: 'custom', auto_start: 1 });
+const unboundManual = dest({ name: 'legacy-manual', platform: 'custom', auto_start: 0 });
+dest({ name: 'theirs', platform: 'twitch', managed_stream_id: otherSlot, auto_start: 1, user_id: 2 });
+
+const manager = require('../server/streaming/restream-manager');
+let started = [];
+manager.startRestream = async (streamId, d) => { started.push([streamId, d.id]); return { status: 'starting' }; };
+
+const live = (managedStreamId, userId = 1) => Number(db.createStream({ user_id: userId, channel_id: chan.id, managed_stream_id: managedStreamId, title: 't', protocol: 'rtmp' }).lastInsertRowid);
+const ids = (list) => list.map(([, destId]) => destId).sort((a, b) => a - b);
+
+let server;
+(async () => {
+ // ── Control: auto-start ──────────────────────────────────────────────────────────
+ const onB = live(slotB);
+ started = [];
+ await manager.autoStartForStream(onB, 1, { protocol: 'rtmp' });
+ assert.deepStrictEqual(ids(started), [bKick], 'going live on slot B auto-starts only slot B auto-start destinations (not slot A, not unbound)');
+
+ const onC = live(slotC);
+ started = [];
+ await manager.autoStartForStream(onC, 1, { protocol: 'rtmp' });
+ assert.deepStrictEqual(started, [], 'a slot with no destinations starts nothing, even though slot A has auto-start ones');
+
+ const onA = live(slotA);
+ started = [];
+ await manager.autoStartForStream(onA, 1, { protocol: 'rtmp' });
+ assert.deepStrictEqual(ids(started), [aTwitch, aYoutube].sort((a, b) => a - b), 'slot A starts its own two');
+
+ const legacy = live(null);
+ started = [];
+ await manager.autoStartForStream(legacy, 1, { protocol: 'rtmp' });
+ assert.deepStrictEqual(ids(started), [unbound], 'a legacy stream with no slot starts only unbound auto-start destinations');
+
+ // ── Control: resume (every enabled destination, auto-start or not) ──────────────
+ started = [];
+ await manager.resumeForStream(onB, 1, { protocol: 'rtmp' });
+ assert.deepStrictEqual(ids(started), [bKick, bManual].sort((a, b) => a - b), 'resume on slot B covers only slot B');
+ started = [];
+ await manager.resumeForStream(onC, 1, { protocol: 'rtmp' });
+ assert.deepStrictEqual(started, [], 'resume on an empty slot resumes nothing');
+ started = [];
+ await manager.resumeForStream(legacy, 1, { protocol: 'rtmp' });
+ assert.deepStrictEqual(ids(started), [unbound, unboundManual].sort((a, b) => a - b), 'resume on a slot-less stream covers only unbound destinations');
+
+ // The owner of the stream decides, not the caller's userId argument.
+ started = [];
+ await manager.autoStartForStream(onB, 2, { protocol: 'rtmp' });
+ assert.deepStrictEqual(ids(started), [bKick], 'a wrong userId cannot pull in another account\'s destinations');
+
+ // ── Display: external viewer counts ──────────────────────────────────────────────
+ manager.setViewerCount(aTwitch, 40, true);
+ manager.setViewerCount(bKick, 7, true);
+ manager.setViewerCount(unbound, 3, true);
+ assert.strictEqual(manager.getExternalViewerCountsForUser(1, slotB).total, 7, 'slot B counts only its own platform viewers');
+ assert.strictEqual(manager.getExternalViewerCountsForUser(1, slotA).total, 40);
+ assert.strictEqual(manager.getExternalViewerCountsForUser(1, null).total, 3, 'no slot counts only unbound destinations');
+
+ // ── Display: routes ─────────────────────────────────────────────────────────────
+ const express = require('express');
+ const app = express();
+ app.use(express.json());
+ app.use('/api/restream', require('../server/streaming/restream-routes'));
+ server = http.createServer(app).listen(0);
+ const call = (method, p, user, body) => new Promise((resolve, reject) => {
+ const data = body ? JSON.stringify(body) : null;
+ const req = http.request({ port: server.address().port, path: p, method, headers: { 'content-type': 'application/json', 'x-test-user': String(user) } }, (res) => {
+ let text = '';
+ res.on('data', (c) => { text += c; });
+ res.on('end', () => { let json = null; try { json = JSON.parse(text); } catch { /* */ } resolve({ status: res.statusCode, json }); });
+ });
+ req.on('error', reject);
+ if (data) req.write(data);
+ req.end();
+ });
+ const listed = (r) => r.json.destinations.map((d) => d.id).sort((a, b) => a - b);
+
+ let r = await call('GET', `/api/restream/destinations?managed_stream_id=${slotB}`, 1);
+ assert.strictEqual(r.status, 200);
+ assert.deepStrictEqual(listed(r), [bKick, bManual].sort((a, b) => a - b), 'slot listing is that slot only');
+ r = await call('GET', '/api/restream/destinations', 1);
+ assert.deepStrictEqual(listed(r), [unbound, unboundManual].sort((a, b) => a - b), 'no slot lists only unbound destinations, not the whole account');
+ r = await call('GET', '/api/restream/destinations?all=1', 1);
+ assert.strictEqual(r.json.destinations.length, 6, '?all=1 is the explicit whole-account listing');
+ assert.ok(r.json.destinations.every((d) => d.user_id === 1), 'and still only the caller\'s own');
+ r = await call('GET', `/api/restream/destinations?managed_stream_id=${otherSlot}`, 1);
+ assert.strictEqual(r.status, 403, 'another account\'s slot is refused');
+
+ // Viewer counts: explicit slot, or the slots live right now (A, B, C and slot-less).
+ r = await call('GET', `/api/restream/viewer-counts?managed_stream_id=${slotB}`, 1);
+ assert.strictEqual(r.json.total, 7);
+ r = await call('GET', '/api/restream/viewer-counts', 1);
+ assert.strictEqual(r.json.total, 50, 'the default counts each live stream\'s own slot once');
+ db.endStream(onA);
+ r = await call('GET', '/api/restream/viewer-counts', 1);
+ assert.strictEqual(r.json.total, 10, 'slot A is not live any more, so its Twitch viewers are not counted');
+ r = await call('GET', `/api/restream/viewer-counts?managed_stream_id=${otherSlot}`, 1);
+ assert.strictEqual(r.status, 403);
+
+ // Manual start of an unbound destination refuses a slotted stream.
+ r = await call('POST', `/api/restream/destinations/${unbound}/start`, 1, { streamId: onB });
+ assert.strictEqual(r.status, 400, 'an unbound destination cannot be started on a stream that has a slot');
+
+ server.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ quiet('restream-slot-scope: ok');
+ process.exit(0);
+})().catch((err) => {
+ console.error = quiet;
+ quiet(err);
+ try { server && server.close(); } catch { /* */ }
+ process.exit(1);
+});