openvibers/openvibe.live Commits
GOLIVE_NOTIFY=events (compatibility register C-85): leave go-live notifications to Network's live.stream.started consumer. While Live's events outbox publishes, a go-live then makes neither the direct POST /internal/events/stream-live nor the fallback push, and logs that it left it to the consumer; unset (the default), any other value, or with the outbox off, the direct call stays. Why: while both paths run, the direct call claims Network's per-streamer window first at every real go-live (production 2026-09-24: 'Unified event sent' for streams 2478, 2479 and 2481, the consumer's delivery a second later), so the consumer can never show the notified outcome C-85 waits for. Rollback: unset it. test/golive-notify-mode.test.js
Changes
@@@ -61,6 +61,10 @@ INTERNAL_API_KEY=change-me-to-a-random-secret# OpenVibe.Events (durable events): stream lifecycle is published through Live's outbox with the# OAuth client above (grant events.event.publish). Unset = off. EVENTS_PUBLISH=off also disables it.#EVENTS_URL=http://127.0.0.1:4300++# Go-live notifications (compatibility register C-85): `events` leaves them to Network's live.stream.started++# consumer (no direct POST /internal/events/stream-live, no fallback push); only while EVENTS_URL publishes.++# Unset = both paths, as before.++#GOLIVE_NOTIFY=events# OpenRe.Stream (roadmap Wave 7): the ingest/restream authority. Unset = off, and every stream slot# is ingested by Live exactly as before. When set, an admin can switch one slot at a time@@@ -7,6 +7,14 @@** Deduped per streamer+slot for 60 minutes: reconnects, deploy restarts and the recorder* re-creating a session must not spam followers.++ *++ * GOLIVE_NOTIFY=events (compatibility register C-85): leave go-lives to Network's live.stream.started++ * consumer. Live then makes neither the direct call nor the fallback push; the event (queued with++ * the streams row by server/events/stream-events.js, retried until Events has it) is the only path,++ * and Network sends the inbox, push, email and Discord alerts. It applies only while Live publishes++ * events; otherwise, and by default, the direct call stays. While both run, this call nearly always++ * claims Network's per-streamer window first, so the consumer can never show a real `notified`:++ * this switch is how C-85 gets its evidence. Rollback: unset it.*/const config = require('../config');const db = require('../db/database');@@@ -18,8 +26,18 @@ const _recent = new Map(); // `${userId}:${slot}` → tsfunction _channelUrl(streamer) { return `${(config.baseUrl || 'https://openvibe.live').replace(/\/$/, '')}/@${encodeURIComponent(streamer.username)}`; }++/** GOLIVE_NOTIFY=events and Live's events outbox is on: Network's consumer is the only go-live path. */++function leftToEvents() {++ if (String(process.env.GOLIVE_NOTIFY || '').trim().toLowerCase() !== 'events') return false;++ try { return !!require('../events/stream-events').status().enabled; } catch { return false; }++}++function notifyFollowersGoLive(streamer, stream, { force = false } = {}) {if (!streamer || !streamer.id) return;++ if (leftToEvents()) {++ console.log(`[GoLive] ${streamer.username}: left to Network's live.stream.started consumer (GOLIVE_NOTIFY=events)`);++ return;++ }const slot = (stream && (stream.managed_stream_id || stream.slot_slug)) || 'default';// Keyed by STREAMER (not slot): a person flapping between slots is still one person// going live. This in-memory guard is only a fast path — openvibe.network enforces the@@@ -96,4 +114,4 @@ function _fallback(streamer, stream, followerNetworkIds) {}, { alreadyNetworkIds: true });}--module.exports = { notifyFollowersGoLive };++module.exports = { notifyFollowersGoLive, leftToEvents };nnew file mode 100644@@@ -0,0 +1,70 @@++'use strict';++// GOLIVE_NOTIFY=events (compatibility register C-85): with Live's events outbox on, a go-live makes no call to++// Network (no POST /internal/events/stream-live, no fallback push) and leaves the followers to Network's++// live.stream.started consumer; the event itself is still queued with the streams row. Unset, or with the++// outbox off, the direct call stays.++++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-golive-mode-'));++process.env.DB_PATH = path.join(tmp, 'live.db');++process.env.INTERNAL_API_KEY = 'test-internal-key';++process.env.OV_NETWORK_INTERNAL_URL = 'http://127.0.0.1:9';++++const calls = [];++global.fetch = async (url, init = {}) => {++ calls.push({ url: String(url), method: init.method || 'GET' });++ return new Response(JSON.stringify({ ok: true, notifications: { sent: 0, total: 0 } }), { status: 200, headers: { 'content-type': 'application/json' } });++};++const logs = [];++const log = console.log;++console.log = (...a) => { logs.push(a.join(' ')); };++++(async () => {++ try {++ const db = require('../server/db/database');++ db.initDb();++ db.getDb().prepare("INSERT INTO users (id, username, display_name, password_hash) VALUES (601, 'caster', 'Caster', 'x')").run();++ const streamEvents = require('../server/events/stream-events');++ const { notifyFollowersGoLive, leftToEvents } = require('../server/streaming/golive-notify');++ const settle = () => new Promise((r) => setTimeout(r, 50));++ const toNetwork = () => calls.filter((c) => c.url.startsWith('http://127.0.0.1:9/')).length;++++ // Default: the direct call, as before.++ delete process.env.GOLIVE_NOTIFY;++ assert.strictEqual(leftToEvents(), false);++ notifyFollowersGoLive({ id: 601, username: 'caster' }, { id: 1, title: 'one' });++ await settle();++ assert.ok(toNetwork() >= 1, 'unset: Live calls Network directly');++++ // GOLIVE_NOTIFY=events while the outbox is off: still the direct call (nothing else would tell anyone).++ process.env.GOLIVE_NOTIFY = 'events';++ assert.strictEqual(streamEvents.status().enabled, false);++ assert.strictEqual(leftToEvents(), false);++++ // With the outbox on: no call at all, and the log says who announces.++ streamEvents.init({ eventsUrl: 'http://127.0.0.1:9', clientSecret: 's3cret', intervalMs: 60000 });++ assert.strictEqual(leftToEvents(), true);++ const before = calls.length;++ notifyFollowersGoLive({ id: 601, username: 'caster' }, { id: 2, title: 'two' }, { force: true });++ await settle();++ assert.deepStrictEqual(calls.slice(before).filter((c) => /\/internal\//.test(c.url)), [], 'no direct call, no fallback push');++ assert.ok(logs.some((l) => l.includes("caster: left to Network's live.stream.started consumer (GOLIVE_NOTIFY=events)")));++ // The event path is intact: a new stream row still queues live.stream.started.++ db.createStream({ user_id: 601, title: 'three', protocol: 'rtmp' });++ assert.strictEqual(db.getDb().prepare("SELECT COUNT(*) AS c FROM event_outbox WHERE envelope LIKE '%live.stream.started%'").get().c, 1);++++ // Any other value keeps both paths.++ process.env.GOLIVE_NOTIFY = 'both';++ assert.strictEqual(leftToEvents(), false);++ streamEvents._reset();++ } finally {++ console.log = log;++ fs.rmSync(tmp, { recursive: true, force: true });++ }++ console.log('golive notify mode: all checks passed');++ process.exit(0);++})().catch((err) => { console.log = log; console.error(err); process.exit(1); });
@@ -61,6 +61,10 @@ INTERNAL_API_KEY=change-me-to-a-random-secret
@@ -61,6 +61,10 @@ INTERNAL_API_KEY=change-me-to-a-random-secret
# OpenVibe.Events (durable events): stream lifecycle is published through Live's outbox with the
# OpenVibe.Events (durable events): stream lifecycle is published through Live's outbox with the
# OAuth client above (grant events.event.publish). Unset = off. EVENTS_PUBLISH=off also disables it.
# OAuth client above (grant events.event.publish). Unset = off. EVENTS_PUBLISH=off also disables it.
#EVENTS_URL=http://127.0.0.1:4300
#EVENTS_URL=http://127.0.0.1:4300
+# Go-live notifications (compatibility register C-85): `events` leaves them to Network's live.stream.started
+# consumer (no direct POST /internal/events/stream-live, no fallback push); only while EVENTS_URL publishes.
+# Unset = both paths, as before.
+#GOLIVE_NOTIFY=events
# OpenRe.Stream (roadmap Wave 7): the ingest/restream authority. Unset = off, and every stream slot
# OpenRe.Stream (roadmap Wave 7): the ingest/restream authority. Unset = off, and every stream slot
# is ingested by Live exactly as before. When set, an admin can switch one slot at a time
# is ingested by Live exactly as before. When set, an admin can switch one slot at a time
@@ -7,6 +7,14 @@
@@ -7,6 +7,14 @@
*
*
* Deduped per streamer+slot for 60 minutes: reconnects, deploy restarts and the recorder
* Deduped per streamer+slot for 60 minutes: reconnects, deploy restarts and the recorder
* re-creating a session must not spam followers.
* re-creating a session must not spam followers.
+ *
+ * GOLIVE_NOTIFY=events (compatibility register C-85): leave go-lives to Network's live.stream.started
+ * consumer. Live then makes neither the direct call nor the fallback push; the event (queued with
+ * the streams row by server/events/stream-events.js, retried until Events has it) is the only path,
+ * and Network sends the inbox, push, email and Discord alerts. It applies only while Live publishes
+ * events; otherwise, and by default, the direct call stays. While both run, this call nearly always
+ * claims Network's per-streamer window first, so the consumer can never show a real `notified`:
+ * this switch is how C-85 gets its evidence. Rollback: unset it.
*/
*/
const config = require('../config');
const config = require('../config');
const db = require('../db/database');
const db = require('../db/database');
@@ -18,8 +26,18 @@ const _recent = new Map(); // `${userId}:${slot}` → ts
@@ -18,8 +26,18 @@ const _recent = new Map(); // `${userId}:${slot}` → ts
function _channelUrl(streamer) { return `${(config.baseUrl || 'https://openvibe.live').replace(/\/$/, '')}/@${encodeURIComponent(streamer.username)}`; }
function _channelUrl(streamer) { return `${(config.baseUrl || 'https://openvibe.live').replace(/\/$/, '')}/@${encodeURIComponent(streamer.username)}`; }
+/** GOLIVE_NOTIFY=events and Live's events outbox is on: Network's consumer is the only go-live path. */
+function leftToEvents() {
+ if (String(process.env.GOLIVE_NOTIFY || '').trim().toLowerCase() !== 'events') return false;
+ try { return !!require('../events/stream-events').status().enabled; } catch { return false; }
+}
+
function notifyFollowersGoLive(streamer, stream, { force = false } = {}) {
function notifyFollowersGoLive(streamer, stream, { force = false } = {}) {
if (!streamer || !streamer.id) return;
if (!streamer || !streamer.id) return;
+ if (leftToEvents()) {
+ console.log(`[GoLive] ${streamer.username}: left to Network's live.stream.started consumer (GOLIVE_NOTIFY=events)`);
+ return;
+ }
const slot = (stream && (stream.managed_stream_id || stream.slot_slug)) || 'default';
const slot = (stream && (stream.managed_stream_id || stream.slot_slug)) || 'default';
// Keyed by STREAMER (not slot): a person flapping between slots is still one person
// Keyed by STREAMER (not slot): a person flapping between slots is still one person
// going live. This in-memory guard is only a fast path — openvibe.network enforces the
// going live. This in-memory guard is only a fast path — openvibe.network enforces the
@@ -96,4 +114,4 @@ function _fallback(streamer, stream, followerNetworkIds) {
@@ -96,4 +114,4 @@ function _fallback(streamer, stream, followerNetworkIds) {
}, { alreadyNetworkIds: true });
}, { alreadyNetworkIds: true });
}
}
-module.exports = { notifyFollowersGoLive };
+module.exports = { notifyFollowersGoLive, leftToEvents };
new file mode 100644
new file mode 100644
@@ -0,0 +1,70 @@
@@ -0,0 +1,70 @@
+'use strict';
+// GOLIVE_NOTIFY=events (compatibility register C-85): with Live's events outbox on, a go-live makes no call to
+// Network (no POST /internal/events/stream-live, no fallback push) and leaves the followers to Network's
+// live.stream.started consumer; the event itself is still queued with the streams row. Unset, or with the
+// outbox off, the direct call stays.
+
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-golive-mode-'));
+process.env.DB_PATH = path.join(tmp, 'live.db');
+process.env.INTERNAL_API_KEY = 'test-internal-key';
+process.env.OV_NETWORK_INTERNAL_URL = 'http://127.0.0.1:9';
+
+const calls = [];
+global.fetch = async (url, init = {}) => {
+ calls.push({ url: String(url), method: init.method || 'GET' });
+ return new Response(JSON.stringify({ ok: true, notifications: { sent: 0, total: 0 } }), { status: 200, headers: { 'content-type': 'application/json' } });
+};
+const logs = [];
+const log = console.log;
+console.log = (...a) => { logs.push(a.join(' ')); };
+
+(async () => {
+ try {
+ const db = require('../server/db/database');
+ db.initDb();
+ db.getDb().prepare("INSERT INTO users (id, username, display_name, password_hash) VALUES (601, 'caster', 'Caster', 'x')").run();
+ const streamEvents = require('../server/events/stream-events');
+ const { notifyFollowersGoLive, leftToEvents } = require('../server/streaming/golive-notify');
+ const settle = () => new Promise((r) => setTimeout(r, 50));
+ const toNetwork = () => calls.filter((c) => c.url.startsWith('http://127.0.0.1:9/')).length;
+
+ // Default: the direct call, as before.
+ delete process.env.GOLIVE_NOTIFY;
+ assert.strictEqual(leftToEvents(), false);
+ notifyFollowersGoLive({ id: 601, username: 'caster' }, { id: 1, title: 'one' });
+ await settle();
+ assert.ok(toNetwork() >= 1, 'unset: Live calls Network directly');
+
+ // GOLIVE_NOTIFY=events while the outbox is off: still the direct call (nothing else would tell anyone).
+ process.env.GOLIVE_NOTIFY = 'events';
+ assert.strictEqual(streamEvents.status().enabled, false);
+ assert.strictEqual(leftToEvents(), false);
+
+ // With the outbox on: no call at all, and the log says who announces.
+ streamEvents.init({ eventsUrl: 'http://127.0.0.1:9', clientSecret: 's3cret', intervalMs: 60000 });
+ assert.strictEqual(leftToEvents(), true);
+ const before = calls.length;
+ notifyFollowersGoLive({ id: 601, username: 'caster' }, { id: 2, title: 'two' }, { force: true });
+ await settle();
+ assert.deepStrictEqual(calls.slice(before).filter((c) => /\/internal\//.test(c.url)), [], 'no direct call, no fallback push');
+ assert.ok(logs.some((l) => l.includes("caster: left to Network's live.stream.started consumer (GOLIVE_NOTIFY=events)")));
+ // The event path is intact: a new stream row still queues live.stream.started.
+ db.createStream({ user_id: 601, title: 'three', protocol: 'rtmp' });
+ assert.strictEqual(db.getDb().prepare("SELECT COUNT(*) AS c FROM event_outbox WHERE envelope LIKE '%live.stream.started%'").get().c, 1);
+
+ // Any other value keeps both paths.
+ process.env.GOLIVE_NOTIFY = 'both';
+ assert.strictEqual(leftToEvents(), false);
+ streamEvents._reset();
+ } finally {
+ console.log = log;
+ fs.rmSync(tmp, { recursive: true, force: true });
+ }
+ console.log('golive notify mode: all checks passed');
+ process.exit(0);
+})().catch((err) => { console.log = log; console.error(err); process.exit(1); });