openvibers/openvibe.live Commits
Canonical subjects: keep the token's subject_id on linked_accounts and req.user; push Live<->Network links to Network's identity_legacy_map daily
Changes
@@@ -117,7 +117,12 @@ function resolveNetworkUser(decoded) {).get(openvibeToolsId);if (linked) {-- return _syncSsoUserFields(db.getUserById(linked.user_id), decoded);++ const user = _syncSsoUserFields(db.getUserById(linked.user_id), decoded);++ if (user && decoded.subject_id) {++ user.subject_id = decoded.subject_id;++ require('./identity-sync').noteSubject(user.id, openvibeToolsId, decoded.subject_id);++ }++ return user;}// Try matching by username (case-insensitive)nnew file mode 100644@@@ -0,0 +1,62 @@++'use strict';++/**++ * Canonical subjects on the Live side (roadmap Wave 1, ADR-001).++ *++ * - noteSubject(): Network tokens carry `subject_id` (usr_<ULID>) next to the integer `sub`. The++ * verified claim is remembered on linked_accounts(service='network').subject_id, so Live can hand++ * out and accept canonical ids without asking Network per request.++ * - syncLegacyMap(): pushes every Live-user <-> Network-account link to Network's++ * identity_legacy_map (POST /internal/identity/legacy-map), so any service can resolve++ * "live user N" without calling Live. Network never repoints an existing mapping; conflicts are++ * logged here for a human to look at.++ */++const db = require('../db/database');++const { OV_NETWORK_INTERNAL_URL, INTERNAL_API_KEY } = require('../utils/notify');++++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const BATCH = 500;++const _known = new Map(); // live user id -> subject id already stored++++/** Store the token's subject id for this Live user (only writes when it changes). */++function noteSubject(liveUserId, networkUserId, subjectId) {++ if (!liveUserId || !SUBJECT_RE.test(String(subjectId || ''))) return;++ if (_known.get(liveUserId) === subjectId) return;++ try {++ db.getDb().prepare("UPDATE linked_accounts SET subject_id = ? WHERE service = 'network' AND user_id = ? AND service_user_id = ? AND (subject_id IS NULL OR subject_id != ?)")++ .run(subjectId, liveUserId, String(networkUserId), subjectId);++ _known.set(liveUserId, subjectId);++ } catch (err) { console.warn('[Identity] subject note failed:', err.message); }++}++++/** Subject id for a Live user id, if Network has told us (via a token) yet. */++function subjectOf(liveUserId) {++ const row = db.getDb().prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ?").get(liveUserId);++ return row ? row.subject_id || null : null;++}++++async function syncLegacyMap() {++ if (!INTERNAL_API_KEY) return { skipped: 'no INTERNAL_API_KEY' };++ const rows = db.getDb().prepare("SELECT user_id, service_user_id FROM linked_accounts WHERE service = 'network' AND service_user_id GLOB '[0-9]*' ORDER BY user_id").all();++ const total = { sent: rows.length, inserted: 0, unchanged: 0, conflicts: 0, rejected: 0 };++ for (let i = 0; i < rows.length; i += BATCH) {++ const entries = rows.slice(i, i + BATCH).map(r => ({++ network_user_id: Number(r.service_user_id), source_system: 'live', source_type: 'user', source_id: String(r.user_id), verified: true,++ }));++ const res = await fetch(`${OV_NETWORK_INTERNAL_URL}/internal/identity/legacy-map`, {++ method: 'POST',++ headers: { 'Content-Type': 'application/json', 'X-Internal-Key': INTERNAL_API_KEY },++ body: JSON.stringify({ entries }),++ signal: AbortSignal.timeout(15_000),++ });++ if (res.status === 404) return { skipped: 'network has no /internal/identity yet' };++ if (!res.ok) throw new Error(`legacy-map ${res.status}`);++ const out = await res.json();++ total.inserted += out.inserted; total.unchanged += out.unchanged;++ total.conflicts += out.conflicts.length; total.rejected += out.rejected.length;++ for (const c of out.conflicts) console.warn(`[Identity] live user ${c.source.split(':').pop()} is mapped to ${c.mapped_to} on Network, not ${c.requested}; left as is`);++ }++ if (total.inserted || total.conflicts || total.rejected) console.log(`[Identity] legacy map sync: ${JSON.stringify(total)}`);++ return total;++}++++module.exports = { noteSubject, subjectOf, syncLegacyMap };@@@ -402,6 +402,10 @@ function initDb() {if (moved || dropped) console.log(`[DB] linked_accounts: ${moved} hobotools→network row(s) renamed, ${dropped} duplicate(s) dropped`);} catch (e) { console.warn('[DB] linked_accounts rename:', e.message); }++ // Canonical subject id (usr_<ULID>) from the Network token, next to the integer network id (Wave 1).++ const laCols = database.prepare('PRAGMA table_info(linked_accounts)').all().map(c => c.name);++ if (!laCols.includes('subject_id')) database.exec('ALTER TABLE linked_accounts ADD COLUMN subject_id TEXT');++ database.exec(`CREATE INDEX IF NOT EXISTS idx_linked_subject ON linked_accounts(subject_id)`);database.exec(`CREATE INDEX IF NOT EXISTS idx_linked_service ON linked_accounts(service, service_user_id)`);database.exec(`CREATE INDEX IF NOT EXISTS idx_linked_user ON linked_accounts(user_id)`);} catch (e) { console.warn('[DB] linked_accounts migration:', e.message); }@@@ -1499,6 +1499,11 @@ async function start() {}, 60000);if (typeof maintenanceInterval.unref === 'function') maintenanceInterval.unref();++ // 8b. Canonical identity: report every Live<->Network link to Network's identity_legacy_map++ // (roadmap Wave 1). Idempotent on Network's side; daily, first run a few minutes after boot.++ require('./utils/jobs').every('identity-legacy-sync', 24 * 60 * 60 * 1000, () => require('./auth/identity-sync').syncLegacyMap(),++ { initialDelayMs: 3 * 60 * 1000, jitterMs: 60 * 1000 });++// 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.nnew file mode 100644@@@ -0,0 +1,80 @@++'use strict';++++// Canonical subjects on the Live side (roadmap Wave 1): the Network token's subject_id lands on++// req.user and linked_accounts, and every Live<->Network link is pushed to Network's++// identity_legacy_map in batches (against a stub Network).++++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-identity-'));++process.env.DB_PATH = path.join(tmp, 'live.db');++process.env.INTERNAL_API_KEY = 'k-test';++++const received = [];++const network = http.createServer((req, res) => {++ let body = '';++ req.on('data', (c) => { body += c; });++ req.on('end', () => {++ assert.strictEqual(req.headers['x-internal-key'], 'k-test');++ assert.strictEqual(req.url, '/internal/identity/legacy-map');++ const { entries } = JSON.parse(body);++ received.push(entries);++ res.setHeader('Content-Type', 'application/json');++ res.end(JSON.stringify({ inserted: entries.length - 1, unchanged: 0, conflicts: [{ index: 0, source: `live:user:${entries[0].source_id}`, mapped_to: 'usr_A', requested: 'usr_B' }], rejected: [] }));++ });++});++++(async () => {++ await new Promise((r) => network.listen(0, '127.0.0.1', r));++ process.env.OV_NETWORK_INTERNAL_URL = `http://127.0.0.1:${network.address().port}`;++++ const db = require('../server/db/database');++ db.initDb();++ const sync = require('../server/auth/identity-sync');++ const d = db.getDb();++++ assert.ok(d.prepare('PRAGMA table_info(linked_accounts)').all().some((c) => c.name === 'subject_id'), 'linked_accounts.subject_id exists');++++ // 600 linked users (two batches) plus one legacy non-numeric row that must not be sent.++ const insUser = d.prepare("INSERT INTO users (username, password_hash, stream_key) VALUES (?, 'x', ?)");++ const insLink = d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, service_username) VALUES (?, 'network', ?, ?)");++ for (let i = 1; i <= 600; i++) {++ const id = insUser.run(`u${i}`, `key${i}`).lastInsertRowid;++ insLink.run(id, String(1000 + i), `u${i}`);++ }++ const odd = insUser.run('odd', 'keyodd').lastInsertRowid;++ insLink.run(odd, 'network:odd', 'odd');++++ const out = await sync.syncLegacyMap();++ assert.strictEqual(received.length, 2, 'batched by 500');++ assert.strictEqual(out.sent, 600, 'non-numeric links are skipped');++ assert.strictEqual(out.conflicts, 2, 'conflicts are counted, not retried');++ const first = received[0][0];++ assert.deepStrictEqual(Object.keys(first).sort(), ['network_user_id', 'source_id', 'source_system', 'source_type', 'verified']);++ assert.strictEqual(first.source_system, 'live');++ assert.strictEqual(typeof first.network_user_id, 'number');++++ // A token's subject_id is stored once, only for a well-formed id and the matching link.++ const uid = d.prepare("SELECT user_id FROM linked_accounts WHERE service_user_id = '1001'").get().user_id;++ sync.noteSubject(uid, 1001, 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ');++ assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ');++ sync.noteSubject(uid, 1001, '42');++ assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'malformed subject ignored');++ sync.noteSubject(uid, 9999, 'usr_01JAB2C3D4E5F6G7H8J9K0MNPR');++ assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'a token for a different network id cannot rewrite the link');++++ // Network without the endpoint yet: skip quietly.++ network.close();++ const old = http.createServer((req, res) => { res.statusCode = 404; res.end('{}'); });++ await new Promise((r) => old.listen(Number(process.env.OV_NETWORK_INTERNAL_URL.split(':').pop()), '127.0.0.1', r));++ assert.deepStrictEqual(await sync.syncLegacyMap(), { skipped: 'network has no /internal/identity yet' });++ old.close();++++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('identity sync: all checks passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -117,7 +117,12 @@ function resolveNetworkUser(decoded) {
@@ -117,7 +117,12 @@ function resolveNetworkUser(decoded) {
).get(openvibeToolsId);
).get(openvibeToolsId);
if (linked) {
if (linked) {
- return _syncSsoUserFields(db.getUserById(linked.user_id), decoded);
+ const user = _syncSsoUserFields(db.getUserById(linked.user_id), decoded);
+ if (user && decoded.subject_id) {
+ user.subject_id = decoded.subject_id;
+ require('./identity-sync').noteSubject(user.id, openvibeToolsId, decoded.subject_id);
+ }
+ return user;
}
}
// Try matching by username (case-insensitive)
// Try matching by username (case-insensitive)
new file mode 100644
new file mode 100644
@@ -0,0 +1,62 @@
@@ -0,0 +1,62 @@
+'use strict';
+/**
+ * Canonical subjects on the Live side (roadmap Wave 1, ADR-001).
+ *
+ * - noteSubject(): Network tokens carry `subject_id` (usr_<ULID>) next to the integer `sub`. The
+ * verified claim is remembered on linked_accounts(service='network').subject_id, so Live can hand
+ * out and accept canonical ids without asking Network per request.
+ * - syncLegacyMap(): pushes every Live-user <-> Network-account link to Network's
+ * identity_legacy_map (POST /internal/identity/legacy-map), so any service can resolve
+ * "live user N" without calling Live. Network never repoints an existing mapping; conflicts are
+ * logged here for a human to look at.
+ */
+const db = require('../db/database');
+const { OV_NETWORK_INTERNAL_URL, INTERNAL_API_KEY } = require('../utils/notify');
+
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const BATCH = 500;
+const _known = new Map(); // live user id -> subject id already stored
+
+/** Store the token's subject id for this Live user (only writes when it changes). */
+function noteSubject(liveUserId, networkUserId, subjectId) {
+ if (!liveUserId || !SUBJECT_RE.test(String(subjectId || ''))) return;
+ if (_known.get(liveUserId) === subjectId) return;
+ try {
+ db.getDb().prepare("UPDATE linked_accounts SET subject_id = ? WHERE service = 'network' AND user_id = ? AND service_user_id = ? AND (subject_id IS NULL OR subject_id != ?)")
+ .run(subjectId, liveUserId, String(networkUserId), subjectId);
+ _known.set(liveUserId, subjectId);
+ } catch (err) { console.warn('[Identity] subject note failed:', err.message); }
+}
+
+/** Subject id for a Live user id, if Network has told us (via a token) yet. */
+function subjectOf(liveUserId) {
+ const row = db.getDb().prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ?").get(liveUserId);
+ return row ? row.subject_id || null : null;
+}
+
+async function syncLegacyMap() {
+ if (!INTERNAL_API_KEY) return { skipped: 'no INTERNAL_API_KEY' };
+ const rows = db.getDb().prepare("SELECT user_id, service_user_id FROM linked_accounts WHERE service = 'network' AND service_user_id GLOB '[0-9]*' ORDER BY user_id").all();
+ const total = { sent: rows.length, inserted: 0, unchanged: 0, conflicts: 0, rejected: 0 };
+ for (let i = 0; i < rows.length; i += BATCH) {
+ const entries = rows.slice(i, i + BATCH).map(r => ({
+ network_user_id: Number(r.service_user_id), source_system: 'live', source_type: 'user', source_id: String(r.user_id), verified: true,
+ }));
+ const res = await fetch(`${OV_NETWORK_INTERNAL_URL}/internal/identity/legacy-map`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json', 'X-Internal-Key': INTERNAL_API_KEY },
+ body: JSON.stringify({ entries }),
+ signal: AbortSignal.timeout(15_000),
+ });
+ if (res.status === 404) return { skipped: 'network has no /internal/identity yet' };
+ if (!res.ok) throw new Error(`legacy-map ${res.status}`);
+ const out = await res.json();
+ total.inserted += out.inserted; total.unchanged += out.unchanged;
+ total.conflicts += out.conflicts.length; total.rejected += out.rejected.length;
+ for (const c of out.conflicts) console.warn(`[Identity] live user ${c.source.split(':').pop()} is mapped to ${c.mapped_to} on Network, not ${c.requested}; left as is`);
+ }
+ if (total.inserted || total.conflicts || total.rejected) console.log(`[Identity] legacy map sync: ${JSON.stringify(total)}`);
+ return total;
+}
+
+module.exports = { noteSubject, subjectOf, syncLegacyMap };
@@ -402,6 +402,10 @@ function initDb() {
@@ -402,6 +402,10 @@ function initDb() {
if (moved || dropped) console.log(`[DB] linked_accounts: ${moved} hobotools→network row(s) renamed, ${dropped} duplicate(s) dropped`);
if (moved || dropped) console.log(`[DB] linked_accounts: ${moved} hobotools→network row(s) renamed, ${dropped} duplicate(s) dropped`);
} catch (e) { console.warn('[DB] linked_accounts rename:', e.message); }
} catch (e) { console.warn('[DB] linked_accounts rename:', e.message); }
+ // Canonical subject id (usr_<ULID>) from the Network token, next to the integer network id (Wave 1).
+ const laCols = database.prepare('PRAGMA table_info(linked_accounts)').all().map(c => c.name);
+ if (!laCols.includes('subject_id')) database.exec('ALTER TABLE linked_accounts ADD COLUMN subject_id TEXT');
+ database.exec(`CREATE INDEX IF NOT EXISTS idx_linked_subject ON linked_accounts(subject_id)`);
database.exec(`CREATE INDEX IF NOT EXISTS idx_linked_service ON linked_accounts(service, service_user_id)`);
database.exec(`CREATE INDEX IF NOT EXISTS idx_linked_service ON linked_accounts(service, service_user_id)`);
database.exec(`CREATE INDEX IF NOT EXISTS idx_linked_user ON linked_accounts(user_id)`);
database.exec(`CREATE INDEX IF NOT EXISTS idx_linked_user ON linked_accounts(user_id)`);
} catch (e) { console.warn('[DB] linked_accounts migration:', e.message); }
} catch (e) { console.warn('[DB] linked_accounts migration:', e.message); }
@@ -1499,6 +1499,11 @@ async function start() {
@@ -1499,6 +1499,11 @@ async function start() {
}, 60000);
}, 60000);
if (typeof maintenanceInterval.unref === 'function') maintenanceInterval.unref();
if (typeof maintenanceInterval.unref === 'function') maintenanceInterval.unref();
+ // 8b. Canonical identity: report every Live<->Network link to Network's identity_legacy_map
+ // (roadmap Wave 1). Idempotent on Network's side; daily, first run a few minutes after boot.
+ require('./utils/jobs').every('identity-legacy-sync', 24 * 60 * 60 * 1000, () => require('./auth/identity-sync').syncLegacyMap(),
+ { initialDelayMs: 3 * 60 * 1000, jitterMs: 60 * 1000 });
+
// 9. Periodic registry refresh — re-syncs config with openvibe.network every 5 minutes.
// 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
// 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.
// unreachable), subsequent refreshes will fix CORS, issuer, and other URL config.
new file mode 100644
new file mode 100644
@@ -0,0 +1,80 @@
@@ -0,0 +1,80 @@
+'use strict';
+
+// Canonical subjects on the Live side (roadmap Wave 1): the Network token's subject_id lands on
+// req.user and linked_accounts, and every Live<->Network link is pushed to Network's
+// identity_legacy_map in batches (against a stub Network).
+
+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-identity-'));
+process.env.DB_PATH = path.join(tmp, 'live.db');
+process.env.INTERNAL_API_KEY = 'k-test';
+
+const received = [];
+const network = http.createServer((req, res) => {
+ let body = '';
+ req.on('data', (c) => { body += c; });
+ req.on('end', () => {
+ assert.strictEqual(req.headers['x-internal-key'], 'k-test');
+ assert.strictEqual(req.url, '/internal/identity/legacy-map');
+ const { entries } = JSON.parse(body);
+ received.push(entries);
+ res.setHeader('Content-Type', 'application/json');
+ res.end(JSON.stringify({ inserted: entries.length - 1, unchanged: 0, conflicts: [{ index: 0, source: `live:user:${entries[0].source_id}`, mapped_to: 'usr_A', requested: 'usr_B' }], rejected: [] }));
+ });
+});
+
+(async () => {
+ await new Promise((r) => network.listen(0, '127.0.0.1', r));
+ process.env.OV_NETWORK_INTERNAL_URL = `http://127.0.0.1:${network.address().port}`;
+
+ const db = require('../server/db/database');
+ db.initDb();
+ const sync = require('../server/auth/identity-sync');
+ const d = db.getDb();
+
+ assert.ok(d.prepare('PRAGMA table_info(linked_accounts)').all().some((c) => c.name === 'subject_id'), 'linked_accounts.subject_id exists');
+
+ // 600 linked users (two batches) plus one legacy non-numeric row that must not be sent.
+ const insUser = d.prepare("INSERT INTO users (username, password_hash, stream_key) VALUES (?, 'x', ?)");
+ const insLink = d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, service_username) VALUES (?, 'network', ?, ?)");
+ for (let i = 1; i <= 600; i++) {
+ const id = insUser.run(`u${i}`, `key${i}`).lastInsertRowid;
+ insLink.run(id, String(1000 + i), `u${i}`);
+ }
+ const odd = insUser.run('odd', 'keyodd').lastInsertRowid;
+ insLink.run(odd, 'network:odd', 'odd');
+
+ const out = await sync.syncLegacyMap();
+ assert.strictEqual(received.length, 2, 'batched by 500');
+ assert.strictEqual(out.sent, 600, 'non-numeric links are skipped');
+ assert.strictEqual(out.conflicts, 2, 'conflicts are counted, not retried');
+ const first = received[0][0];
+ assert.deepStrictEqual(Object.keys(first).sort(), ['network_user_id', 'source_id', 'source_system', 'source_type', 'verified']);
+ assert.strictEqual(first.source_system, 'live');
+ assert.strictEqual(typeof first.network_user_id, 'number');
+
+ // A token's subject_id is stored once, only for a well-formed id and the matching link.
+ const uid = d.prepare("SELECT user_id FROM linked_accounts WHERE service_user_id = '1001'").get().user_id;
+ sync.noteSubject(uid, 1001, 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ');
+ assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ');
+ sync.noteSubject(uid, 1001, '42');
+ assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'malformed subject ignored');
+ sync.noteSubject(uid, 9999, 'usr_01JAB2C3D4E5F6G7H8J9K0MNPR');
+ assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'a token for a different network id cannot rewrite the link');
+
+ // Network without the endpoint yet: skip quietly.
+ network.close();
+ const old = http.createServer((req, res) => { res.statusCode = 404; res.end('{}'); });
+ await new Promise((r) => old.listen(Number(process.env.OV_NETWORK_INTERNAL_URL.split(':').pop()), '127.0.0.1', r));
+ assert.deepStrictEqual(await sync.syncLegacyMap(), { skipped: 'network has no /internal/identity yet' });
+ old.close();
+
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('identity sync: all checks passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });