openvibers/openvibe.media Commits
Media follows account merges (roadmap WS-B task 5, ADR-029). network.subject.merged arrives at POST /internal/events (the same signed, loopback-only endpoint as the revocations; ensureSubscription now keeps one subscription per topic) and server/subject-merge.js repoints media_objects.owner_subject from the folded-in subject to the survivor in one transaction, once per merge_id (subject_merges; source network only). VOD and clip rows name Live's user ids and follow Live's own consumer. test/revocations.test.js: both subscriptions created once, the objects moved, a redelivery moves nothing, another source ignored.
Changes
@@@ -7,12 +7,14 @@* MEDIA_INBOUND_EVENTS_SECRET (signature v2 only) and refused when it came through the proxy. A* delivery moves the person's cutoff (openvibe-sdk createRevocationStore: only ever forward, source* network only); user-auth.js then refuses their older tokens, so openvibe.media's navbar shows them-- * signed out at once. The subscription is created at boot when EVENTS_URL, the service secret and-- * MEDIA_INBOUND_EVENTS_SECRET are set (grant media events.subscription.manage).++ * signed out at once. The same endpoint takes network.subject.merged (ADR-029): a folded-in account's media objects++ * move to the survivor (server/subject-merge.js). The subscriptions (one per topic) are created at boot when++ * EVENTS_URL, the service secret and MEDIA_INBOUND_EVENTS_SECRET are set (grant media events.subscription.manage).*/const express = require('express');const TOPIC = 'network.user.token_valid_after';++const TOPICS = [TOPIC, 'network.subject.merged'];let store = null;function cutoffs() {if (!store) store = require('openvibe-sdk/auth').createRevocationStore(require('./db/database').getDb(), { table: 'token_revocations' });@@@ -34,13 +36,13 @@ function handler() {for (const k of keys) { d = parseDelivery(raw, req.headers, k, { requireV2: true }); if (d) break; }if (!d || !d.event) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }stats.received++;-- const outcome = cutoffs().apply(d.event);++ const outcome = d.event.event_type === 'network.subject.merged' ? require('./subject-merge').apply(d.event) : cutoffs().apply(d.event);if (outcome === 'revoked') stats.revoked++;res.json({ event_id: d.event.event_id || null, outcome });}];}--/** Create Media's subscription to TOPIC if missing (idempotent; an operator-disabled one is left alone). */++/** Create Media's subscriptions to TOPICS where missing (idempotent; an operator-disabled one is left alone). */async function ensureSubscription({ fetchImpl = globalThis.fetch, log = console } = {}) {const eventsUrl = String(process.env.EVENTS_URL || '').replace(/\/+$/, '');const secret = secrets()[0];@@@ -57,12 +59,16 @@ async function ensureSubscription({ fetchImpl = globalThis.fetch, log = consoleconst list = await fetchImpl(`${eventsUrl}/api/v1/subscriptions`, { headers });if (!list.ok) throw new Error(`Events answered ${list.status} listing subscriptions`);const subs = ((await list.json()).subscriptions || []);-- if (subs.some((s) => s.topic_pattern === TOPIC && s.endpoint === endpoint)) return 'exists';-- const r = await fetchImpl(`${eventsUrl}/api/v1/subscriptions`, { method: 'POST', headers: { ...headers, 'content-type': 'application/json' }, body: JSON.stringify({ topic_pattern: TOPIC, endpoint, secret }) });-- if (r.status === 409) return 'exists';-- if (!r.ok) throw new Error(`Events answered ${r.status} creating the ${TOPIC} subscription`);-- log.log && log.log(`[Events] subscription created: ${TOPIC} → ${endpoint}`);-- return 'created';++ let created = 0;++ for (const topic of TOPICS) {++ if (subs.some((s) => s.topic_pattern === topic && s.endpoint === endpoint)) continue;++ const r = await fetchImpl(`${eventsUrl}/api/v1/subscriptions`, { method: 'POST', headers: { ...headers, 'content-type': 'application/json' }, body: JSON.stringify({ topic_pattern: topic, endpoint, secret }) });++ if (r.status === 409) continue;++ if (!r.ok) throw new Error(`Events answered ${r.status} creating the ${topic} subscription`);++ log.log && log.log(`[Events] subscription created: ${topic} → ${endpoint}`);++ created++;++ }++ return created ? 'created' : 'exists';}--module.exports = { handler, ensureSubscription, isRevoked, stats, TOPIC, _cutoffs: cutoffs };++module.exports = { handler, ensureSubscription, isRevoked, stats, TOPIC, TOPICS, _cutoffs: cutoffs };nnew file mode 100644@@@ -0,0 +1,38 @@++'use strict';++/**++ * network.subject.merged → Media (roadmap WS-B task 5, ADR-029; Contracts 0.69.0). Two Network accounts became one:++ * `from` is an alias of `into`. Media repoints what it owns by subject: media_objects.owner_subject from `from` to++ * `into`, in one transaction, once per merge_id (subject_merges). VOD and clip rows name Live's user ids, which Live's++ * own consumer handles. Arrives at POST /internal/events (server/revocations.js), signed and loopback-only.++ */++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const MERGE_RE = /^mrg_[0-9A-HJKMNP-TV-Z]{26}$/;++const TOPIC = 'network.subject.merged';++++function ensureSchema(db) {++ db.exec(`CREATE TABLE IF NOT EXISTS subject_merges (++ merge_id TEXT PRIMARY KEY,++ from_subject TEXT NOT NULL,++ into_subject TEXT NOT NULL,++ objects INTEGER NOT NULL DEFAULT 0,++ applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))++ )`);++}++++/** One envelope → 'merged' | 'unchanged' | 'ignored:<why>'. */++function apply(ev, { db = require('./db/database').getDb() } = {}) {++ if (!ev || ev.event_type !== TOPIC) return 'ignored:type';++ if (ev.source !== 'network') return 'ignored:source';++ const p = ev.payload && typeof ev.payload === 'object' ? ev.payload : {};++ if (!MERGE_RE.test(String(p.merge_id || '')) || !SUBJECT_RE.test(String(p.from || '')) || !SUBJECT_RE.test(String(p.into || '')) || p.from === p.into) return 'ignored:payload';++ ensureSchema(db);++ return db.transaction(() => {++ if (db.prepare('SELECT 1 FROM subject_merges WHERE merge_id = ?').get(p.merge_id)) return 'unchanged';++ const objects = db.prepare('UPDATE media_objects SET owner_subject = ? WHERE owner_subject = ?').run(p.into, p.from).changes;++ db.prepare('INSERT INTO subject_merges (merge_id, from_subject, into_subject, objects) VALUES (?, ?, ?, ?)').run(p.merge_id, p.from, p.into, objects);++ console.log(`[Merge] ${p.merge_id}: ${objects} object(s) now owned by the survivor`);++ return 'merged';++ })();++}++++module.exports = { apply, ensureSchema, TOPIC };@@@ -45,6 +45,18 @@ const ev = (over = {}) => ({ event_id: 'evt_01J8Z3Q4R5S6T7V8W9X0Y1Z2A3', event_tassert.strictEqual(revocations.isRevoked(old), true, 'older tokens are refused');assert.strictEqual(revocations.isRevoked({ subject_id: SUBJECT, iat: at / 1000 + 5 }), false, 'newer ones are fine');assert.strictEqual((await post(ev())).body.outcome, 'unchanged', 'a redelivery');++++ // network.subject.merged (ADR-029): the folded-in account's objects are the survivor's, once per merge.++ const db = require('../server/db/database').getDb();++ const FROM = 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2B4', INTO = 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2C5';++ const obj = db.prepare("INSERT INTO media_objects (id, app_id, namespace, kind, owner_subject) VALUES (?, 'live', 'live', 'file', ?)");++ obj.run('med_01J8Z3Q4R5S6T7V8W9X0Y1Z2D6', FROM); obj.run('med_01J8Z3Q4R5S6T7V8W9X0Y1Z2E7', FROM); obj.run('med_01J8Z3Q4R5S6T7V8W9X0Y1Z2F8', SUBJECT);++ const merged = (over = {}) => ev({ event_id: 'evt_01J8Z3Q4R5S6T7V8W9X0Y1Z2G9', event_type: 'network.subject.merged', payload: { merge_id: 'mrg_01J8Z3Q4R5S6T7V8W9X0Y1Z2H0', from: FROM, into: INTO, merged_at: '2026-09-27T21:00:00.000Z', initiated_by: 'person' }, ...over });++ assert.strictEqual((await post(merged({ source: 'live' }))).body.outcome, 'ignored:source');++ assert.strictEqual((await post(merged())).body.outcome, 'merged');++ assert.deepStrictEqual(db.prepare('SELECT owner_subject AS o, COUNT(*) AS n FROM media_objects GROUP BY o ORDER BY o').all().map((r) => [r.o, r.n]), [[SUBJECT, 1], [INTO, 2]].sort());++ assert.strictEqual((await post(merged())).body.outcome, 'unchanged', 'a redelivery moves nothing');++ assert.strictEqual(db.prepare('SELECT objects FROM subject_merges').get().objects, 2);const src = fs.readFileSync(path.join(__dirname, '../server/user-auth.js'), 'utf8');assert.ok(/require\('\.\/revocations'\)\.isRevoked\(claims\)/.test(src), 'the site token check consults it');@@@ -58,8 +70,7 @@ const ev = (over = {}) => ({ event_id: 'evt_01J8Z3Q4R5S6T7V8W9X0Y1Z2A3', event_t};assert.strictEqual(await revocations.ensureSubscription({ fetchImpl: fake, log: { log() {} } }), 'created');assert.strictEqual(await revocations.ensureSubscription({ fetchImpl: fake, log: { log() {} } }), 'exists');-- assert.strictEqual(subs.length, 1);-- assert.strictEqual(subs[0].topic_pattern, revocations.TOPIC);++ assert.deepStrictEqual(subs.map((s) => s.topic_pattern), [revocations.TOPIC, 'network.subject.merged'], 'one subscription per topic, created once');} finally {srv.close();fs.rmSync(tmp, { recursive: true, force: true });
@@ -7,12 +7,14 @@
@@ -7,12 +7,14 @@
* MEDIA_INBOUND_EVENTS_SECRET (signature v2 only) and refused when it came through the proxy. A
* MEDIA_INBOUND_EVENTS_SECRET (signature v2 only) and refused when it came through the proxy. A
* delivery moves the person's cutoff (openvibe-sdk createRevocationStore: only ever forward, source
* delivery moves the person's cutoff (openvibe-sdk createRevocationStore: only ever forward, source
* network only); user-auth.js then refuses their older tokens, so openvibe.media's navbar shows them
* network only); user-auth.js then refuses their older tokens, so openvibe.media's navbar shows them
- * signed out at once. The subscription is created at boot when EVENTS_URL, the service secret and
- * MEDIA_INBOUND_EVENTS_SECRET are set (grant media events.subscription.manage).
+ * signed out at once. The same endpoint takes network.subject.merged (ADR-029): a folded-in account's media objects
+ * move to the survivor (server/subject-merge.js). The subscriptions (one per topic) are created at boot when
+ * EVENTS_URL, the service secret and MEDIA_INBOUND_EVENTS_SECRET are set (grant media events.subscription.manage).
*/
*/
const express = require('express');
const express = require('express');
const TOPIC = 'network.user.token_valid_after';
const TOPIC = 'network.user.token_valid_after';
+const TOPICS = [TOPIC, 'network.subject.merged'];
let store = null;
let store = null;
function cutoffs() {
function cutoffs() {
if (!store) store = require('openvibe-sdk/auth').createRevocationStore(require('./db/database').getDb(), { table: 'token_revocations' });
if (!store) store = require('openvibe-sdk/auth').createRevocationStore(require('./db/database').getDb(), { table: 'token_revocations' });
@@ -34,13 +36,13 @@ function handler() {
@@ -34,13 +36,13 @@ function handler() {
for (const k of keys) { d = parseDelivery(raw, req.headers, k, { requireV2: true }); if (d) break; }
for (const k of keys) { d = parseDelivery(raw, req.headers, k, { requireV2: true }); if (d) break; }
if (!d || !d.event) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }
if (!d || !d.event) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }
stats.received++;
stats.received++;
- const outcome = cutoffs().apply(d.event);
+ const outcome = d.event.event_type === 'network.subject.merged' ? require('./subject-merge').apply(d.event) : cutoffs().apply(d.event);
if (outcome === 'revoked') stats.revoked++;
if (outcome === 'revoked') stats.revoked++;
res.json({ event_id: d.event.event_id || null, outcome });
res.json({ event_id: d.event.event_id || null, outcome });
}];
}];
}
}
-/** Create Media's subscription to TOPIC if missing (idempotent; an operator-disabled one is left alone). */
+/** Create Media's subscriptions to TOPICS where missing (idempotent; an operator-disabled one is left alone). */
async function ensureSubscription({ fetchImpl = globalThis.fetch, log = console } = {}) {
async function ensureSubscription({ fetchImpl = globalThis.fetch, log = console } = {}) {
const eventsUrl = String(process.env.EVENTS_URL || '').replace(/\/+$/, '');
const eventsUrl = String(process.env.EVENTS_URL || '').replace(/\/+$/, '');
const secret = secrets()[0];
const secret = secrets()[0];
@@ -57,12 +59,16 @@ async function ensureSubscription({ fetchImpl = globalThis.fetch, log = console
@@ -57,12 +59,16 @@ async function ensureSubscription({ fetchImpl = globalThis.fetch, log = console
const list = await fetchImpl(`${eventsUrl}/api/v1/subscriptions`, { headers });
const list = await fetchImpl(`${eventsUrl}/api/v1/subscriptions`, { headers });
if (!list.ok) throw new Error(`Events answered ${list.status} listing subscriptions`);
if (!list.ok) throw new Error(`Events answered ${list.status} listing subscriptions`);
const subs = ((await list.json()).subscriptions || []);
const subs = ((await list.json()).subscriptions || []);
- if (subs.some((s) => s.topic_pattern === TOPIC && s.endpoint === endpoint)) return 'exists';
- const r = await fetchImpl(`${eventsUrl}/api/v1/subscriptions`, { method: 'POST', headers: { ...headers, 'content-type': 'application/json' }, body: JSON.stringify({ topic_pattern: TOPIC, endpoint, secret }) });
- if (r.status === 409) return 'exists';
- if (!r.ok) throw new Error(`Events answered ${r.status} creating the ${TOPIC} subscription`);
- log.log && log.log(`[Events] subscription created: ${TOPIC} → ${endpoint}`);
- return 'created';
+ let created = 0;
+ for (const topic of TOPICS) {
+ if (subs.some((s) => s.topic_pattern === topic && s.endpoint === endpoint)) continue;
+ const r = await fetchImpl(`${eventsUrl}/api/v1/subscriptions`, { method: 'POST', headers: { ...headers, 'content-type': 'application/json' }, body: JSON.stringify({ topic_pattern: topic, endpoint, secret }) });
+ if (r.status === 409) continue;
+ if (!r.ok) throw new Error(`Events answered ${r.status} creating the ${topic} subscription`);
+ log.log && log.log(`[Events] subscription created: ${topic} → ${endpoint}`);
+ created++;
+ }
+ return created ? 'created' : 'exists';
}
}
-module.exports = { handler, ensureSubscription, isRevoked, stats, TOPIC, _cutoffs: cutoffs };
+module.exports = { handler, ensureSubscription, isRevoked, stats, TOPIC, TOPICS, _cutoffs: cutoffs };
new file mode 100644
new file mode 100644
@@ -0,0 +1,38 @@
@@ -0,0 +1,38 @@
+'use strict';
+/**
+ * network.subject.merged → Media (roadmap WS-B task 5, ADR-029; Contracts 0.69.0). Two Network accounts became one:
+ * `from` is an alias of `into`. Media repoints what it owns by subject: media_objects.owner_subject from `from` to
+ * `into`, in one transaction, once per merge_id (subject_merges). VOD and clip rows name Live's user ids, which Live's
+ * own consumer handles. Arrives at POST /internal/events (server/revocations.js), signed and loopback-only.
+ */
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const MERGE_RE = /^mrg_[0-9A-HJKMNP-TV-Z]{26}$/;
+const TOPIC = 'network.subject.merged';
+
+function ensureSchema(db) {
+ db.exec(`CREATE TABLE IF NOT EXISTS subject_merges (
+ merge_id TEXT PRIMARY KEY,
+ from_subject TEXT NOT NULL,
+ into_subject TEXT NOT NULL,
+ objects INTEGER NOT NULL DEFAULT 0,
+ applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+ )`);
+}
+
+/** One envelope → 'merged' | 'unchanged' | 'ignored:<why>'. */
+function apply(ev, { db = require('./db/database').getDb() } = {}) {
+ if (!ev || ev.event_type !== TOPIC) return 'ignored:type';
+ if (ev.source !== 'network') return 'ignored:source';
+ const p = ev.payload && typeof ev.payload === 'object' ? ev.payload : {};
+ if (!MERGE_RE.test(String(p.merge_id || '')) || !SUBJECT_RE.test(String(p.from || '')) || !SUBJECT_RE.test(String(p.into || '')) || p.from === p.into) return 'ignored:payload';
+ ensureSchema(db);
+ return db.transaction(() => {
+ if (db.prepare('SELECT 1 FROM subject_merges WHERE merge_id = ?').get(p.merge_id)) return 'unchanged';
+ const objects = db.prepare('UPDATE media_objects SET owner_subject = ? WHERE owner_subject = ?').run(p.into, p.from).changes;
+ db.prepare('INSERT INTO subject_merges (merge_id, from_subject, into_subject, objects) VALUES (?, ?, ?, ?)').run(p.merge_id, p.from, p.into, objects);
+ console.log(`[Merge] ${p.merge_id}: ${objects} object(s) now owned by the survivor`);
+ return 'merged';
+ })();
+}
+
+module.exports = { apply, ensureSchema, TOPIC };
@@ -45,6 +45,18 @@ const ev = (over = {}) => ({ event_id: 'evt_01J8Z3Q4R5S6T7V8W9X0Y1Z2A3', event_t
@@ -45,6 +45,18 @@ const ev = (over = {}) => ({ event_id: 'evt_01J8Z3Q4R5S6T7V8W9X0Y1Z2A3', event_t
assert.strictEqual(revocations.isRevoked(old), true, 'older tokens are refused');
assert.strictEqual(revocations.isRevoked(old), true, 'older tokens are refused');
assert.strictEqual(revocations.isRevoked({ subject_id: SUBJECT, iat: at / 1000 + 5 }), false, 'newer ones are fine');
assert.strictEqual(revocations.isRevoked({ subject_id: SUBJECT, iat: at / 1000 + 5 }), false, 'newer ones are fine');
assert.strictEqual((await post(ev())).body.outcome, 'unchanged', 'a redelivery');
assert.strictEqual((await post(ev())).body.outcome, 'unchanged', 'a redelivery');
+
+ // network.subject.merged (ADR-029): the folded-in account's objects are the survivor's, once per merge.
+ const db = require('../server/db/database').getDb();
+ const FROM = 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2B4', INTO = 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2C5';
+ const obj = db.prepare("INSERT INTO media_objects (id, app_id, namespace, kind, owner_subject) VALUES (?, 'live', 'live', 'file', ?)");
+ obj.run('med_01J8Z3Q4R5S6T7V8W9X0Y1Z2D6', FROM); obj.run('med_01J8Z3Q4R5S6T7V8W9X0Y1Z2E7', FROM); obj.run('med_01J8Z3Q4R5S6T7V8W9X0Y1Z2F8', SUBJECT);
+ const merged = (over = {}) => ev({ event_id: 'evt_01J8Z3Q4R5S6T7V8W9X0Y1Z2G9', event_type: 'network.subject.merged', payload: { merge_id: 'mrg_01J8Z3Q4R5S6T7V8W9X0Y1Z2H0', from: FROM, into: INTO, merged_at: '2026-09-27T21:00:00.000Z', initiated_by: 'person' }, ...over });
+ assert.strictEqual((await post(merged({ source: 'live' }))).body.outcome, 'ignored:source');
+ assert.strictEqual((await post(merged())).body.outcome, 'merged');
+ assert.deepStrictEqual(db.prepare('SELECT owner_subject AS o, COUNT(*) AS n FROM media_objects GROUP BY o ORDER BY o').all().map((r) => [r.o, r.n]), [[SUBJECT, 1], [INTO, 2]].sort());
+ assert.strictEqual((await post(merged())).body.outcome, 'unchanged', 'a redelivery moves nothing');
+ assert.strictEqual(db.prepare('SELECT objects FROM subject_merges').get().objects, 2);
const src = fs.readFileSync(path.join(__dirname, '../server/user-auth.js'), 'utf8');
const src = fs.readFileSync(path.join(__dirname, '../server/user-auth.js'), 'utf8');
assert.ok(/require\('\.\/revocations'\)\.isRevoked\(claims\)/.test(src), 'the site token check consults it');
assert.ok(/require\('\.\/revocations'\)\.isRevoked\(claims\)/.test(src), 'the site token check consults it');
@@ -58,8 +70,7 @@ const ev = (over = {}) => ({ event_id: 'evt_01J8Z3Q4R5S6T7V8W9X0Y1Z2A3', event_t
@@ -58,8 +70,7 @@ const ev = (over = {}) => ({ event_id: 'evt_01J8Z3Q4R5S6T7V8W9X0Y1Z2A3', event_t
};
};
assert.strictEqual(await revocations.ensureSubscription({ fetchImpl: fake, log: { log() {} } }), 'created');
assert.strictEqual(await revocations.ensureSubscription({ fetchImpl: fake, log: { log() {} } }), 'created');
assert.strictEqual(await revocations.ensureSubscription({ fetchImpl: fake, log: { log() {} } }), 'exists');
assert.strictEqual(await revocations.ensureSubscription({ fetchImpl: fake, log: { log() {} } }), 'exists');
- assert.strictEqual(subs.length, 1);
- assert.strictEqual(subs[0].topic_pattern, revocations.TOPIC);
+ assert.deepStrictEqual(subs.map((s) => s.topic_pattern), [revocations.TOPIC, 'network.subject.merged'], 'one subscription per topic, created once');
} finally {
} finally {
srv.close();
srv.close();
fs.rmSync(tmp, { recursive: true, force: true });
fs.rmSync(tmp, { recursive: true, force: true });