openvibers/openrestream Commits

Session lineage from Live's canonical resolver (D20): the coordinator asks OpenVibe.Live's /internal/lineage/resolve (live.lineage.resolve, OpenRe's service token) which channel each Live-linked definition belongs to (mirror_to_live or a live:managed_stream ref; owner subject and slot id, never a display name), keeps the answer in definition_lineage (refreshed every 15 minutes, 20 a tick, stops for the tick when Live is down), and openre.session.* events carry payload.lineage { channel, rule } once resolved. OPENRE_LINEAGE=off disables it. test/lineage.test.js

OP OpenVibers committed 24 Sept 2026 · ef21212930 · 6 files, +166 −4


Changes

ddiff --git a/server/coordinator.js b/server/coordinator.js
iindex 62d85d9..3ee5ab4 100644
---- a/server/coordinator.js
++++ b/server/coordinator.js
@@@ -19,7 +19,7 @@
const { KINDS } = require('./store/workers');
const { assertNotDrill } = require('./config');
--function createCoordinator({ rt, media, log = console, holder = `coordinator:${process.pid}` }) {
++function createCoordinator({ rt, media, lineage = null, log = console, holder = `coordinator:${process.pid}` }) {
const { db, store, config, clock } = rt;
assertNotDrill(config, 'The session coordinator');
const now = () => clock.now();
@@@ -96,6 +96,7 @@ function createCoordinator({ rt, media, log = console, holder = `coordinator:${p
if (!takeLease()) return { skipped: 'lease held by another coordinator' };
const out = syncTick();
if (media) out.recordingSteps = await store.recordings.process(media);
++ if (lineage) out.lineage = await lineage.refresh();
stats.ticks++;
stats.lastTickAt = now();
stats.lastError = null;
@@@ -125,7 +126,7 @@ function createCoordinator({ rt, media, log = console, holder = `coordinator:${p
await rt.events.stop();
lease.release.run(holder);
},
-- stats: () => ({ ...stats }),
++ stats: () => ({ ...stats, lineage: lineage ? lineage.stats() : null }),
};
}
ddiff --git a/server/db.js b/server/db.js
iindex b8d0b45..4579510 100644
---- a/server/db.js
++++ b/server/db.js
@@@ -50,6 +50,14 @@ CREATE TABLE IF NOT EXISTS external_refs (
);
CREATE INDEX IF NOT EXISTS idx_external_refs_ref ON external_refs(service, type, ref_id);
++-- The Live channel a definition belongs to, as OpenVibe.Live's canonical resolver answers it
++-- (lineage.resolution@1, roadmap D20). Refreshed by the coordinator; session events carry it.
++CREATE TABLE IF NOT EXISTS definition_lineage (
++ definition_id TEXT PRIMARY KEY REFERENCES stream_definitions(id) ON DELETE CASCADE,
++ resolution TEXT NOT NULL,
++ checked_at INTEGER NOT NULL
++);
++
-- Ingest keys are stored as SHA-256 of a 256-bit random secret; the secret is shown once.
CREATE TABLE IF NOT EXISTS ingest_keys (
id TEXT PRIMARY KEY,
ddiff --git a/server/lineage.js b/server/lineage.js
nnew file mode 100644
iindex 0000000..1b49702
---- /dev/null
++++ b/server/lineage.js
@@@ -0,0 +1,65 @@
++'use strict';
++/**
++ * Live lineage for OpenRe definitions (roadmap D20): which OpenVibe.Live channel a stream belongs to,
++ * answered by Live's canonical resolver (GET /internal/lineage/resolve, capability live.lineage.resolve,
++ * OpenRe's Network service token for audience openvibe.live) instead of a mapping of OpenRe's own.
++ * A definition is asked about when it mirrors into Live or points at a Live slot (external ref
++ * live:managed_stream), with its owner subject and that slot id; display names are never sent.
++ *
++ * The coordinator calls refresh() on its tick: at most `batch` definitions whose answer is missing or
++ * older than `maxAgeMs`, stored in definition_lineage and carried by session events (payload.lineage).
++ * Off without OV_OAUTH_CLIENT_SECRET, or with OPENRE_LINEAGE=off.
++ */
++const { createServiceTokenClient } = require('openvibe-sdk/auth');
++
++function createLineage({ db, config, env = process.env, fetchImpl = globalThis.fetch, now = () => Date.now(), maxAgeMs = 15 * 60 * 1000, batch = 20, log = console }) {
++ const base = String(env.OV_LIVE_INTERNAL_URL || 'http://127.0.0.1:3000').replace(/\/+$/, '');
++ const enabled = Boolean(config.oauth && config.oauth.clientSecret) && env.OPENRE_LINEAGE !== 'off' && !config.drill;
++ const tokens = enabled ? createServiceTokenClient({ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, audience: 'openvibe.live', fetch: fetchImpl }) : null;
++ const stats = { refreshed: 0, resolved: 0, unresolved: 0, failed: 0, lastError: null };
++ const q = {
++ due: db.prepare(`SELECT d.id, d.owner_subject,
++ (SELECT r.ref_id FROM external_refs r WHERE r.definition_id = d.id AND r.service = 'live' AND r.type = 'managed_stream' LIMIT 1) AS slot_id
++ FROM stream_definitions d LEFT JOIN definition_lineage l ON l.definition_id = d.id
++ WHERE d.state = 'active'
++ AND (d.mirror_to_live = 1 OR EXISTS (SELECT 1 FROM external_refs r WHERE r.definition_id = d.id AND r.service = 'live'))
++ AND (l.checked_at IS NULL OR l.checked_at < ?)
++ ORDER BY l.checked_at IS NOT NULL, l.checked_at LIMIT ?`),
++ save: db.prepare(`INSERT INTO definition_lineage (definition_id, resolution, checked_at) VALUES (?, ?, ?)
++ ON CONFLICT(definition_id) DO UPDATE SET resolution = excluded.resolution, checked_at = excluded.checked_at`),
++ };
++
++ async function resolveOne(def) {
++ const qs = new URLSearchParams();
++ if (def.owner_subject) qs.set('owner_subject', def.owner_subject);
++ if (def.slot_id && /^[0-9]{1,15}$/.test(def.slot_id)) qs.set('slot_id', def.slot_id);
++ const res = await fetchImpl(`${base}/internal/lineage/resolve?${qs}`, { headers: { Accept: 'application/json', Authorization: `Bearer ${await tokens.getToken()}` }, signal: AbortSignal.timeout(10000) });
++ if (res.status === 401 && tokens.invalidate) tokens.invalidate();
++ const body = await res.json().catch(() => null);
++ if (!res.ok || !body || (body.status !== 'resolved' && body.status !== 'unresolved')) throw new Error(`resolver answered ${res.status}`);
++ return body;
++ }
++
++ async function refresh() {
++ if (!enabled) return { skipped: 'off' };
++ const out = { checked: 0, resolved: 0 };
++ for (const def of q.due.all(now() - maxAgeMs, batch)) {
++ try {
++ const r = await resolveOne(def);
++ q.save.run(def.id, JSON.stringify(r), now());
++ out.checked++; stats.refreshed++;
++ if (r.status === 'resolved') { out.resolved++; stats.resolved++; } else stats.unresolved++;
++ } catch (err) {
++ stats.failed++;
++ if (err.message !== stats.lastError) log.warn(`[lineage] ${def.id}: ${err.message}`);
++ stats.lastError = err.message;
++ break; // Live is down or refusing: try again next tick
++ }
++ }
++ return out;
++ }
++
++ return { enabled, refresh, stats: () => ({ enabled, ...stats }) };
++}
++
++module.exports = { createLineage };
ddiff --git a/server/store/sessions.js b/server/store/sessions.js
iindex 7ab5032..ac15e5b 100644
---- a/server/store/sessions.js
++++ b/server/store/sessions.js
@@@ -33,8 +33,17 @@ function createSessions({ db, config, events, clock, definitions, workers }) {
ofWorker: db.prepare(`SELECT * FROM ingest_sessions WHERE worker_id = ? AND state IN ${OPEN}`),
expiredLeases: db.prepare(`SELECT * FROM ingest_sessions WHERE state IN ${OPEN} AND lease_expires_at < ?`),
live: db.prepare("SELECT * FROM ingest_sessions WHERE state = 'live' ORDER BY live_at"),
++ lineage: db.prepare('SELECT resolution FROM definition_lineage WHERE definition_id = ?'),
};
++ /** The Live channel this definition belongs to (the resolver's answer), or null while unresolved. */
++ function liveLineage(definitionId) {
++ const row = q.lineage.get(definitionId);
++ const r = row ? parseJson(row.resolution, null) : null;
++ if (!r || r.status !== 'resolved' || !r.channel) return null;
++ return { channel: { service: 'live', id: String(r.channel.id), slug: r.channel.slug, owner_subject: r.channel.owner_subject || null }, rule: r.rule || null };
++ }
++
function shape(s) {
if (!s) return null;
return { ...s, media_info: parseJson(s.media_info, null) };
@@@ -70,6 +79,7 @@ function createSessions({ db, config, events, clock, definitions, workers }) {
worker: worker ? { id: worker.id, kind: worker.kind, generation: worker.generation } : null,
external_refs: definition.external_refs,
mirror_to_live: definition.mirror_to_live,
++ lineage: liveLineage(definition.id),
...extra,
};
return {
@@@ -207,6 +217,7 @@ function createSessions({ db, config, events, clock, definitions, workers }) {
return {
get,
++ liveLineage,
list,
admit,
transition,
ddiff --git a/test/lineage.test.js b/test/lineage.test.js
nnew file mode 100644
iindex 0000000..b9f0b4a
---- /dev/null
++++ b/test/lineage.test.js
@@@ -0,0 +1,76 @@
++'use strict';
++// Live lineage (D20): the coordinator asks OpenVibe.Live's canonical resolver which channel a Live-linked
++// definition belongs to, with OpenRe's service token; session events carry the answer.
++const assert = require('assert');
++const { runtime, manualClock, suite, OWNER, outboxEnvelopes, silent } = require('./helpers');
++const { createCoordinator } = require('../server/coordinator');
++const { createLineage } = require('../server/lineage');
++
++const t = suite('lineage');
++
++function setup(answer) {
++ const clock = manualClock();
++ const rt = runtime({ clock });
++ const asked = [];
++ const fetchImpl = async (url, opts = {}) => {
++ const u = new URL(url);
++ const reply = (status, body) => ({ status, ok: status < 300, json: async () => body });
++ if (u.pathname === '/oauth/token') return reply(200, { access_token: 'svc-openre', expires_in: 300 });
++ assert.strictEqual(u.pathname, '/internal/lineage/resolve');
++ assert.strictEqual(opts.headers.Authorization, 'Bearer svc-openre');
++ asked.push(Object.fromEntries(u.searchParams));
++ return typeof answer === 'function' ? answer(u) : reply(200, answer);
++ };
++ const config = { ...rt.config, oauth: { ...rt.config.oauth, clientSecret: 'openre-secret-for-tests' } };
++ const lineage = createLineage({ db: rt.db, config, env: { OV_LIVE_INTERNAL_URL: 'http://live.test' }, fetchImpl, now: () => clock.now(), log: silent });
++ const coordinator = createCoordinator({ rt, media: null, lineage, log: silent });
++ return { rt, clock, lineage, coordinator, asked };
++}
++const RESOLVED = { status: 'resolved', channel: { id: '17', slug: 'goosely', owner_subject: OWNER }, rule: 'slot', confidence: 'exact' };
++
++t('a Live-linked definition is resolved with its owner and slot; an unlinked one is never asked about', async () => {
++ const { rt, lineage, asked } = setup(RESOLVED);
++ const linked = rt.store.definitions.create({ owner_subject: OWNER, mirror_to_live: true, external_refs: [{ service: 'live', type: 'managed_stream', id: '12' }] }).definition;
++ rt.store.definitions.create({ owner_subject: OWNER, title: 'OpenRe only' });
++ const out = await lineage.refresh();
++ assert.deepStrictEqual(out, { checked: 1, resolved: 1 });
++ assert.deepStrictEqual(asked, [{ owner_subject: OWNER, slot_id: '12' }], 'owner subject and slot id, never a display name');
++ const row = rt.db.prepare('SELECT resolution FROM definition_lineage WHERE definition_id = ?').get(linked.id);
++ assert.strictEqual(JSON.parse(row.resolution).channel.slug, 'goosely');
++ assert.deepStrictEqual(await lineage.refresh(), { checked: 0, resolved: 0 }, 'fresh answers are not asked again');
++});
++
++t('session events carry the resolved channel', async () => {
++ const { rt, coordinator } = setup(RESOLVED);
++ const { definition, key } = rt.store.definitions.create({ owner_subject: OWNER, mirror_to_live: true, external_refs: [{ service: 'live', type: 'managed_stream', id: '12' }] });
++ await coordinator.tick();
++ const w = rt.store.workers.register({ kind: 'rtmp-ingest', endpoints: { flvPort: 19361, rtmpPlayPort: 19360 } }); rt.store.workers.ready(w.id);
++ const a = rt.store.sessions.admit({ definition, key, protocol: 'rtmp', worker: rt.store.workers.get(w.id) });
++ rt.store.sessions.transition(a.session.id, 'live');
++ const started = outboxEnvelopes(rt.db).find((e) => e.event_type === 'openre.session.started');
++ assert.deepStrictEqual(started.payload.lineage, { channel: { service: 'live', id: '17', slug: 'goosely', owner_subject: OWNER }, rule: 'slot' });
++});
++
++t('unresolved answers are kept but not carried; a stale answer is asked again; Live down waits for the next tick', async () => {
++ let mode = 'unresolved';
++ const { rt, clock, lineage } = setup((u) => (mode === 'down' ? { status: 503, ok: false, json: async () => ({}) } : { status: 200, ok: true, json: async () => (mode === 'unresolved' ? { status: 'unresolved', reason: 'no_match' } : RESOLVED) }));
++ const { definition } = rt.store.definitions.create({ owner_subject: OWNER, mirror_to_live: true });
++ assert.deepStrictEqual(await lineage.refresh(), { checked: 1, resolved: 0 });
++ assert.strictEqual(rt.store.sessions.liveLineage(definition.id), null, 'an unresolved answer is not carried');
++ mode = 'down';
++ clock.advance(16 * 60 * 1000);
++ assert.deepStrictEqual(await lineage.refresh(), { checked: 0, resolved: 0 });
++ assert.strictEqual(lineage.stats().failed, 1);
++ mode = 'resolved';
++ assert.deepStrictEqual(await lineage.refresh(), { checked: 1, resolved: 1 });
++});
++
++t('off without a service secret', async () => {
++ const clock = manualClock();
++ const rt = runtime({ clock });
++ const off = createLineage({ db: rt.db, config: { ...rt.config, oauth: { ...rt.config.oauth, clientSecret: '' } }, env: {}, fetchImpl: () => { throw new Error('no calls'); }, log: silent });
++ assert.strictEqual(off.enabled, false);
++ assert.deepStrictEqual(await off.refresh(), { skipped: 'off' });
++});
++
++t.run();
ddiff --git a/workers/coordinator.js b/workers/coordinator.js
iindex 9c786f2..cff089f 100644
---- a/workers/coordinator.js
++++ b/workers/coordinator.js
@@@ -18,9 +18,10 @@ if (require.main === module) {
exitIfDrill(config, 'openre-session-coordinator');
const rt = openRuntime({ config });
const media = createMediaClient({ config });
-- const coordinator = createCoordinator({ rt, media });
++ const lineage = require('../server/lineage').createLineage({ db: rt.db, config });
++ const coordinator = createCoordinator({ rt, media, lineage });
coordinator.start();
-- console.log(`[coordinator] running (events relay ${rt.events.configured ? 'on' : 'off: EVENTS_URL/OV_OAUTH_CLIENT_SECRET not set'}, recordings ${media.configured ? `on → ${config.media.url} app ${config.media.appId}` : 'off'})`);
++ console.log(`[coordinator] running (events relay ${rt.events.configured ? 'on' : 'off: EVENTS_URL/OV_OAUTH_CLIENT_SECRET not set'}, recordings ${media.configured ? `on → ${config.media.url} app ${config.media.appId}` : 'off'}, Live lineage ${lineage.enabled ? 'on' : 'off'})`);
const prune = setInterval(() => { try { rt.events.outbox.prune(); } catch { /* next time */ } }, 6 * 60 * 60 * 1000);
prune.unref();
const shutdown = (sig) => {