openvibers/openvibe.live Commits
Live follows account merges (roadmap WS-B task 5, ADR-029; openvibe-contracts 0.68.0 -> 0.70.0). network.subject.merged arrives at POST /internal/network-events (the --network subscription now asks for it) and server/auth/subject-merge.js applies it once per merge_id (subject_merges). With a Live user for each account, the folded-in user's follows (as follower and as the channel), channel points (as viewer and as the channel; a pair both hold is summed, each move logged out and in in channel_points_log with reason account_merge) and streams move to the survivor's user in one transaction, a follow it already has or following oneself dropped and counted, and the folded-in user is marked merged_into. With only the folded-in account on Live, that user is relinked to the survivor (subject and Network user id, resolved from Network). The handler answers after the merge applied, so a failure is redelivered. test/subject-merge.test.js: moves and clashes, summed points and the balanced log, streams, the relink, redelivery, nothing-to-do and ignored cases. npm test 141/141.
Changes
@@@ -25,7 +25,7 @@"multer": "^1.4.5-lts.1","node-media-server": "^2.7.1","onnxruntime-node": "^1.20.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.68.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.70.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.23.0","sanitize-html": "^2.17.7",@@@ -2814,9 +2814,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.68.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.68.0",-- "integrity": "sha512-5XJ/MwzJj4vQlDyMqB/zZf59Bw/Cq7M14UTBJ23uPzXLjRTTQx1jO6/tV76l8/j1C95iuM/Hm7gkFn72X843og==",++ "version": "0.70.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.70.0",++ "integrity": "sha512-NUjKoaP66vsRfyurIXhjzNja/9ELwD8smsMGJ0BuZ0LFpRocTc3yc9rCtHfIztwiuO50sC3E26J3DDlJUvpJyg==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -44,7 +44,7 @@"multer": "^1.4.5-lts.1","node-media-server": "^2.7.1","onnxruntime-node": "^1.20.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.68.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.70.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.23.0","sanitize-html": "^2.17.7",@@@ -31,7 +31,7 @@ const fs = require('fs');const TOPICS = ['media.vod.*', 'media.clip.*', 'media.storage.*'];const DEFAULT_ENDPOINT = 'http://127.0.0.1:3000/internal/media-events';--const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted'];++const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted', 'network.subject.merged'];const NETWORK_ENDPOINT = 'http://127.0.0.1:3000/internal/network-events';const DEFAULT_LIVE_ENV = '/etc/openvibe/live.env';@@@ -97,4 +97,4 @@ async function backfillSubjects() {return total;}--module.exports = { noteSubject, subjectOf, syncLegacyMap, backfillSubjects };++module.exports = { noteSubject, subjectOf, syncLegacyMap, backfillSubjects, networkPost };@@@ -10,6 +10,8 @@* token_valid_after the person's older tokens are refused (./revocations.js)* updated the person's profile, role and ban (./subject-projection.js, WS-B task 2)* follow.created/.deleted Live's follows table as a projection of Network's graph (../social/network-follows.js)++ * subject.merged two accounts became one (ADR-029): the folded-in Live user's follows, channel points and streams++ * move to the survivor's (./subject-merge.js), once per merge** Applying is idempotent (a cutoff only ever moves forward; a profile revision only up), so a redelivery* is a no-op without an inbox. Anything signed but not ours to act on is acknowledged (204).@@@ -23,9 +25,10 @@ const notifyChat = (userId) => { try { const cs = require('../chat/chat-server')function secret() { return process.env.LIVE_EVENTS_SECRET || process.env.MEDIA_EVENTS_SECRET || ''; }--/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>'. */++/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>' (a merge: a promise of 'merged' | 'relinked' | 'nothing' | …). */function apply(ev) {if (!ev || !EVENT_ID_RE.test(String(ev.event_id || ''))) return 'ignored:envelope';++ if (ev.event_type === 'network.subject.merged') return ev.source === 'network' ? require('./subject-merge').apply(ev) : 'ignored:source';const follow = ev.event_type === 'network.follow.created' || ev.event_type === 'network.follow.deleted';if (ev.event_type !== 'network.user.token_valid_after' && ev.event_type !== 'network.user.updated' && !follow) return 'ignored:type';if (ev.source !== 'network') return 'ignored:source';@@@ -47,8 +50,13 @@ function handler(req, res) {const delivery = parseDelivery(req.rawBody, req.headers, key, { requireV2: true });if (!delivery) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }stats.received++;++ const count = (out) => { if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated' || out === 'followed' || out === 'unfollowed' || out === 'merged' || out === 'relinked') stats.profiles++; else stats.ignored++; };const out = apply(delivery.event);-- if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated' || out === 'followed' || out === 'unfollowed') stats.profiles++; else stats.ignored++;++ if (out && typeof out.then === 'function') {++ // A merge (the one async case): answer after it applied, so a failure is redelivered.++ return out.then((o) => { count(o); res.status(204).end(); }, (e) => { console.error('[NetworkEvents] merge failed:', e.message); res.status(500).json({ error: 'merge not applied' }); });++ }++ count(out);res.status(204).end();}nnew file mode 100644@@@ -0,0 +1,121 @@++'use strict';++/**++ * network.subject.merged → Live (roadmap WS-B task 5, ADR-029; Contracts 0.69.0). Two Network accounts became one:++ * `from` is now an alias of `into`. Live has at most one user linked to each (linked_accounts, service 'network'):++ *++ * neither, or only the survivor's nothing of Live's to move++ * only the folded-in account's that Live user becomes the survivor's: its network link names `into` (and the++ * survivor's Network user id, resolved from Network)++ * both in one transaction, the folded-in user's follows (as follower and as the followed++ * channel), channel points (as viewer and as the channel; balances of the same pair++ * are summed, each move logged in channel_points_log once) and streams move to the++ * survivor's user. A follow the survivor already has, or following oneself, is++ * dropped and counted. The folded-in Live user stays, marked merged_into.++ *++ * Live's follows are a projection of Network's graph (ADR-030), which Network already merged; this keeps the++ * projection in step without waiting for a rebuild. VODs and clips are Media's (it follows the same event), chat is++ * Chat's. Applied once per merge_id (subject_merges).++ */++const db = require('../db/database');++++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const MERGE_RE = /^mrg_[0-9A-HJKMNP-TV-Z]{26}$/;++++function ensureSchema(d) {++ d.exec(`CREATE TABLE IF NOT EXISTS subject_merges (++ merge_id TEXT PRIMARY KEY,++ from_subject TEXT NOT NULL,++ into_subject TEXT NOT NULL,++ from_user_id INTEGER,++ into_user_id INTEGER,++ outcome TEXT NOT NULL,++ applied_at DATETIME DEFAULT CURRENT_TIMESTAMP++ )`);++ const cols = d.prepare('PRAGMA table_info(users)').all().map((c) => c.name);++ if (!cols.includes('merged_into')) d.exec('ALTER TABLE users ADD COLUMN merged_into INTEGER');++}++++const liveUserOf = (d, subject) => {++ const r = d.prepare("SELECT user_id FROM linked_accounts WHERE service = 'network' AND subject_id = ? ORDER BY id LIMIT 1").get(subject);++ return r ? r.user_id : null;++};++++/** Move everything of Live user `a` to Live user `b` (one transaction). → counts. */++function moveUser(d, a, b, mergeId) {++ const out = { follows: 0, followers: 0, follows_dropped: 0, points: 0, points_summed: 0, streams: 0 };++ // Follows: as the follower, then as the followed channel.++ for (const [col, other, key] of [['follower_id', 'streamer_id', 'follows'], ['streamer_id', 'follower_id', 'followers']]) {++ for (const f of d.prepare(`SELECT id, ${other} AS o FROM follows WHERE ${col} = ?`).all(a)) {++ const clash = f.o === b || d.prepare(`SELECT 1 FROM follows WHERE ${col} = ? AND ${other} = ?`).get(b, f.o);++ if (clash) { d.prepare('DELETE FROM follows WHERE id = ?').run(f.id); out.follows_dropped++; }++ else { d.prepare(`UPDATE follows SET ${col} = ? WHERE id = ?`).run(b, f.id); out[key]++; }++ }++ }++ // Channel points (loyalty, ADR-012): first as the viewer, then as the channel; the same pair twice is summed.++ const logMove = d.prepare('INSERT OR IGNORE INTO channel_points_log (idempotency_key, user_id, streamer_id, delta, reason) VALUES (?, ?, ?, ?, ?)');++ const add = (u, s, bal) => {++ const have = d.prepare('SELECT 1 FROM channel_points WHERE user_id = ? AND streamer_id = ?').get(u, s);++ if (have) { d.prepare('UPDATE channel_points SET balance = balance + ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? AND streamer_id = ?').run(bal, u, s); out.points_summed++; }++ else d.prepare('INSERT INTO channel_points (user_id, streamer_id, balance) VALUES (?, ?, ?)').run(u, s, bal);++ };++ const move = (fromU, fromS, toU, toS, bal, key) => {++ d.prepare('DELETE FROM channel_points WHERE user_id = ? AND streamer_id = ?').run(fromU, fromS);++ add(toU, toS, bal);++ if (bal) {++ logMove.run(`live:cp:merge:${mergeId}:${key}:out`, fromU, fromS, -bal, 'account_merge');++ logMove.run(`live:cp:merge:${mergeId}:${key}:in`, toU, toS, bal, 'account_merge');++ }++ out.points++;++ };++ for (const p of d.prepare('SELECT streamer_id AS s, balance FROM channel_points WHERE user_id = ?').all(a)) move(a, p.s, b, p.s === a ? b : p.s, p.balance, `v:${p.s}`);++ for (const p of d.prepare('SELECT user_id AS u, balance FROM channel_points WHERE streamer_id = ?').all(a)) move(p.u, a, p.u, b, p.balance, `c:${p.u}`);++ out.streams = d.prepare('UPDATE streams SET user_id = ? WHERE user_id = ?').run(b, a).changes;++ d.prepare('UPDATE users SET merged_into = ? WHERE id = ?').run(b, a);++ return out;++}++++/**++ * Apply one network.subject.merged envelope → 'merged' | 'relinked' | 'nothing' | 'unchanged' | 'ignored:<why>'.++ * resolveNetworkId(subject) → the survivor's Network user id (injected by tests; defaults to Network's resolve-batch).++ */++async function apply(ev, { resolveNetworkId = defaultResolve } = {}) {++ const p = ev && 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';++ const d = db.getDb();++ ensureSchema(d);++ if (d.prepare('SELECT 1 FROM subject_merges WHERE merge_id = ?').get(p.merge_id)) return 'unchanged';++ const a = liveUserOf(d, p.from);++ const b = liveUserOf(d, p.into);++ let result; let outcome;++ if (!a) { result = 'nothing'; outcome = { reason: 'no Live user for the folded-in account' }; }++ else if (!b) {++ const networkId = await resolveNetworkId(p.into);++ try {++ d.prepare("UPDATE linked_accounts SET subject_id = ?, service_user_id = COALESCE(?, service_user_id) WHERE service = 'network' AND user_id = ?")++ .run(p.into, networkId == null ? null : String(networkId), a);++ result = 'relinked'; outcome = { user: a, network_user_id: networkId == null ? null : Number(networkId) };++ } catch (e) {++ // Another Live user already carries the survivor's Network id without its subject: left for staff, logged.++ result = 'conflict'; outcome = { user: a, error: String(e.message).slice(0, 200) };++ }++ } else if (a === b) { result = 'nothing'; outcome = { reason: 'one Live user for both' }; }++ else {++ outcome = d.transaction(() => moveUser(d, a, b, p.merge_id))();++ result = 'merged';++ }++ d.prepare('INSERT OR IGNORE INTO subject_merges (merge_id, from_subject, into_subject, from_user_id, into_user_id, outcome) VALUES (?, ?, ?, ?, ?, ?)')++ .run(p.merge_id, p.from, p.into, a, b, JSON.stringify({ result, ...outcome }));++ console.log(`[Merge] ${p.merge_id}: ${result} ${JSON.stringify(outcome)}`);++ return result;++}++++async function defaultResolve(subject) {++ try {++ const { networkPost } = require('./identity-sync');++ const r = await networkPost('/internal/identity/resolve-batch', { subject_ids: [subject] });++ const hit = r && r.results && r.results[subject];++ return hit && hit.network_user_id != null ? hit.network_user_id : null;++ } catch { return null; }++}++++module.exports = { apply, ensureSchema, moveUser };nnew file mode 100644@@@ -0,0 +1,94 @@++'use strict';++// network.subject.merged → Live (roadmap WS-B task 5, ADR-029): through POST /internal/network-events (signed), the++// folded-in Live user's follows (both ways), channel points (both ways, a shared pair summed, every move logged once)++// and streams move to the survivor's user; a follow it already has and following oneself are dropped; the folded-in++// user is marked merged_into. With only the folded-in account on Live, that user is relinked to the survivor. A++// redelivery changes nothing; a merge nobody on Live had is acknowledged. The subscription asks for the topic.++// node test/subject-merge.test.js++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-live-merge-'));++process.env.DB_PATH = path.join(tmp, 'live.db');++process.env.NODE_ENV = 'test';++process.env.LIVE_EVENTS_SECRET = 's'.repeat(40);++const quiet = console.log;++console.log = (...a) => { if (!/^\[/.test(String(a[0]))) quiet(...a); };++console.warn = () => {};++++const express = require('express');++const { ids, validate } = require('openvibe-contracts');++const { signDeliveryHeaders } = require('openvibe-sdk/events');++const db = require('../server/db/database');++db.initDb();++const networkEvents = require('../server/auth/network-events');++const subjectMerge = require('../server/auth/subject-merge');++const { NETWORK_TOPICS } = require('../scripts/subscribe-media-events');++++const KEEP = ids.newId('user'), FOLD = ids.newId('user'), X = ids.newId('user'), LONE = ids.newId('user'), LONE_INTO = ids.newId('user');++const d = db.getDb();++d.prepare(`INSERT INTO users (id, username, password_hash) VALUES (10, 'keep', '$sso$'), (11, 'fold', '$sso$'), (12, 'xena', '$sso$'), (13, 'yuri', '$sso$'), (14, 'lone', '$sso$')`).run();++d.prepare(`INSERT INTO linked_accounts (user_id, service, service_user_id, service_username, subject_id) VALUES++ (10, 'network', '100', 'keep', ?), (11, 'network', '101', 'fold', ?), (12, 'network', '102', 'xena', ?), (14, 'network', '104', 'lone', ?)`).run(KEEP, FOLD, X, LONE);++// Follows: fold follows xena (keep does too: a clash), fold follows yuri, yuri follows fold's channel, fold follows keep (self after merge).++d.prepare('INSERT INTO follows (follower_id, streamer_id) VALUES (11, 12), (10, 12), (11, 13), (13, 11), (11, 10)').run();++// Channel points: fold in xena's channel (keep has some there: summed), yuri in fold's channel, fold in keep's channel.++d.prepare('INSERT INTO channel_points (user_id, streamer_id, balance) VALUES (11, 12, 30), (10, 12, 5), (13, 11, 7), (11, 10, 9)').run();++d.prepare("INSERT INTO streams (user_id, title) VALUES (11, 'old stream'), (11, 'another'), (10, 'mine')").run();++++const envelope = (from, into, mergeId = `mrg_${ids.ulid()}`) => {++ const payload = { merge_id: mergeId, from, into, merged_at: new Date().toISOString(), initiated_by: 'person', split_until: new Date(Date.now() + 30 * 86400000).toISOString() };++ assert.ok(validate('network.subject.merged@1', payload).valid);++ return { event_id: ids.newId('event'), event_type: 'network.subject.merged', version: 1, source: 'network', actor: { type: 'user', id: into },++ timestamp: payload.merged_at, visibility: 'internal', subject: { type: 'user', id: into }, payload };++};++++(async () => {++ const app = express();++ app.use(express.json({ verify: (req, res, buf) => { req.rawBody = buf; } }));++ app.post('/internal/network-events', networkEvents.handler);++ const server = http.createServer(app);++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ const url = `http://127.0.0.1:${server.address().port}/internal/network-events`;++ const send = async (ev) => { const raw = JSON.stringify({ event: ev, seq: 1 }); return (await fetch(url, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw, process.env.LIVE_EVENTS_SECRET) }, body: raw })).status; };++ try {++ assert.ok(NETWORK_TOPICS.includes('network.subject.merged'), 'the subscription asks for merges');++++ const ev = envelope(FOLD, KEEP);++ assert.strictEqual(await send(ev), 204);++ const follows = d.prepare('SELECT follower_id AS a, streamer_id AS b FROM follows ORDER BY a, b').all().map((f) => `${f.a}>${f.b}`);++ assert.deepStrictEqual(follows, ['10>12', '10>13', '13>10'], 'moved; the clash with keep>xena and fold>keep (now self) dropped');++ const points = d.prepare('SELECT user_id AS u, streamer_id AS s, balance FROM channel_points ORDER BY u, s').all().map((p) => `${p.u}@${p.s}=${p.balance}`);++ assert.deepStrictEqual(points, ['10@10=9', '10@12=35', '13@10=7'], 'summed where both had points; the channel side moved too');++ const log = d.prepare("SELECT delta FROM channel_points_log WHERE reason = 'account_merge'").all().map((l) => l.delta);++ assert.strictEqual(log.reduce((n, x) => n + x, 0), 0, 'every move logged out and in');++ assert.strictEqual(log.length, 6);++ assert.deepStrictEqual(d.prepare('SELECT user_id FROM streams ORDER BY id').all().map((s) => s.user_id), [10, 10, 10]);++ assert.strictEqual(d.prepare('SELECT merged_into FROM users WHERE id = 11').get().merged_into, 10);++ const rec = d.prepare('SELECT * FROM subject_merges WHERE merge_id = ?').get(ev.payload.merge_id);++ assert.deepStrictEqual([rec.from_user_id, rec.into_user_id, JSON.parse(rec.outcome).result], [11, 10, 'merged']);++++ // A redelivery changes nothing.++ assert.strictEqual(await send(ev), 204);++ assert.strictEqual(await subjectMerge.apply(ev), 'unchanged');++ assert.strictEqual(d.prepare("SELECT COUNT(*) AS n FROM channel_points_log WHERE reason = 'account_merge'").get().n, 6);++++ // Only the folded-in account had a Live user: it becomes the survivor's.++ const r = await subjectMerge.apply(envelope(LONE, LONE_INTO), { resolveNetworkId: async (s) => (s === LONE_INTO ? 205 : null) });++ assert.strictEqual(r, 'relinked');++ assert.deepStrictEqual(d.prepare("SELECT subject_id, service_user_id FROM linked_accounts WHERE user_id = 14 AND service = 'network'").get(), { subject_id: LONE_INTO, service_user_id: '205' });++++ // Nobody on Live: acknowledged, recorded.++ assert.strictEqual(await subjectMerge.apply(envelope(ids.newId('user'), ids.newId('user'))), 'nothing');++ // Not Network's, or a malformed payload: ignored.++ assert.strictEqual(await networkEvents.apply({ ...envelope(FOLD, KEEP), source: 'media' }), 'ignored:source');++ assert.strictEqual(await subjectMerge.apply({ payload: { merge_id: 'x', from: FOLD, into: KEEP } }), 'ignored:payload');++ } finally {++ server.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ }++ console.log('subject merge: all checks passed');++})().catch((e) => { console.error(e); process.exit(1); });
@@ -25,7 +25,7 @@
@@ -25,7 +25,7 @@
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
"onnxruntime-node": "^1.20.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.68.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.70.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.23.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.23.0",
"sanitize-html": "^2.17.7",
"sanitize-html": "^2.17.7",
@@ -2814,9 +2814,9 @@
@@ -2814,9 +2814,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.68.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.68.0",
- "integrity": "sha512-5XJ/MwzJj4vQlDyMqB/zZf59Bw/Cq7M14UTBJ23uPzXLjRTTQx1jO6/tV76l8/j1C95iuM/Hm7gkFn72X843og==",
+ "version": "0.70.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.70.0",
+ "integrity": "sha512-NUjKoaP66vsRfyurIXhjzNja/9ELwD8smsMGJ0BuZ0LFpRocTc3yc9rCtHfIztwiuO50sC3E26J3DDlJUvpJyg==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -44,7 +44,7 @@
@@ -44,7 +44,7 @@
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
"onnxruntime-node": "^1.20.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.68.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.70.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.23.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.23.0",
"sanitize-html": "^2.17.7",
"sanitize-html": "^2.17.7",
@@ -31,7 +31,7 @@ const fs = require('fs');
@@ -31,7 +31,7 @@ const fs = require('fs');
const TOPICS = ['media.vod.*', 'media.clip.*', 'media.storage.*'];
const TOPICS = ['media.vod.*', 'media.clip.*', 'media.storage.*'];
const DEFAULT_ENDPOINT = 'http://127.0.0.1:3000/internal/media-events';
const DEFAULT_ENDPOINT = 'http://127.0.0.1:3000/internal/media-events';
-const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted'];
+const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted', 'network.subject.merged'];
const NETWORK_ENDPOINT = 'http://127.0.0.1:3000/internal/network-events';
const NETWORK_ENDPOINT = 'http://127.0.0.1:3000/internal/network-events';
const DEFAULT_LIVE_ENV = '/etc/openvibe/live.env';
const DEFAULT_LIVE_ENV = '/etc/openvibe/live.env';
@@ -97,4 +97,4 @@ async function backfillSubjects() {
@@ -97,4 +97,4 @@ async function backfillSubjects() {
return total;
return total;
}
}
-module.exports = { noteSubject, subjectOf, syncLegacyMap, backfillSubjects };
+module.exports = { noteSubject, subjectOf, syncLegacyMap, backfillSubjects, networkPost };
@@ -10,6 +10,8 @@
@@ -10,6 +10,8 @@
* token_valid_after the person's older tokens are refused (./revocations.js)
* token_valid_after the person's older tokens are refused (./revocations.js)
* updated the person's profile, role and ban (./subject-projection.js, WS-B task 2)
* updated the person's profile, role and ban (./subject-projection.js, WS-B task 2)
* follow.created/.deleted Live's follows table as a projection of Network's graph (../social/network-follows.js)
* follow.created/.deleted Live's follows table as a projection of Network's graph (../social/network-follows.js)
+ * subject.merged two accounts became one (ADR-029): the folded-in Live user's follows, channel points and streams
+ * move to the survivor's (./subject-merge.js), once per merge
*
*
* Applying is idempotent (a cutoff only ever moves forward; a profile revision only up), so a redelivery
* Applying is idempotent (a cutoff only ever moves forward; a profile revision only up), so a redelivery
* is a no-op without an inbox. Anything signed but not ours to act on is acknowledged (204).
* is a no-op without an inbox. Anything signed but not ours to act on is acknowledged (204).
@@ -23,9 +25,10 @@ const notifyChat = (userId) => { try { const cs = require('../chat/chat-server')
@@ -23,9 +25,10 @@ const notifyChat = (userId) => { try { const cs = require('../chat/chat-server')
function secret() { return process.env.LIVE_EVENTS_SECRET || process.env.MEDIA_EVENTS_SECRET || ''; }
function secret() { return process.env.LIVE_EVENTS_SECRET || process.env.MEDIA_EVENTS_SECRET || ''; }
-/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>'. */
+/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>' (a merge: a promise of 'merged' | 'relinked' | 'nothing' | …). */
function apply(ev) {
function apply(ev) {
if (!ev || !EVENT_ID_RE.test(String(ev.event_id || ''))) return 'ignored:envelope';
if (!ev || !EVENT_ID_RE.test(String(ev.event_id || ''))) return 'ignored:envelope';
+ if (ev.event_type === 'network.subject.merged') return ev.source === 'network' ? require('./subject-merge').apply(ev) : 'ignored:source';
const follow = ev.event_type === 'network.follow.created' || ev.event_type === 'network.follow.deleted';
const follow = ev.event_type === 'network.follow.created' || ev.event_type === 'network.follow.deleted';
if (ev.event_type !== 'network.user.token_valid_after' && ev.event_type !== 'network.user.updated' && !follow) return 'ignored:type';
if (ev.event_type !== 'network.user.token_valid_after' && ev.event_type !== 'network.user.updated' && !follow) return 'ignored:type';
if (ev.source !== 'network') return 'ignored:source';
if (ev.source !== 'network') return 'ignored:source';
@@ -47,8 +50,13 @@ function handler(req, res) {
@@ -47,8 +50,13 @@ function handler(req, res) {
const delivery = parseDelivery(req.rawBody, req.headers, key, { requireV2: true });
const delivery = parseDelivery(req.rawBody, req.headers, key, { requireV2: true });
if (!delivery) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }
if (!delivery) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }
stats.received++;
stats.received++;
+ const count = (out) => { if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated' || out === 'followed' || out === 'unfollowed' || out === 'merged' || out === 'relinked') stats.profiles++; else stats.ignored++; };
const out = apply(delivery.event);
const out = apply(delivery.event);
- if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated' || out === 'followed' || out === 'unfollowed') stats.profiles++; else stats.ignored++;
+ if (out && typeof out.then === 'function') {
+ // A merge (the one async case): answer after it applied, so a failure is redelivered.
+ return out.then((o) => { count(o); res.status(204).end(); }, (e) => { console.error('[NetworkEvents] merge failed:', e.message); res.status(500).json({ error: 'merge not applied' }); });
+ }
+ count(out);
res.status(204).end();
res.status(204).end();
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,121 @@
@@ -0,0 +1,121 @@
+'use strict';
+/**
+ * network.subject.merged → Live (roadmap WS-B task 5, ADR-029; Contracts 0.69.0). Two Network accounts became one:
+ * `from` is now an alias of `into`. Live has at most one user linked to each (linked_accounts, service 'network'):
+ *
+ * neither, or only the survivor's nothing of Live's to move
+ * only the folded-in account's that Live user becomes the survivor's: its network link names `into` (and the
+ * survivor's Network user id, resolved from Network)
+ * both in one transaction, the folded-in user's follows (as follower and as the followed
+ * channel), channel points (as viewer and as the channel; balances of the same pair
+ * are summed, each move logged in channel_points_log once) and streams move to the
+ * survivor's user. A follow the survivor already has, or following oneself, is
+ * dropped and counted. The folded-in Live user stays, marked merged_into.
+ *
+ * Live's follows are a projection of Network's graph (ADR-030), which Network already merged; this keeps the
+ * projection in step without waiting for a rebuild. VODs and clips are Media's (it follows the same event), chat is
+ * Chat's. Applied once per merge_id (subject_merges).
+ */
+const db = require('../db/database');
+
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const MERGE_RE = /^mrg_[0-9A-HJKMNP-TV-Z]{26}$/;
+
+function ensureSchema(d) {
+ d.exec(`CREATE TABLE IF NOT EXISTS subject_merges (
+ merge_id TEXT PRIMARY KEY,
+ from_subject TEXT NOT NULL,
+ into_subject TEXT NOT NULL,
+ from_user_id INTEGER,
+ into_user_id INTEGER,
+ outcome TEXT NOT NULL,
+ applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
+ )`);
+ const cols = d.prepare('PRAGMA table_info(users)').all().map((c) => c.name);
+ if (!cols.includes('merged_into')) d.exec('ALTER TABLE users ADD COLUMN merged_into INTEGER');
+}
+
+const liveUserOf = (d, subject) => {
+ const r = d.prepare("SELECT user_id FROM linked_accounts WHERE service = 'network' AND subject_id = ? ORDER BY id LIMIT 1").get(subject);
+ return r ? r.user_id : null;
+};
+
+/** Move everything of Live user `a` to Live user `b` (one transaction). → counts. */
+function moveUser(d, a, b, mergeId) {
+ const out = { follows: 0, followers: 0, follows_dropped: 0, points: 0, points_summed: 0, streams: 0 };
+ // Follows: as the follower, then as the followed channel.
+ for (const [col, other, key] of [['follower_id', 'streamer_id', 'follows'], ['streamer_id', 'follower_id', 'followers']]) {
+ for (const f of d.prepare(`SELECT id, ${other} AS o FROM follows WHERE ${col} = ?`).all(a)) {
+ const clash = f.o === b || d.prepare(`SELECT 1 FROM follows WHERE ${col} = ? AND ${other} = ?`).get(b, f.o);
+ if (clash) { d.prepare('DELETE FROM follows WHERE id = ?').run(f.id); out.follows_dropped++; }
+ else { d.prepare(`UPDATE follows SET ${col} = ? WHERE id = ?`).run(b, f.id); out[key]++; }
+ }
+ }
+ // Channel points (loyalty, ADR-012): first as the viewer, then as the channel; the same pair twice is summed.
+ const logMove = d.prepare('INSERT OR IGNORE INTO channel_points_log (idempotency_key, user_id, streamer_id, delta, reason) VALUES (?, ?, ?, ?, ?)');
+ const add = (u, s, bal) => {
+ const have = d.prepare('SELECT 1 FROM channel_points WHERE user_id = ? AND streamer_id = ?').get(u, s);
+ if (have) { d.prepare('UPDATE channel_points SET balance = balance + ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? AND streamer_id = ?').run(bal, u, s); out.points_summed++; }
+ else d.prepare('INSERT INTO channel_points (user_id, streamer_id, balance) VALUES (?, ?, ?)').run(u, s, bal);
+ };
+ const move = (fromU, fromS, toU, toS, bal, key) => {
+ d.prepare('DELETE FROM channel_points WHERE user_id = ? AND streamer_id = ?').run(fromU, fromS);
+ add(toU, toS, bal);
+ if (bal) {
+ logMove.run(`live:cp:merge:${mergeId}:${key}:out`, fromU, fromS, -bal, 'account_merge');
+ logMove.run(`live:cp:merge:${mergeId}:${key}:in`, toU, toS, bal, 'account_merge');
+ }
+ out.points++;
+ };
+ for (const p of d.prepare('SELECT streamer_id AS s, balance FROM channel_points WHERE user_id = ?').all(a)) move(a, p.s, b, p.s === a ? b : p.s, p.balance, `v:${p.s}`);
+ for (const p of d.prepare('SELECT user_id AS u, balance FROM channel_points WHERE streamer_id = ?').all(a)) move(p.u, a, p.u, b, p.balance, `c:${p.u}`);
+ out.streams = d.prepare('UPDATE streams SET user_id = ? WHERE user_id = ?').run(b, a).changes;
+ d.prepare('UPDATE users SET merged_into = ? WHERE id = ?').run(b, a);
+ return out;
+}
+
+/**
+ * Apply one network.subject.merged envelope → 'merged' | 'relinked' | 'nothing' | 'unchanged' | 'ignored:<why>'.
+ * resolveNetworkId(subject) → the survivor's Network user id (injected by tests; defaults to Network's resolve-batch).
+ */
+async function apply(ev, { resolveNetworkId = defaultResolve } = {}) {
+ const p = ev && 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';
+ const d = db.getDb();
+ ensureSchema(d);
+ if (d.prepare('SELECT 1 FROM subject_merges WHERE merge_id = ?').get(p.merge_id)) return 'unchanged';
+ const a = liveUserOf(d, p.from);
+ const b = liveUserOf(d, p.into);
+ let result; let outcome;
+ if (!a) { result = 'nothing'; outcome = { reason: 'no Live user for the folded-in account' }; }
+ else if (!b) {
+ const networkId = await resolveNetworkId(p.into);
+ try {
+ d.prepare("UPDATE linked_accounts SET subject_id = ?, service_user_id = COALESCE(?, service_user_id) WHERE service = 'network' AND user_id = ?")
+ .run(p.into, networkId == null ? null : String(networkId), a);
+ result = 'relinked'; outcome = { user: a, network_user_id: networkId == null ? null : Number(networkId) };
+ } catch (e) {
+ // Another Live user already carries the survivor's Network id without its subject: left for staff, logged.
+ result = 'conflict'; outcome = { user: a, error: String(e.message).slice(0, 200) };
+ }
+ } else if (a === b) { result = 'nothing'; outcome = { reason: 'one Live user for both' }; }
+ else {
+ outcome = d.transaction(() => moveUser(d, a, b, p.merge_id))();
+ result = 'merged';
+ }
+ d.prepare('INSERT OR IGNORE INTO subject_merges (merge_id, from_subject, into_subject, from_user_id, into_user_id, outcome) VALUES (?, ?, ?, ?, ?, ?)')
+ .run(p.merge_id, p.from, p.into, a, b, JSON.stringify({ result, ...outcome }));
+ console.log(`[Merge] ${p.merge_id}: ${result} ${JSON.stringify(outcome)}`);
+ return result;
+}
+
+async function defaultResolve(subject) {
+ try {
+ const { networkPost } = require('./identity-sync');
+ const r = await networkPost('/internal/identity/resolve-batch', { subject_ids: [subject] });
+ const hit = r && r.results && r.results[subject];
+ return hit && hit.network_user_id != null ? hit.network_user_id : null;
+ } catch { return null; }
+}
+
+module.exports = { apply, ensureSchema, moveUser };
new file mode 100644
new file mode 100644
@@ -0,0 +1,94 @@
@@ -0,0 +1,94 @@
+'use strict';
+// network.subject.merged → Live (roadmap WS-B task 5, ADR-029): through POST /internal/network-events (signed), the
+// folded-in Live user's follows (both ways), channel points (both ways, a shared pair summed, every move logged once)
+// and streams move to the survivor's user; a follow it already has and following oneself are dropped; the folded-in
+// user is marked merged_into. With only the folded-in account on Live, that user is relinked to the survivor. A
+// redelivery changes nothing; a merge nobody on Live had is acknowledged. The subscription asks for the topic.
+// node test/subject-merge.test.js
+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-live-merge-'));
+process.env.DB_PATH = path.join(tmp, 'live.db');
+process.env.NODE_ENV = 'test';
+process.env.LIVE_EVENTS_SECRET = 's'.repeat(40);
+const quiet = console.log;
+console.log = (...a) => { if (!/^\[/.test(String(a[0]))) quiet(...a); };
+console.warn = () => {};
+
+const express = require('express');
+const { ids, validate } = require('openvibe-contracts');
+const { signDeliveryHeaders } = require('openvibe-sdk/events');
+const db = require('../server/db/database');
+db.initDb();
+const networkEvents = require('../server/auth/network-events');
+const subjectMerge = require('../server/auth/subject-merge');
+const { NETWORK_TOPICS } = require('../scripts/subscribe-media-events');
+
+const KEEP = ids.newId('user'), FOLD = ids.newId('user'), X = ids.newId('user'), LONE = ids.newId('user'), LONE_INTO = ids.newId('user');
+const d = db.getDb();
+d.prepare(`INSERT INTO users (id, username, password_hash) VALUES (10, 'keep', '$sso$'), (11, 'fold', '$sso$'), (12, 'xena', '$sso$'), (13, 'yuri', '$sso$'), (14, 'lone', '$sso$')`).run();
+d.prepare(`INSERT INTO linked_accounts (user_id, service, service_user_id, service_username, subject_id) VALUES
+ (10, 'network', '100', 'keep', ?), (11, 'network', '101', 'fold', ?), (12, 'network', '102', 'xena', ?), (14, 'network', '104', 'lone', ?)`).run(KEEP, FOLD, X, LONE);
+// Follows: fold follows xena (keep does too: a clash), fold follows yuri, yuri follows fold's channel, fold follows keep (self after merge).
+d.prepare('INSERT INTO follows (follower_id, streamer_id) VALUES (11, 12), (10, 12), (11, 13), (13, 11), (11, 10)').run();
+// Channel points: fold in xena's channel (keep has some there: summed), yuri in fold's channel, fold in keep's channel.
+d.prepare('INSERT INTO channel_points (user_id, streamer_id, balance) VALUES (11, 12, 30), (10, 12, 5), (13, 11, 7), (11, 10, 9)').run();
+d.prepare("INSERT INTO streams (user_id, title) VALUES (11, 'old stream'), (11, 'another'), (10, 'mine')").run();
+
+const envelope = (from, into, mergeId = `mrg_${ids.ulid()}`) => {
+ const payload = { merge_id: mergeId, from, into, merged_at: new Date().toISOString(), initiated_by: 'person', split_until: new Date(Date.now() + 30 * 86400000).toISOString() };
+ assert.ok(validate('network.subject.merged@1', payload).valid);
+ return { event_id: ids.newId('event'), event_type: 'network.subject.merged', version: 1, source: 'network', actor: { type: 'user', id: into },
+ timestamp: payload.merged_at, visibility: 'internal', subject: { type: 'user', id: into }, payload };
+};
+
+(async () => {
+ const app = express();
+ app.use(express.json({ verify: (req, res, buf) => { req.rawBody = buf; } }));
+ app.post('/internal/network-events', networkEvents.handler);
+ const server = http.createServer(app);
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ const url = `http://127.0.0.1:${server.address().port}/internal/network-events`;
+ const send = async (ev) => { const raw = JSON.stringify({ event: ev, seq: 1 }); return (await fetch(url, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw, process.env.LIVE_EVENTS_SECRET) }, body: raw })).status; };
+ try {
+ assert.ok(NETWORK_TOPICS.includes('network.subject.merged'), 'the subscription asks for merges');
+
+ const ev = envelope(FOLD, KEEP);
+ assert.strictEqual(await send(ev), 204);
+ const follows = d.prepare('SELECT follower_id AS a, streamer_id AS b FROM follows ORDER BY a, b').all().map((f) => `${f.a}>${f.b}`);
+ assert.deepStrictEqual(follows, ['10>12', '10>13', '13>10'], 'moved; the clash with keep>xena and fold>keep (now self) dropped');
+ const points = d.prepare('SELECT user_id AS u, streamer_id AS s, balance FROM channel_points ORDER BY u, s').all().map((p) => `${p.u}@${p.s}=${p.balance}`);
+ assert.deepStrictEqual(points, ['10@10=9', '10@12=35', '13@10=7'], 'summed where both had points; the channel side moved too');
+ const log = d.prepare("SELECT delta FROM channel_points_log WHERE reason = 'account_merge'").all().map((l) => l.delta);
+ assert.strictEqual(log.reduce((n, x) => n + x, 0), 0, 'every move logged out and in');
+ assert.strictEqual(log.length, 6);
+ assert.deepStrictEqual(d.prepare('SELECT user_id FROM streams ORDER BY id').all().map((s) => s.user_id), [10, 10, 10]);
+ assert.strictEqual(d.prepare('SELECT merged_into FROM users WHERE id = 11').get().merged_into, 10);
+ const rec = d.prepare('SELECT * FROM subject_merges WHERE merge_id = ?').get(ev.payload.merge_id);
+ assert.deepStrictEqual([rec.from_user_id, rec.into_user_id, JSON.parse(rec.outcome).result], [11, 10, 'merged']);
+
+ // A redelivery changes nothing.
+ assert.strictEqual(await send(ev), 204);
+ assert.strictEqual(await subjectMerge.apply(ev), 'unchanged');
+ assert.strictEqual(d.prepare("SELECT COUNT(*) AS n FROM channel_points_log WHERE reason = 'account_merge'").get().n, 6);
+
+ // Only the folded-in account had a Live user: it becomes the survivor's.
+ const r = await subjectMerge.apply(envelope(LONE, LONE_INTO), { resolveNetworkId: async (s) => (s === LONE_INTO ? 205 : null) });
+ assert.strictEqual(r, 'relinked');
+ assert.deepStrictEqual(d.prepare("SELECT subject_id, service_user_id FROM linked_accounts WHERE user_id = 14 AND service = 'network'").get(), { subject_id: LONE_INTO, service_user_id: '205' });
+
+ // Nobody on Live: acknowledged, recorded.
+ assert.strictEqual(await subjectMerge.apply(envelope(ids.newId('user'), ids.newId('user'))), 'nothing');
+ // Not Network's, or a malformed payload: ignored.
+ assert.strictEqual(await networkEvents.apply({ ...envelope(FOLD, KEEP), source: 'media' }), 'ignored:source');
+ assert.strictEqual(await subjectMerge.apply({ payload: { merge_id: 'x', from: FOLD, into: KEEP } }), 'ignored:payload');
+ } finally {
+ server.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ }
+ console.log('subject merge: all checks passed');
+})().catch((e) => { console.error(e); process.exit(1); });