openvibers/openvibe.media Commits

Media outcomes as durable events: vod/clip ready|failed, object uploaded and storage alerts also go to OpenVibe.Events through an outbox (openvibe-sdk 0.2.2); webhooks unchanged; openvibe-contracts v0.8.0

OP OpenVibers committed 23 Sept 2026 · 1122f527e1 · 8 files, +219 −5


Changes

ddiff --git a/.env.example b/.env.example
iindex 3a60647..1a49df9 100644
---- a/.env.example
++++ b/.env.example
@@@ -57,3 +57,7 @@ MEDIA_R2_SECRET_ACCESS_KEY=
# VOD_DISK_CRIT_GB=5
# MAX_VOD_SIZE_MB=2048
# VOD_NO_NORMALIZE=0
++
++# OpenVibe.Events (durable events): vod/clip/object/storage outcomes are also published through
++# Media's outbox with the OAuth client credentials (grant events.event.publish). Unset = off.
++#EVENTS_URL=http://127.0.0.1:4300
ddiff --git a/package-lock.json b/package-lock.json
iindex cdc166f..a308ef0 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -16,7 +16,8 @@
"dotenv": "^16.4.5",
"express": "^4.19.2",
"multer": "^1.4.5-lts.1",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.7.0",
++ "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",
"sharp": "^0.33.4"
}
@@@ -1918,9 +1919,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.7.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.7.0",
-- "integrity": "sha512-gUrjxbTy/HX0QMQGoDPphmFdwm7iitmNMMchs9tDSDGR47FRFHL/RgBgIKYR1S7M6ICu261YycFV8+YVglafXA==",
++ "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",
@@@ -1933,6 +1934,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 ac901f9..0a3ff3a 100644
---- a/package.json
++++ b/package.json
@@@ -11,7 +11,8 @@
},
"license": "MIT",
"dependencies": {
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.7.0",
++ "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",
"@aws-sdk/client-s3": "^3.600.0",
"@aws-sdk/lib-storage": "^3.600.0",
"@aws-sdk/s3-request-presigner": "^3.600.0",
ddiff --git a/server/events.js b/server/events.js
nnew file mode 100644
iindex 0000000..d5f1ff1
---- /dev/null
++++ b/server/events.js
@@@ -0,0 +1,110 @@
++'use strict';
++/**
++ * Media → OpenVibe.Events (roadmap Wave 3, ADR-004): the outcomes Media already announces by
++ * per-app webhook are also written as durable events to Media's own `event_outbox` table and
++ * relayed to OpenVibe.Events with Media's service token (audience openvibe.events, capability
++ * events.event.publish). The webhooks stay as they are until every consumer reads events.
++ *
++ * vod.ready | vod.failed → media.vod.ready | media.vod.failed (subject vod <id>)
++ * clip.ready | clip.failed → media.clip.ready | media.clip.failed (subject clip <id>)
++ * media.object.uploaded → media.object.uploaded (subject object <id>)
++ * storage.alert | storage.recovered → media.storage.alert | media.storage.recovered
++ *
++ * The event is enqueued right after the state change commits (its own SQLite transaction, a local
++ * insert): the loss window is a process crash between those two statements. Delivery from the
++ * outbox on is at least once. Every event is `internal` visibility: a VOD's own visibility decides
++ * who may see it, and consumers (Live, Search) apply it.
++ *
++ * Off unless EVENTS_URL and OV_OAUTH_CLIENT_SECRET are set (EVENTS_PUBLISH=off disables 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 TYPES = {
++ 'vod.ready': ['media.vod.ready', 'vod'],
++ 'vod.failed': ['media.vod.failed', 'vod'],
++ 'clip.ready': ['media.clip.ready', 'clip'],
++ 'clip.failed': ['media.clip.failed', 'clip'],
++ 'media.object.uploaded': ['media.object.uploaded', 'object'],
++ 'storage.alert': ['media.storage.alert', 'storage'],
++ 'storage.recovered': ['media.storage.recovered', 'storage'],
++};
++
++let outbox = null;
++const stats = { queued: 0, lastError: null };
++
++function init({
++ eventsUrl = (process.env.EVENTS_URL || '').replace(/\/+$/, ''),
++ clientSecret = process.env.OV_OAUTH_CLIENT_SECRET || '',
++ clientId = process.env.OV_OAUTH_CLIENT_ID || 'media',
++ networkUrl = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, ''),
++ fetchImpl, intervalMs = 2000,
++} = {}) {
++ if (outbox) return outbox;
++ if (process.env.EVENTS_PUBLISH === 'off' || !eventsUrl || !clientSecret) return null;
++ const tokens = createServiceTokenClient({ tokenUrl: `${networkUrl}/oauth/token`, clientId, clientSecret, fetch: fetchImpl });
++ const client = createClient({ baseUrls: { events: eventsUrl }, tokenProvider: tokens, fetch: fetchImpl, retries: 0 });
++ outbox = createOutbox(db.getDb(), {
++ events: createEventsClient(client, { source: 'media' }),
++ intervalMs,
++ onError: (err) => {
++ const msg = err && err.message;
++ if (msg !== stats.lastError) console.warn('[Events] publish failed (will retry):', msg);
++ stats.lastError = msg;
++ },
++ });
++ outbox.ensureSchema();
++ outbox.start();
++ const prune = setInterval(() => { try { outbox.prune(); } catch { /* next time */ } }, 6 * 60 * 60 * 1000);
++ if (prune.unref) prune.unref();
++ console.log(`[Events] media outcomes → ${eventsUrl} (${outbox.pending()} pending)`);
++ return outbox;
++}
++
++// Events caps payloads (64 KB by default): the free-text fields stay behind the Media API.
++function slim(appId, data) {
++ const { meta, ai_overview, ...rest } = data || {};
++ return { app_id: appId || null, ...rest };
++}
++
++/** Queue the durable twin of a webhook event. Never throws; returns the envelope or null. */
++function emit(webhookEvent, appId, data) {
++ if (!outbox) return null;
++ const map = TYPES[webhookEvent];
++ if (!map) return null;
++ const [eventType, subjectType] = map;
++ const id = data && (data.object_id || data.id);
++ const subject = subjectType === 'storage'
++ ? { type: 'storage', id: String((data && data.kind) || 'storage') }
++ : { type: subjectType, id: String(id == null ? 'unknown' : id) };
++ try {
++ let env = null;
++ db.getDb().transaction(() => {
++ env = outbox.enqueue({
++ event_type: eventType,
++ actor: { type: 'service', id: 'media' },
++ subject,
++ visibility: 'internal',
++ priority: /failed|alert/.test(eventType) ? 'important' : 'low',
++ payload: slim(appId, data),
++ });
++ })();
++ stats.queued++;
++ setImmediate(() => outbox && outbox.kick());
++ return env;
++ } catch (err) {
++ console.warn(`[Events] ${eventType} not queued:`, err.message);
++ return null;
++ }
++}
++
++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; stats.queued = 0; stats.lastError = null; }
++
++module.exports = { init, emit, status, TYPES, _reset };
ddiff --git a/server/index.js b/server/index.js
iindex 26d3cb1..bbbc5a4 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -197,6 +197,8 @@ const server = app.listen(config.port, config.host, () => {
console.log(`[Media] OpenVibe.Media listening on ${config.host}:${config.port} (${config.nodeEnv})`);
console.log(`[Media] Data: db=${config.db.path} vods=${config.vod.path} clips=${config.vod.clipsPath}`);
console.log(`[Media] RTP ingest pool: udp ${config.rtp.portMin}-${config.rtp.portMax} (127.0.0.1)`);
++ // Durable events (roadmap Wave 3): webhook outcomes also go to OpenVibe.Events. Off without EVENTS_URL.
++ try { require('./events').init(); } catch (err) { console.warn('[Events] not started:', err.message); }
});
let shuttingDown = false;
@@@ -208,6 +210,7 @@ function shutdown(signal) {
try { vodStorage.stop(); } catch { /* */ }
try { healthJob.stop(); } catch { /* */ }
try { auth.stopJwksRefresh(); } catch { /* */ }
++ try { require('./events')._reset(); } catch { /* */ }
for (const t of timers) clearInterval(t);
server.close(() => {
try { db.close(); } catch { /* */ }
ddiff --git a/server/vod/vod-storage.js b/server/vod/vod-storage.js
iindex 6366c3f..7de3320 100644
---- a/server/vod/vod-storage.js
++++ b/server/vod/vod-storage.js
@@@ -751,6 +751,7 @@ async function emitStorageEvent(event, kind, data, settings = getSettings()) {
alertLastSentAt.set(cooldownKey, Date.now());
const payload = { kind, ...data, at: new Date().toISOString() };
(event === 'storage.alert' ? console.error : console.warn)(`[VodStorage] ${event} (${kind}): ${JSON.stringify(data)}`);
++ try { require('../events').emit(event, null, payload); } catch { /* never blocks the alert */ }
let apps = [];
try { apps = db.listApps().filter(a => a.webhook_url); } catch { apps = []; }
let webhooks;
ddiff --git a/server/webhooks.js b/server/webhooks.js
iindex 5bd6aff..d15260b 100644
---- a/server/webhooks.js
++++ b/server/webhooks.js
@@@ -41,6 +41,11 @@ async function _post(url, rawBody, signature) {
* @param {object} data event payload
*/
async function sendWebhook(appOrId, event, data) {
++ // The durable twin (OpenVibe.Events) is queued whether or not this app has a webhook URL.
++ // Storage alerts go to every app, so server/vod/vod-storage.js queues those once itself.
++ if (!String(event).startsWith('storage.')) {
++ try { require('./events').emit(event, typeof appOrId === 'string' ? appOrId : appOrId && appOrId.app_id, data); } catch { /* never blocks a webhook */ }
++ }
let app = appOrId;
if (typeof appOrId === 'string') {
try { app = db.getApp(appOrId); } catch { app = null; }
ddiff --git a/test/events.test.js b/test/events.test.js
nnew file mode 100644
iindex 0000000..347f5c9
---- /dev/null
++++ b/test/events.test.js
@@@ -0,0 +1,80 @@
++'use strict';
++// Media → OpenVibe.Events (roadmap Wave 3): webhook outcomes are also queued in the outbox and
++// relayed with a service token; storage alerts are queued once, not once per app; disabled
++// without EVENTS_URL. Stub Network token endpoint + 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-media-events-'));
++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
++process.env.DB_PATH = path.join(tmp, 'media.db');
++process.env.VOD_PATH = dir('vods');
++process.env.CLIPS_PATH = dir('clips');
++process.env.FILES_PATH = dir('files');
++process.env.THUMBNAILS_PATH = dir('thumbnails');
++process.env.PASTES_PATH = dir('pastes');
++process.env.OBJECTS_PATH = dir('objects');
++
++const published = [];
++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') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));
++ if (req.url === '/api/v1/events') {
++ 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}`;
++ const db = require('../server/db/database');
++ db.getDb();
++ const events = require('../server/events');
++ const { sendWebhook } = require('../server/webhooks');
++
++ assert.strictEqual(events.init({ eventsUrl: '', clientSecret: '' }), null, 'off without EVENTS_URL');
++ assert.strictEqual(events.emit('vod.ready', 'live', { id: 1 }), null);
++
++ const outbox = events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 50 });
++ assert.ok(outbox);
++
++ // A webhook to an app with no webhook URL still queues the durable event.
++ await sendWebhook('no-such-app', 'vod.ready', { id: 42, title: 'Stream', meta: { big: 'x'.repeat(100000) }, ai_overview: 'long text' });
++ await sendWebhook('no-such-app', 'clip.failed', { id: 7, status: 'failed' });
++ await sendWebhook('no-such-app', 'unknown.event', { id: 1 });
++ await outbox.flush();
++ assert.strictEqual(published.length, 2);
++ assert.strictEqual(published[0].event_type, 'media.vod.ready');
++ assert.deepStrictEqual(published[0].subject, { type: 'vod', id: '42' });
++ assert.strictEqual(published[0].source, 'media');
++ assert.strictEqual(published[0].visibility, 'internal');
++ assert.strictEqual(published[0].payload.app_id, 'no-such-app');
++ assert.ok(!('meta' in published[0].payload) && !('ai_overview' in published[0].payload), 'free text stays behind the API');
++ assert.strictEqual(published[1].event_type, 'media.clip.failed');
++ assert.strictEqual(published[1].priority, 'important');
++
++ // Storage alerts: queued once by the storage module, never per app by sendWebhook.
++ await sendWebhook('no-such-app', 'storage.alert', { kind: 'disk' });
++ await outbox.flush();
++ assert.strictEqual(published.length, 2);
++ events.emit('storage.alert', null, { kind: 'disk', free_gb: 3 });
++ await outbox.flush();
++ assert.strictEqual(published.length, 3);
++ assert.deepStrictEqual(published[2].subject, { type: 'storage', id: 'disk' });
++
++ assert.strictEqual(events.status().enabled, true);
++ events._reset();
++ stub.close();
++ console.log('✅ media events: outbox, relay, storage alerts once, off by default');
++})().catch((err) => { console.error(err); process.exit(1); });