openvibers/openvibe.live Commits
Identity: Live asks Network for the subject of every linked account it has no token subject for yet (backfillSubjects, POST /internal/identity/resolve-batch by Network user id, 500 per call), right after the daily legacy-map sync. Only 5 of 328 production links had one, because subject ids were only learned from tokens people presented since Wave 1; module migrations (chat.preferences), stream events, the follower list and the lineage resolver all key on it. An answer is stored only when it names the same Network user id with a well-formed usr_ subject, and never over a subject Live already has. test/identity-sync.test.js covers batching, the skipped rows, mismatched and malformed answers, the rerun and a Network without the endpoint
Changes
@@@ -9,6 +9,10 @@* 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.++ * - backfillSubjects(): asks Network (POST /internal/identity/resolve-batch, by Network user id)++ * for the subject of every linked account Live has not seen a token subject for yet, so people++ * who have not signed in since Wave 1 still have one (module migrations, events, the lineage++ * resolver all key on it). Only an answer naming the same Network user id is stored.*/const db = require('../db/database');const { OV_NETWORK_INTERNAL_URL, INTERNAL_API_KEY } = require('../utils/notify');@@@ -59,4 +63,32 @@ async function syncLegacyMap() {return total;}--module.exports = { noteSubject, subjectOf, syncLegacyMap };++async function backfillSubjects() {++ 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 subject_id IS NULL AND service_user_id GLOB '[0-9]*' ORDER BY user_id").all();++ const total = { asked: rows.length, stored: 0, unknown: 0 };++ const store = db.getDb().prepare("UPDATE linked_accounts SET subject_id = ? WHERE service = 'network' AND user_id = ? AND service_user_id = ? AND subject_id IS NULL");++ for (let i = 0; i < rows.length; i += BATCH) {++ const batch = rows.slice(i, i + BATCH);++ const res = await fetch(`${OV_NETWORK_INTERNAL_URL}/internal/identity/resolve-batch`, {++ method: 'POST',++ headers: { 'Content-Type': 'application/json', 'X-Internal-Key': INTERNAL_API_KEY },++ body: JSON.stringify({ system: 'network', type: 'user', ids: batch.map(r => String(r.service_user_id)) }),++ signal: AbortSignal.timeout(15_000),++ });++ if (res.status === 404) return { ...total, skipped: 'network has no /internal/identity yet' };++ if (!res.ok) throw new Error(`resolve-batch ${res.status}`);++ const { results = {} } = await res.json();++ for (const r of batch) {++ const p = results[String(r.service_user_id)];++ const sid = p && p.subject && p.subject.type === 'user' ? p.subject.id : null;++ if (!sid || !SUBJECT_RE.test(sid) || String(p.network_user_id) !== String(r.service_user_id)) { total.unknown++; continue; }++ total.stored += store.run(sid, r.user_id, String(r.service_user_id)).changes;++ _known.set(r.user_id, sid);++ }++ }++ if (total.stored || total.unknown) console.log(`[Identity] subject backfill: ${JSON.stringify(total)}`);++ return total;++}++++module.exports = { noteSubject, subjectOf, syncLegacyMap, backfillSubjects };@@@ -1489,7 +1489,12 @@ async function start() {// 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(),++ // Then fill in the subject of every linked account no token has told Live about yet.++ require('./utils/jobs').every('identity-legacy-sync', 24 * 60 * 60 * 1000, async () => {++ const sync = require('./auth/identity-sync');++ await sync.syncLegacyMap();++ return sync.backfillSubjects();++ },{ initialDelayMs: 3 * 60 * 1000, jitterMs: 60 * 1000 });// 8b2. Raw analytics retention (ADR-021): events older than 30 days go, in bounded batches;@@@ -15,13 +15,30 @@ process.env.DB_PATH = path.join(tmp, 'live.db');process.env.INTERNAL_API_KEY = 'k-test';const received = [];++const resolveCalls = [];const mode = { old: false };++// A well-formed subject per Network user id (Crockford base32, 26 chars).++const subjectFor = (n) => `usr_01J${String(n).padStart(23, '0')}`;const network = http.createServer((req, res) => {let body = '';req.on('data', (c) => { body += c; });req.on('end', () => {if (mode.old) { res.statusCode = 404; return res.end('{}'); }assert.strictEqual(req.headers['x-internal-key'], 'k-test');++ if (req.url === '/internal/identity/resolve-batch') {++ const q = JSON.parse(body);++ resolveCalls.push(q);++ const results = {};++ for (const id of q.ids) {++ const n = Number(id);++ if (n === 1002) results[id] = null; // Network does not know it++ else if (n === 1003) results[id] = { subject: { type: 'user', id: subjectFor(n) }, network_user_id: 7 }; // answers for someone else++ else if (n === 1004) results[id] = { subject: { type: 'user', id: 'not-a-subject' }, network_user_id: n };++ else results[id] = { subject: { type: 'user', id: subjectFor(n) }, network_user_id: n };++ }++ res.setHeader('Content-Type', 'application/json');++ return res.end(JSON.stringify({ results }));++ }assert.strictEqual(req.url, '/internal/identity/legacy-map');const { entries } = JSON.parse(body);received.push(entries);@@@ -69,9 +86,28 @@ const network = http.createServer((req, res) => {sync.noteSubject(uid, 9999, 'usr_01JAB2C3D4E5F6G7H8J9K0MNPR');assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'a token for a different network id cannot rewrite the link');++ // Backfill: every link without a subject is asked by Network user id, in batches; only an answer++ // naming the same Network user id with a well-formed subject is stored, and a known one is kept.++ const back = await sync.backfillSubjects();++ assert.strictEqual(resolveCalls.length, 2, 'batched by 500');++ assert.strictEqual(resolveCalls[0].system, 'network');++ assert.strictEqual(resolveCalls[0].type, 'user');++ assert.ok(!resolveCalls.flatMap(q => q.ids).includes('1001'), 'a link with a subject is not asked');++ assert.ok(!resolveCalls.flatMap(q => q.ids).includes('network:odd'), 'non-numeric links are not asked');++ assert.deepStrictEqual(back, { asked: 599, stored: 596, unknown: 3 });++ assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'the token subject stays');++ const linkOf = (n) => d.prepare('SELECT user_id FROM linked_accounts WHERE service_user_id = ?').get(String(n)).user_id;++ assert.strictEqual(sync.subjectOf(linkOf(1600)), subjectFor(1600));++ assert.strictEqual(sync.subjectOf(linkOf(1002)), null, 'unknown to Network');++ assert.strictEqual(sync.subjectOf(linkOf(1003)), null, 'an answer for another Network user is ignored');++ assert.strictEqual(sync.subjectOf(linkOf(1004)), null, 'a malformed subject is ignored');++ resolveCalls.length = 0;++ assert.deepStrictEqual(await sync.backfillSubjects(), { asked: 3, stored: 0, unknown: 3 }, 'the next run asks only what is still missing');++// Network without the endpoint yet: skip quietly.mode.old = true;assert.deepStrictEqual(await sync.syncLegacyMap(), { skipped: 'network has no /internal/identity yet' });++ assert.deepStrictEqual(await sync.backfillSubjects(), { asked: 3, stored: 0, unknown: 0, skipped: 'network has no /internal/identity yet' });network.close();fs.rmSync(tmp, { recursive: true, force: true });
@@ -9,6 +9,10 @@
@@ -9,6 +9,10 @@
* identity_legacy_map (POST /internal/identity/legacy-map), so any service can resolve
* 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
* "live user N" without calling Live. Network never repoints an existing mapping; conflicts are
* logged here for a human to look at.
* logged here for a human to look at.
+ * - backfillSubjects(): asks Network (POST /internal/identity/resolve-batch, by Network user id)
+ * for the subject of every linked account Live has not seen a token subject for yet, so people
+ * who have not signed in since Wave 1 still have one (module migrations, events, the lineage
+ * resolver all key on it). Only an answer naming the same Network user id is stored.
*/
*/
const db = require('../db/database');
const db = require('../db/database');
const { OV_NETWORK_INTERNAL_URL, INTERNAL_API_KEY } = require('../utils/notify');
const { OV_NETWORK_INTERNAL_URL, INTERNAL_API_KEY } = require('../utils/notify');
@@ -59,4 +63,32 @@ async function syncLegacyMap() {
@@ -59,4 +63,32 @@ async function syncLegacyMap() {
return total;
return total;
}
}
-module.exports = { noteSubject, subjectOf, syncLegacyMap };
+async function backfillSubjects() {
+ 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 subject_id IS NULL AND service_user_id GLOB '[0-9]*' ORDER BY user_id").all();
+ const total = { asked: rows.length, stored: 0, unknown: 0 };
+ const store = db.getDb().prepare("UPDATE linked_accounts SET subject_id = ? WHERE service = 'network' AND user_id = ? AND service_user_id = ? AND subject_id IS NULL");
+ for (let i = 0; i < rows.length; i += BATCH) {
+ const batch = rows.slice(i, i + BATCH);
+ const res = await fetch(`${OV_NETWORK_INTERNAL_URL}/internal/identity/resolve-batch`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json', 'X-Internal-Key': INTERNAL_API_KEY },
+ body: JSON.stringify({ system: 'network', type: 'user', ids: batch.map(r => String(r.service_user_id)) }),
+ signal: AbortSignal.timeout(15_000),
+ });
+ if (res.status === 404) return { ...total, skipped: 'network has no /internal/identity yet' };
+ if (!res.ok) throw new Error(`resolve-batch ${res.status}`);
+ const { results = {} } = await res.json();
+ for (const r of batch) {
+ const p = results[String(r.service_user_id)];
+ const sid = p && p.subject && p.subject.type === 'user' ? p.subject.id : null;
+ if (!sid || !SUBJECT_RE.test(sid) || String(p.network_user_id) !== String(r.service_user_id)) { total.unknown++; continue; }
+ total.stored += store.run(sid, r.user_id, String(r.service_user_id)).changes;
+ _known.set(r.user_id, sid);
+ }
+ }
+ if (total.stored || total.unknown) console.log(`[Identity] subject backfill: ${JSON.stringify(total)}`);
+ return total;
+}
+
+module.exports = { noteSubject, subjectOf, syncLegacyMap, backfillSubjects };
@@ -1489,7 +1489,12 @@ async function start() {
@@ -1489,7 +1489,12 @@ async function start() {
// 8b. Canonical identity: report every Live<->Network link to Network's identity_legacy_map
// 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.
// (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(),
+ // Then fill in the subject of every linked account no token has told Live about yet.
+ require('./utils/jobs').every('identity-legacy-sync', 24 * 60 * 60 * 1000, async () => {
+ const sync = require('./auth/identity-sync');
+ await sync.syncLegacyMap();
+ return sync.backfillSubjects();
+ },
{ initialDelayMs: 3 * 60 * 1000, jitterMs: 60 * 1000 });
{ initialDelayMs: 3 * 60 * 1000, jitterMs: 60 * 1000 });
// 8b2. Raw analytics retention (ADR-021): events older than 30 days go, in bounded batches;
// 8b2. Raw analytics retention (ADR-021): events older than 30 days go, in bounded batches;
@@ -15,13 +15,30 @@ process.env.DB_PATH = path.join(tmp, 'live.db');
@@ -15,13 +15,30 @@ process.env.DB_PATH = path.join(tmp, 'live.db');
process.env.INTERNAL_API_KEY = 'k-test';
process.env.INTERNAL_API_KEY = 'k-test';
const received = [];
const received = [];
+const resolveCalls = [];
const mode = { old: false };
const mode = { old: false };
+// A well-formed subject per Network user id (Crockford base32, 26 chars).
+const subjectFor = (n) => `usr_01J${String(n).padStart(23, '0')}`;
const network = http.createServer((req, res) => {
const network = http.createServer((req, res) => {
let body = '';
let body = '';
req.on('data', (c) => { body += c; });
req.on('data', (c) => { body += c; });
req.on('end', () => {
req.on('end', () => {
if (mode.old) { res.statusCode = 404; return res.end('{}'); }
if (mode.old) { res.statusCode = 404; return res.end('{}'); }
assert.strictEqual(req.headers['x-internal-key'], 'k-test');
assert.strictEqual(req.headers['x-internal-key'], 'k-test');
+ if (req.url === '/internal/identity/resolve-batch') {
+ const q = JSON.parse(body);
+ resolveCalls.push(q);
+ const results = {};
+ for (const id of q.ids) {
+ const n = Number(id);
+ if (n === 1002) results[id] = null; // Network does not know it
+ else if (n === 1003) results[id] = { subject: { type: 'user', id: subjectFor(n) }, network_user_id: 7 }; // answers for someone else
+ else if (n === 1004) results[id] = { subject: { type: 'user', id: 'not-a-subject' }, network_user_id: n };
+ else results[id] = { subject: { type: 'user', id: subjectFor(n) }, network_user_id: n };
+ }
+ res.setHeader('Content-Type', 'application/json');
+ return res.end(JSON.stringify({ results }));
+ }
assert.strictEqual(req.url, '/internal/identity/legacy-map');
assert.strictEqual(req.url, '/internal/identity/legacy-map');
const { entries } = JSON.parse(body);
const { entries } = JSON.parse(body);
received.push(entries);
received.push(entries);
@@ -69,9 +86,28 @@ const network = http.createServer((req, res) => {
@@ -69,9 +86,28 @@ const network = http.createServer((req, res) => {
sync.noteSubject(uid, 9999, 'usr_01JAB2C3D4E5F6G7H8J9K0MNPR');
sync.noteSubject(uid, 9999, 'usr_01JAB2C3D4E5F6G7H8J9K0MNPR');
assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'a token for a different network id cannot rewrite the link');
assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'a token for a different network id cannot rewrite the link');
+ // Backfill: every link without a subject is asked by Network user id, in batches; only an answer
+ // naming the same Network user id with a well-formed subject is stored, and a known one is kept.
+ const back = await sync.backfillSubjects();
+ assert.strictEqual(resolveCalls.length, 2, 'batched by 500');
+ assert.strictEqual(resolveCalls[0].system, 'network');
+ assert.strictEqual(resolveCalls[0].type, 'user');
+ assert.ok(!resolveCalls.flatMap(q => q.ids).includes('1001'), 'a link with a subject is not asked');
+ assert.ok(!resolveCalls.flatMap(q => q.ids).includes('network:odd'), 'non-numeric links are not asked');
+ assert.deepStrictEqual(back, { asked: 599, stored: 596, unknown: 3 });
+ assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'the token subject stays');
+ const linkOf = (n) => d.prepare('SELECT user_id FROM linked_accounts WHERE service_user_id = ?').get(String(n)).user_id;
+ assert.strictEqual(sync.subjectOf(linkOf(1600)), subjectFor(1600));
+ assert.strictEqual(sync.subjectOf(linkOf(1002)), null, 'unknown to Network');
+ assert.strictEqual(sync.subjectOf(linkOf(1003)), null, 'an answer for another Network user is ignored');
+ assert.strictEqual(sync.subjectOf(linkOf(1004)), null, 'a malformed subject is ignored');
+ resolveCalls.length = 0;
+ assert.deepStrictEqual(await sync.backfillSubjects(), { asked: 3, stored: 0, unknown: 3 }, 'the next run asks only what is still missing');
+
// Network without the endpoint yet: skip quietly.
// Network without the endpoint yet: skip quietly.
mode.old = true;
mode.old = true;
assert.deepStrictEqual(await sync.syncLegacyMap(), { skipped: 'network has no /internal/identity yet' });
assert.deepStrictEqual(await sync.syncLegacyMap(), { skipped: 'network has no /internal/identity yet' });
+ assert.deepStrictEqual(await sync.backfillSubjects(), { asked: 3, stored: 0, unknown: 0, skipped: 'network has no /internal/identity yet' });
network.close();
network.close();
fs.rmSync(tmp, { recursive: true, force: true });
fs.rmSync(tmp, { recursive: true, force: true });