openvibers/openvibe.live Commits

Live: OpenVibe Live on by default for OpenRestream — a stream made on openre.stream gets its own slot on the person's channel (#65)

OP OpenVibers committed 9 Oct 2026 · 3e112dac2c · 4 files, +196 −0

- POST /internal/openre/slots (capability live.openre.slot.bind, contracts 0.126.0; loopback and a Network service
  token, like every internal route): server/openre/bind.js creates a new slot for the person (by their Network
  subject) with ingest_authority openre and openre_stream_id set, or returns the one already bound to that stream.
  Its sessions then go live on openvibe.live/@username through the existing mirror. An existing slot is never
  changed; a transaction-scoped advisory lock keeps two binds for one stream from making two slots.
- Refusals: 400 bad input; 409 live.no_account (no Live account yet), live.slot_taken (someone else's stream),
  live.slot_limit (the account's slots are full); 403 live.account_banned.
- test/openre-bind.test.js: bind, idempotency under a race, the refusals, an untouched existing slot, and a
  session going live through the mirror (and not when switched off). docs/broadcasting.md: the OpenRestream section.

Changes

ddiff --git a/docs/broadcasting.md b/docs/broadcasting.md
iindex 02947ba6..327f5bf9 100644
---- a/docs/broadcasting.md
++++ b/docs/broadcasting.md
@@@ -22,6 +22,19 @@ OpenVibe.Live supports four live broadcast methods, each suited to different use
- **Setup**: Choose *Streaming method: WHIP*, copy the WHIP URL into OBS (Service: WHIP) or `ffmpeg -f whip`
- **Features**: VP8/H.264 + Opus, auto-creates the live session on first POST, open CORS so a static page can publish with `RTCPeerConnection` + `fetch()` — see the [WHIP Ingest API](whip.md) / [Publishing from a browser](whip.md#publishing-from-a-browser) and the hosted [browser publisher](https://openvibe.live/whip-publisher.html)
++### OpenRestream (OBS, restreaming everywhere)
++- **Best for**: Streaming to Twitch, YouTube, Kick and other servers at the same time as your OpenVibe Live channel
++- **Setup**: Create a stream on [openre.stream](https://openre.stream), copy its RTMP server and key into OBS, and add
++ your other destinations there
++- **On Live by default**: each stream made on OpenRestream gets its own slot on your channel, marked *managed on
++ OpenRestream*. Its sessions go live on `openvibe.live/@you` with chat, clips and VODs, through the OpenRestream mirror
++ (`server/openre/mirror.js`). Switch "Show it on my OpenVibe Live channel" off on OpenRestream and Live ignores that
++ stream; your other destinations keep going.
++- **How**: OpenRestream calls `POST /internal/openre/slots` (capability `live.openre.slot.bind`,
++ [server/openre/bind.js](../server/openre/bind.js)) with a Network service token. Live creates a new slot with
++ `ingest_authority = 'openre'`, or returns the one already bound to that stream. It never changes an existing slot.
++ A person needs a Live account first, which they get by signing in to openvibe.live once.
++
### JSMPEG (FFmpeg)
- **Best for**: Headless/embedded streaming (Raspberry Pi, IP cameras, 3D printers)
- **Latency**: ~1 second
ddiff --git a/server/internal/routes.js b/server/internal/routes.js
iindex 36b0378e..6a1350e9 100644
---- a/server/internal/routes.js
++++ b/server/internal/routes.js
@@@ -6,6 +6,8 @@ const config = require('../config');
const db = require('../db/database');
const { guard } = require('../net/service-guard');
++const { http } = require('openvibe-contracts');
++const { bindSlot, BindError } = require('../openre/bind');
// Service-to-service only, loopback only (nothing that came through nginx): each route checks the one capability it
// performs on a Network service token (server/net/service-guard.js); nothing else gets in (X-Internal-Key: plan T2).
@@@ -64,4 +66,19 @@ router.post('/user-avatar', guard('live.avatar.write'), async (req, res) => {
}
});
++// ── OpenVibe Live on by default for OpenRestream (live.openre.slot.bind) ─────────
++// OpenRestream asks for a slot on the person's channel bound to their OpenRestream stream (server/openre/bind.js).
++router.post('/openre/slots', guard('live.openre.slot.bind'), async (req, res) => {
++ try {
++ const out = await bindSlot(req.body || {});
++ return res.status(out.created ? 201 : 200).json(out);
++ } catch (err) {
++ if (err instanceof BindError) {
++ return http.sendProblem(res, err.status, err.code, { detail: err.message, ctx: http.requestContext(req.headers) });
++ }
++ console.error('[Internal] openre/slots error:', err.message);
++ return res.status(500).json({ ok: false, error: 'slot bind failed' });
++ }
++});
++
module.exports = router;
ddiff --git a/server/openre/bind.js b/server/openre/bind.js
nnew file mode 100644
iindex 00000000..904bb71c
---- /dev/null
++++ b/server/openre/bind.js
@@@ -0,0 +1,78 @@
++'use strict';
++/**
++ * OpenVibe Live on by default for OpenRestream (contracts 0.126.0, capability live.openre.slot.bind).
++ *
++ * A stream someone makes on openre.stream gets its own slot on their Live channel: a managed_streams row ingested by
++ * OpenRestream from the start (ingest_authority 'openre', openre_stream_id = the stream). Its sessions then reach
++ * `streams` through the mirror (mirror.js) like any switched slot, so the person is live on openvibe.live/@username
++ * with chat, clips and VODs. Nothing about an existing slot changes: this only ever creates a new one, or returns the
++ * one already bound to that stream.
++ *
++ * The slot's own Live key is random and never shown: Live refuses it while the slot is on OpenRestream
++ * (authority.refusesLiveIngest), and the broadcaster's key is OpenRestream's.
++ */
++const crypto = require('crypto');
++const db = require('../db/database');
++const config = require('../config');
++
++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++const STREAM_RE = /^std_[0-9A-HJKMNP-TV-Z]{26}$/;
++const PROTOCOLS = { rtmp: { protocol: 'rtmp', method: 'obs' }, webrtc: { protocol: 'webrtc', method: 'whip' }, jsmpeg: { protocol: 'jsmpeg', method: null } };
++
++class BindError extends Error {
++ constructor(status, code, message) { super(message); this.status = status; this.code = code; }
++}
++
++function channelUrl(username) {
++ return `${String(config.baseUrl || 'https://openvibe.live').replace(/\/+$/, '')}/@${encodeURIComponent(username)}`;
++}
++
++/** The Live user a Network subject signed in as (linked_accounts, written at sign-in), or null. */
++async function userForSubject(subject) {
++ const row = await db.get("SELECT user_id FROM linked_accounts WHERE service = 'network' AND subject_id = ? ORDER BY user_id LIMIT 1", [subject]);
++ return row ? await db.getUserById(row.user_id) : null;
++}
++
++/**
++ * Find or create the slot bound to one OpenRestream stream. { managed_stream_id, channel_url, created }, or throws
++ * BindError (400 bad input, 409 live.no_account / live.slot_taken / live.slot_limit, 403 live.account_banned).
++ */
++async function bindSlot({ subject, openre_stream_id: streamId, title, protocol } = {}) {
++ if (!SUBJECT_RE.test(String(subject || ''))) throw new BindError(400, 'request.invalid', 'subject must be a usr_ subject id');
++ if (!STREAM_RE.test(String(streamId || ''))) throw new BindError(400, 'request.invalid', 'openre_stream_id must be an OpenRestream std_ id');
++ const kind = PROTOCOLS[protocol || 'rtmp'];
++ if (!kind) throw new BindError(400, 'request.invalid', 'protocol must be rtmp, webrtc or jsmpeg');
++ const user = await userForSubject(subject);
++ if (!user) throw new BindError(409, 'live.no_account', 'this person has no OpenVibe Live account yet: they sign in to openvibe.live once');
++ if (user.is_banned) throw new BindError(403, 'live.account_banned', 'this Live account is banned');
++ const name = String(title || '').replace(/[\u0000-\u001f\u007f]/g, ' ').replace(/\s+/g, ' ').trim().slice(0, 200) || 'OpenRestream';
++
++ return await db.getDb().tx(async () => {
++ // One bind per stream at a time: two calls for the same stream wait here, and the second finds the first's slot.
++ await db.get('SELECT pg_advisory_xact_lock(hashtext(?))', [`live.openre.slot:${streamId}`]);
++ const bound = await db.get('SELECT id, user_id FROM managed_streams WHERE openre_stream_id = ? LIMIT 1', [streamId]);
++ if (bound) {
++ if (bound.user_id !== user.id) throw new BindError(409, 'live.slot_taken', 'that OpenRestream stream is bound to another person\'s slot');
++ return { managed_stream_id: bound.id, channel_url: channelUrl(user.username), created: false };
++ }
++ const limit = await db.getManagedStreamLimit(user);
++ if (await db.countManagedStreamsByUser(user.id) >= limit) {
++ throw new BindError(409, 'live.slot_limit', `this Live account already has its ${limit} stream slots`);
++ }
++ const channel = await db.ensureChannel(user.id);
++ const created = await db.createManagedStream({
++ user_id: user.id,
++ channel_id: channel && channel.id,
++ title: name,
++ protocol: kind.protocol,
++ streaming_method: kind.method,
++ stream_key: crypto.randomBytes(20).toString('hex'),
++ });
++ const id = Number(created.lastInsertRowid);
++ await db.run("UPDATE managed_streams SET ingest_authority = 'openre', openre_stream_id = ?, updated_at = ov_now() WHERE id = ?", [streamId, id]);
++ console.log(`[OpenRestream] stream ${streamId} bound to new slot ${id} (${user.username})`);
++ return { managed_stream_id: id, channel_url: channelUrl(user.username), created: true };
++ });
++}
++
++module.exports = { bindSlot, BindError, userForSubject };
ddiff --git a/test/openre-bind.test.js b/test/openre-bind.test.js
nnew file mode 100644
iindex 00000000..c194d7f1
---- /dev/null
++++ b/test/openre-bind.test.js
@@@ -0,0 +1,88 @@
++'use strict';
++
++// OpenVibe Live on by default for OpenRestream (contracts 0.126.0, live.openre.slot.bind): a stream someone makes on
++// openre.stream gets its own new slot on their Live channel, ingested by OpenRestream, and its sessions become live
++// `streams` rows through the mirror. Existing slots are never touched; a second bind for the stream returns the same
++// slot; someone else's stream, a person with no Live account and a full set of slots are refused.
++
++const assert = require('assert');
++
++process.env.OPENRE_URL = 'http://127.0.0.1:9'; // configured (never called here: a started event needs no call)
++process.env.OV_OAUTH_CLIENT_SECRET = 'test-secret';
++
++const SUBJECT = 'usr_01J0000000000000000000000B';
++const OTHER = 'usr_01J0000000000000000000000C';
++const NOBODY = 'usr_01J0000000000000000000000D';
++const STREAM = 'std_01J0000000000000000000000B';
++const SES = 'ses_01J0000000000000000000000B';
++
++(async () => {
++ const db = require('../server/db/database');
++ await db.initDb();
++ const d = db.getDb();
++ await d.prepare("INSERT INTO users (id, username, display_name, password_hash, stream_key) OVERRIDING SYSTEM VALUE VALUES (801, 'restreamer', 'Restreamer', 'x', 'personalkey801')").run();
++ await d.prepare("INSERT INTO users (id, username, display_name, password_hash) OVERRIDING SYSTEM VALUE VALUES (802, 'other', 'Other', 'x')").run();
++ await d.prepare(`INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (801, 'network', '81', '${SUBJECT}')`).run();
++ await d.prepare(`INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (802, 'network', '82', '${OTHER}')`).run();
++ const existing = await db.createManagedStream({ user_id: 801, title: 'My usual slot', protocol: 'rtmp', streaming_method: 'obs', stream_key: 'slotkey801' });
++ const before = await db.getManagedStreamById(existing.lastInsertRowid);
++
++ const { bindSlot, BindError } = require('../server/openre/bind');
++ const refused = async (input, status, code) => {
++ await assert.rejects(() => bindSlot(input), (err) => err instanceof BindError && err.status === status && err.code === code, `${code}`);
++ };
++
++ // Bad input and no Live account.
++ await refused({ subject: 'nope', openre_stream_id: STREAM }, 400, 'request.invalid');
++ await refused({ subject: SUBJECT, openre_stream_id: 'std_x' }, 400, 'request.invalid');
++ await refused({ subject: SUBJECT, openre_stream_id: STREAM, protocol: 'hls' }, 400, 'request.invalid');
++ await refused({ subject: NOBODY, openre_stream_id: STREAM }, 409, 'live.no_account');
++ console.log(' ✓ bad input and a person with no Live account are refused');
++
++ // A new slot, ingested by OpenRestream; the existing slot is untouched.
++ const first = await bindSlot({ subject: SUBJECT, openre_stream_id: STREAM, title: ' Saturday\u0007 streams ', protocol: 'rtmp' });
++ assert.strictEqual(first.created, true);
++ assert.match(first.channel_url, /^https?:\/\/[^/]+\/@restreamer$/);
++ const slot = await db.getManagedStreamById(first.managed_stream_id);
++ assert.strictEqual(slot.user_id, 801);
++ assert.strictEqual(slot.ingest_authority, 'openre');
++ assert.strictEqual(slot.openre_stream_id, STREAM);
++ assert.strictEqual(slot.title, 'Saturday streams');
++ assert.strictEqual(slot.protocol, 'rtmp');
++ assert.ok(slot.stream_key && slot.stream_key !== 'personalkey801' && slot.stream_key !== 'slotkey801', 'a fresh Live key nobody is shown');
++ const after = await db.getManagedStreamById(before.id);
++ assert.deepStrictEqual({ a: after.ingest_authority, k: after.stream_key, o: after.openre_stream_id }, { a: before.ingest_authority, k: before.stream_key, o: before.openre_stream_id }, 'the existing slot is untouched');
++ console.log(' ✓ a stream gets a new slot on the person\'s channel, ingested by OpenRestream; the existing slot is untouched');
++
++ // Idempotent, also when two binds race.
++ const [again, race] = await Promise.all([bindSlot({ subject: SUBJECT, openre_stream_id: STREAM }), bindSlot({ subject: SUBJECT, openre_stream_id: STREAM })]);
++ assert.deepStrictEqual([again.managed_stream_id, again.created, race.managed_stream_id, race.created], [first.managed_stream_id, false, first.managed_stream_id, false]);
++ assert.strictEqual((await d.prepare('SELECT count(*)::int AS n FROM managed_streams WHERE openre_stream_id = ?').get(STREAM)).n, 1);
++ console.log(' ✓ a second bind (and two at once) returns the same slot');
++
++ // Someone else's stream id, and a full set of slots.
++ await refused({ subject: OTHER, openre_stream_id: STREAM }, 409, 'live.slot_taken');
++ await d.prepare('UPDATE users SET max_managed_streams = 1 WHERE id = 802').run();
++ await db.createManagedStream({ user_id: 802, title: 'Only slot', protocol: 'rtmp', stream_key: 'slotkey802' });
++ await refused({ subject: OTHER, openre_stream_id: 'std_01J0000000000000000000000C' }, 409, 'live.slot_limit');
++ console.log(' ✓ another person\'s stream and a full set of slots are refused');
++
++ // The mirror shows the stream's session live on the channel.
++ const mirror = require('../server/openre/mirror');
++ const event = {
++ event_id: 'evt_01J0000000000000000000000B', event_type: 'openre.session.started', version: 1, source: 'openre',
++ actor: { type: 'user', id: SUBJECT }, timestamp: new Date().toISOString(),
++ subject: { type: 'ingest_session', id: SES, revision: 1 },
++ payload: { session_id: SES, external_refs: [{ service: 'live', type: 'managed_stream', id: String(first.managed_stream_id) }], mirror_to_live: true, started_at: new Date().toISOString() },
++ };
++ const r = await mirror.apply(event);
++ assert.strictEqual(r.outcome, 'created');
++ const live = await d.prepare('SELECT id, user_id, is_live FROM streams WHERE managed_stream_id = ? AND is_live = 1').get(first.managed_stream_id);
++ assert.ok(live && live.user_id === 801, 'the session is a live stream on the person\'s channel');
++ const off = await mirror.apply({ ...event, event_id: 'evt_01J0000000000000000000000C', subject: { ...event.subject, id: 'ses_01J0000000000000000000000C' }, payload: { ...event.payload, session_id: 'ses_01J0000000000000000000000C', mirror_to_live: false } });
++ assert.strictEqual(off.outcome, 'not_mirrored', 'switched off on OpenRestream: not shown on Live');
++ console.log(' ✓ the bound slot\'s session goes live on the channel through the mirror; switched off, it does not');
++
++ console.log('openre bind: all checks passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });