openvibers/openvibe.live Commits

Stream lifecycle as durable events: live.stream.started/ended go to OpenVibe.Events through a transactional outbox (openvibe-sdk 0.2.2); openvibe-contracts v0.8.0

OP OpenVibers committed 23 Sept 2026 · 8a678c8943 · 7 files, +281 −19


Changes

ddiff --git a/.env.example b/.env.example
iindex 31f53b75..779ce86d 100644
---- a/.env.example
++++ b/.env.example
@@@ -38,6 +38,9 @@ OV_OAUTH_CLIENT_ID=live
OV_OAUTH_CLIENT_SECRET=
# Per-service key for Network's /internal/* endpoints (header X-Internal-Key)
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
# Path to the Network's RS256 public key (offline JWT verification)
#OV_NETWORK_PUBLIC_KEY=./data/keys/openvibe-network-public.pem
ddiff --git a/package-lock.json b/package-lock.json
iindex 60623f05..ff92f949 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -25,7 +25,8 @@
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.5.1",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.8.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.2.2",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.0.0",
"sanitize-html": "^2.17.7",
"sdp-transform": "^3.0.0",
@@@ -2813,9 +2814,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.5.1",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.5.1",
-- "integrity": "sha512-czQpWKEo8cY56Crj5l5Mv8r5vlIIj0vn/HMxv0OlIu5YTuSSlpvXyayYEfn3GQ5orDhaMIhOUlzmQMxUtWXaLA==",
++ "version": "0.8.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.8.0",
++ "integrity": "sha512-2Msdb+k+fJZf9kQTPS2jdY8+o9FhL9tBuu3FzArHs3JcBcc06ReHpOi2sgnSBgQcSvZSaSsWT2YVbmhjnEplqQ==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
@@@ -2828,6 +2829,15 @@
"node": ">=20"
}
},
++ "node_modules/openvibe-sdk": {
++ "version": "0.2.2",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.2.2",
++ "integrity": "sha512-kDifUp3EcmnTWzElAmaCZnX659WErcnm4+qVEoAiYQWX/w/1p4TWvoHbUPsrX65TSROVR3tiyVkl801n/0yyCw==",
++ "license": "MIT",
++ "engines": {
++ "node": ">=20"
++ }
++ },
"node_modules/openvibe-shared": {
"version": "1.0.0",
"resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.0.0",
ddiff --git a/package.json b/package.json
iindex 8706f192..d5b6174e 100644
---- a/package.json
++++ b/package.json
@@@ -43,7 +43,8 @@
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.5.1",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.8.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.2.2",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.0.0",
"sanitize-html": "^2.17.7",
"sdp-transform": "^3.0.0",
ddiff --git a/server/db/database.js b/server/db/database.js
iindex 73d879d5..30c778a0 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -2927,23 +2927,41 @@ function getStreamHistoryByManagedStream(managedStreamId, userId, limit = 20) {
`, [managedStreamId, userId, limit]);
}
++// Stream lifecycle hook (server/events/stream-events.js registers it at boot). It runs inside the
++// same transaction as the write, so the durable event exists if and only if the change committed
++// (roadmap Wave 3, ADR-004). A failing hook is logged and never blocks going live or ending.
++let streamLifecycleHook = null;
++function onStreamLifecycle(fn) { streamLifecycleHook = typeof fn === 'function' ? fn : null; }
++function fireStreamLifecycle(kind, streamId) {
++ if (!streamLifecycleHook) return;
++ try { streamLifecycleHook(kind, streamId); } catch (err) { console.warn(`[Events] stream ${kind} event for ${streamId} not queued:`, err.message); }
++}
++
function createStream({ user_id, channel_id, managed_stream_id, control_config_id, title, description, category, protocol, is_nsfw, thumbnail_url }) {
-- return run(
-- `INSERT INTO streams (user_id, channel_id, managed_stream_id, control_config_id, title, description, category, protocol, is_nsfw, thumbnail_url, is_live, started_at)
-- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, CURRENT_TIMESTAMP)`,
-- [user_id, channel_id || null, managed_stream_id || null, control_config_id || null, title || 'Untitled Stream', description || '', category || null, protocol || 'webrtc', is_nsfw ? 1 : 0, thumbnail_url || null]
-- );
++ return getDb().transaction(() => {
++ const result = run(
++ `INSERT INTO streams (user_id, channel_id, managed_stream_id, control_config_id, title, description, category, protocol, is_nsfw, thumbnail_url, is_live, started_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, CURRENT_TIMESTAMP)`,
++ [user_id, channel_id || null, managed_stream_id || null, control_config_id || null, title || 'Untitled Stream', description || '', category || null, protocol || 'webrtc', is_nsfw ? 1 : 0, thumbnail_url || null]
++ );
++ fireStreamLifecycle('started', result.lastInsertRowid);
++ return result;
++ })();
}
function endStream(streamId) {
-- const stream = get('SELECT started_at FROM streams WHERE id = ?', [streamId]);
-- if (!stream) return null;
-- return run(
-- `UPDATE streams SET is_live = 0, ended_at = CURRENT_TIMESTAMP,
-- duration_seconds = CAST((julianday(CURRENT_TIMESTAMP) - julianday(started_at)) * 86400 AS INTEGER)
-- WHERE id = ?`,
-- [streamId]
-- );
++ return getDb().transaction(() => {
++ const stream = get('SELECT started_at, is_live FROM streams WHERE id = ?', [streamId]);
++ if (!stream) return null;
++ const result = run(
++ `UPDATE streams SET is_live = 0, ended_at = CURRENT_TIMESTAMP,
++ duration_seconds = CAST((julianday(CURRENT_TIMESTAMP) - julianday(started_at)) * 86400 AS INTEGER)
++ WHERE id = ?`,
++ [streamId]
++ );
++ if (stream.is_live) fireStreamLifecycle('ended', streamId);
++ return result;
++ })();
}
/**
@@@ -8393,7 +8411,7 @@ module.exports = {
ensureStreamerRoleOnFeed,
// Streams (sessions)
getLiveStreams, getRecentStreams, getStreamById, setStreamAiCategory, effectiveCategory, getStreamByUserId, getLiveStreamsByUserId, getLiveStreamsByControlConfigId, getStreamsByUserId, getStreamHistoryByManagedStream,
-- createStream, endStream, endOtherLiveStreamsForSlot, updateViewerCount,
++ createStream, endStream, onStreamLifecycle, endOtherLiveStreamsForSlot, updateViewerCount,
addStreamMemory, getStreamMemories, getLatestStreamMemory, updateStreamAiOverview,
setVodAiOverview, setClipAiOverview, setVodTranscript, setClipTranscript, getStreamMemoriesInRange,
getVodsNeedingOverview, getClipsNeedingOverview, getVodsNeedingTimeline, getVodsNeedingTranscript, getClipsNeedingTranscript, getPastesNeedingAnalysis,
ddiff --git a/server/events/stream-events.js b/server/events/stream-events.js
nnew file mode 100644
iindex 00000000..17b4e6d9
---- /dev/null
++++ b/server/events/stream-events.js
@@@ -0,0 +1,110 @@
++'use strict';
++/**
++ * Live → OpenVibe.Events: stream lifecycle as durable events (roadmap Wave 3, ADR-004, ADR-020).
++ *
++ * live.stream.started a stream row went live (subject: stream <id>, revision 1)
++ * live.stream.ended a live stream row ended (subject: stream <id>, revision 2)
++ *
++ * Events are written to Live's own `event_outbox` table inside the same SQLite transaction as the
++ * streams row (database.js onStreamLifecycle), and a relay publishes them with Live's service
++ * token (audience openvibe.events, capability events.event.publish). If Events or Network is down
++ * the rows wait and are retried with backoff; going live never waits on either.
++ *
++ * Off unless EVENTS_URL and OV_OAUTH_CLIENT_SECRET are set (EVENTS_PUBLISH=off disables it).
++ * Payloads carry public channel facts only (the stream is already listed publicly); consumers such
++ * as Network's go-live notifications decide who hears about it.
++ */
++const { createClient } = require('openvibe-sdk/core');
++const { createServiceTokenClient } = require('openvibe-sdk/auth');
++const { createEventsClient, createOutbox } = require('openvibe-sdk/events');
++const db = require('../db/database');
++const identity = require('../auth/identity-sync');
++
++const EVENTS_URL = (process.env.EVENTS_URL || '').replace(/\/+$/, '');
++const NETWORK_INTERNAL_URL = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
++const CLIENT_ID = process.env.OV_OAUTH_CLIENT_ID || 'live';
++const CLIENT_SECRET = process.env.OV_OAUTH_CLIENT_SECRET || '';
++const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;
++
++let outbox = null;
++const stats = { queued: 0, lastError: null };
++
++/** The envelope for one lifecycle change, read inside the caller's transaction. */
++function envelopeFor(kind, streamId) {
++ const s = db.getDb().prepare(`SELECT s.id, s.user_id, s.title, s.category, s.protocol, s.is_nsfw, s.started_at, s.ended_at,
++ s.duration_seconds, s.managed_stream_id, u.username, u.display_name
++ FROM streams s JOIN users u ON u.id = s.user_id WHERE s.id = ?`).get(streamId);
++ if (!s) return null;
++ const subjectId = identity.subjectOf(s.user_id);
++ const channel = { username: s.username, display_name: s.display_name || s.username, url: `https://openvibe.live/${encodeURIComponent(s.username)}` };
++ if (subjectId) channel.subject = { type: 'user', id: subjectId };
++ const payload = {
++ stream_id: s.id,
++ channel,
++ title: s.title,
++ category: s.category || null,
++ protocol: s.protocol,
++ is_nsfw: !!s.is_nsfw,
++ started_at: toIso(s.started_at),
++ };
++ if (kind === 'ended') {
++ payload.ended_at = toIso(s.ended_at);
++ payload.duration_seconds = s.duration_seconds == null ? null : Number(s.duration_seconds);
++ }
++ return {
++ event_type: `live.stream.${kind}`,
++ actor: subjectId ? { type: 'user', id: subjectId } : { type: 'service', id: 'live' },
++ subject: { type: 'stream', id: String(s.id), revision: kind === 'started' ? 1 : 2 },
++ visibility: 'public',
++ priority: 'important',
++ payload,
++ };
++}
++
++// SQLite CURRENT_TIMESTAMP is UTC without a zone ('2026-09-23 01:30:00').
++function toIso(v) {
++ if (!v) return null;
++ const d = new Date(String(v).includes('T') ? v : `${String(v).replace(' ', 'T')}Z`);
++ return Number.isNaN(d.getTime()) ? null : d.toISOString();
++}
++
++function init({ eventsUrl = EVENTS_URL, clientSecret = CLIENT_SECRET, fetchImpl, intervalMs } = {}) {
++ if (outbox) return outbox;
++ if (process.env.EVENTS_PUBLISH === 'off' || !eventsUrl || !clientSecret) {
++ return null;
++ }
++ const tokens = createServiceTokenClient({ tokenUrl: `${NETWORK_INTERNAL_URL}/oauth/token`, clientId: CLIENT_ID, clientSecret, fetch: fetchImpl });
++ const client = createClient({ baseUrls: { events: eventsUrl }, tokenProvider: tokens, fetch: fetchImpl, retries: 0 });
++ const events = createEventsClient(client, { source: 'live' });
++ outbox = createOutbox(db.getDb(), {
++ events,
++ intervalMs: intervalMs || 2000,
++ onError: (err) => {
++ const msg = err && err.message;
++ if (msg !== stats.lastError) console.warn('[Events] publish failed (will retry):', msg);
++ stats.lastError = msg;
++ },
++ });
++ outbox.ensureSchema();
++ db.onStreamLifecycle((kind, streamId) => {
++ const env = envelopeFor(kind, streamId);
++ if (!env) return;
++ outbox.enqueue(env);
++ stats.queued++;
++ setImmediate(() => outbox && outbox.kick());
++ });
++ outbox.start();
++ const prune = setInterval(() => { try { outbox.prune(); } catch { /* next time */ } }, PRUNE_EVERY_MS);
++ if (prune.unref) prune.unref();
++ console.log(`[Events] stream lifecycle → ${eventsUrl} (${outbox.pending()} pending)`);
++ return outbox;
++}
++
++function status() {
++ if (!outbox) return { enabled: false };
++ return { enabled: true, pending: outbox.pending(), rejected: outbox.rejected(), queued_since_boot: stats.queued, last_error: stats.lastError };
++}
++
++function _reset() { if (outbox) outbox.stop(); outbox = null; db.onStreamLifecycle(null); stats.queued = 0; stats.lastError = null; }
++
++module.exports = { init, status, envelopeFor, _reset };
ddiff --git a/server/index.js b/server/index.js
iindex f4073845..134d3da2 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -1471,6 +1471,10 @@ async function start() {
require('./utils/jobs').every('identity-legacy-sync', 24 * 60 * 60 * 1000, () => require('./auth/identity-sync').syncLegacyMap(),
{ initialDelayMs: 3 * 60 * 1000, jitterMs: 60 * 1000 });
++ // 8c. Durable events (roadmap Wave 3): stream lifecycle goes to OpenVibe.Events through the
++ // transactional outbox (server/events/stream-events.js). Off unless EVENTS_URL is set.
++ try { require('./events/stream-events').init(); } catch (err) { console.warn('[Events] not started:', err.message); }
++
// 9. Periodic registry refresh — re-syncs config with openvibe.network every 5 minutes.
// This is a safety net: if the startup refresh failed (openvibe.network was temporarily
// unreachable), subsequent refreshes will fix CORS, issuer, and other URL config.
ddiff --git a/test/stream-events.test.js b/test/stream-events.test.js
nnew file mode 100644
iindex 00000000..41f499ba
---- /dev/null
++++ b/test/stream-events.test.js
@@@ -0,0 +1,116 @@
++'use strict';
++
++// Stream lifecycle → OpenVibe.Events (roadmap Wave 3): createStream/endStream enqueue
++// live.stream.started / live.stream.ended in the same transaction as the streams row, and the
++// relay publishes them with a service token (stub Network + stub Events).
++
++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-stream-events-'));
++process.env.DB_PATH = path.join(tmp, 'live.db');
++
++const published = [];
++const mode = { down: false };
++const stub = http.createServer((req, res) => {
++ let body = '';
++ req.on('data', (c) => { body += c; });
++ req.on('end', () => {
++ res.setHeader('Content-Type', 'application/json');
++ if (req.url === '/oauth/token') {
++ const p = new URLSearchParams(body);
++ assert.strictEqual(p.get('grant_type'), 'client_credentials');
++ assert.strictEqual(p.get('audience'), 'openvibe.events');
++ return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));
++ }
++ if (req.url === '/api/v1/events' && req.method === 'POST') {
++ if (mode.down) { res.statusCode = 503; return res.end('{"code":"events.unavailable"}'); }
++ assert.strictEqual(req.headers.authorization, 'Bearer tok');
++ const parsed = JSON.parse(body);
++ const list = parsed.events || [parsed];
++ const results = list.map((e) => { published.push(e); return { event_id: e.event_id, seq: published.length, duplicate: false }; });
++ return res.end(JSON.stringify(parsed.events ? { results } : results[0]));
++ }
++ res.statusCode = 404; res.end('{}');
++ });
++});
++
++(async () => {
++ await new Promise((r) => stub.listen(0, '127.0.0.1', r));
++ const base = `http://127.0.0.1:${stub.address().port}`;
++ process.env.OV_NETWORK_INTERNAL_URL = base;
++
++ const db = require('../server/db/database');
++ db.initDb();
++ const d = db.getDb();
++ d.prepare("INSERT INTO users (id, username, display_name, password_hash) VALUES (501, 'streamer', 'Streamer', 'x')").run();
++ d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (501, 'network', '77', 'usr_01J0000000000000000000000Z')").run();
++
++ const streamEvents = require('../server/events/stream-events');
++
++ // Disabled without a secret or URL: going live works and nothing is queued.
++ assert.strictEqual(streamEvents.init({ eventsUrl: '', clientSecret: '' }), null);
++ const plain = db.createStream({ user_id: 501, title: 'no events', protocol: 'rtmp' });
++ assert.ok(plain.lastInsertRowid);
++ db.endStream(plain.lastInsertRowid);
++
++ const outbox = streamEvents.init({ eventsUrl: base, clientSecret: 's3cret', intervalMs: 50 });
++ assert.ok(outbox);
++
++ // Started: queued in the same transaction and published.
++ const started = db.createStream({ user_id: 501, title: 'Hello', category: 'irl', protocol: 'webrtc', is_nsfw: 0 });
++ const id = Number(started.lastInsertRowid);
++ await outbox.flush();
++ assert.strictEqual(published.length, 1);
++ const ev = published[0];
++ assert.strictEqual(ev.event_type, 'live.stream.started');
++ assert.strictEqual(ev.source, 'live');
++ assert.deepStrictEqual(ev.subject, { type: 'stream', id: String(id), revision: 1 });
++ assert.deepStrictEqual(ev.actor, { type: 'user', id: 'usr_01J0000000000000000000000Z' });
++ assert.strictEqual(ev.visibility, 'public');
++ assert.strictEqual(ev.payload.title, 'Hello');
++ assert.strictEqual(ev.payload.channel.username, 'streamer');
++ assert.match(ev.payload.started_at, /^\d{4}-\d\d-\d\dT/);
++ assert.match(ev.event_id, /^evt_[0-9A-HJKMNP-TV-Z]{26}$/);
++
++ // Ending a stream that is already ended emits nothing; ending a live one emits ended.
++ db.endStream(id);
++ db.endStream(id);
++ await outbox.flush();
++ assert.strictEqual(published.length, 2);
++ assert.strictEqual(published[1].event_type, 'live.stream.ended');
++ assert.strictEqual(published[1].subject.revision, 2);
++ assert.ok('duration_seconds' in published[1].payload);
++
++ // A rolled-back go-live leaves no event.
++ assert.throws(() => d.transaction(() => { db.createStream({ user_id: 501, title: 'rolled back' }); throw new Error('abort'); })());
++ assert.strictEqual(outbox.pending(), 0);
++
++ // Events down: the stream still goes live; the event waits and is published once Events is back.
++ mode.down = true;
++ const later = db.createStream({ user_id: 501, title: 'while down' });
++ assert.ok(later.lastInsertRowid);
++ await outbox.flush();
++ assert.strictEqual(outbox.pending(), 1);
++ mode.down = false;
++ d.prepare('UPDATE event_outbox SET next_attempt_at = 0').run();
++ await outbox.flush();
++ assert.strictEqual(outbox.pending(), 0);
++ assert.strictEqual(published[2].payload.title, 'while down');
++
++ // A hook failure never blocks going live.
++ d.exec('ALTER TABLE event_outbox RENAME TO event_outbox_moved');
++ const log = console.warn; console.warn = () => {};
++ const safe = db.createStream({ user_id: 501, title: 'hook broken' });
++ console.warn = log;
++ assert.ok(safe.lastInsertRowid);
++ d.exec('ALTER TABLE event_outbox_moved RENAME TO event_outbox');
++
++ assert.strictEqual(streamEvents.status().enabled, true);
++ streamEvents._reset();
++ stub.close();
++ console.log('✅ stream lifecycle events: outbox, relay, retry, rollback');
++})().catch((err) => { console.error(err); process.exit(1); });