openvibers/openvibe.live Commits
Live takes part in account export and deletion (roadmap WS-B task 7, ADR-033; openvibe-contracts 0.70.0 -> 0.71.0). network.account.export_requested and network.account.deleted arrive at POST /internal/network-events (the --network subscription now asks for both) and server/auth/account-data.js applies them once per export or deletion (account_data_events), answering after Network took the part or confirmation, so a failure is redelivered without erasing twice.
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.70.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.71.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.1","sanitize-html": "^2.17.7",@@@ -2814,9 +2814,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.70.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.70.0",-- "integrity": "sha512-NUjKoaP66vsRfyurIXhjzNja/9ELwD8smsMGJ0BuZ0LFpRocTc3yc9rCtHfIztwiuO50sC3E26J3DDlJUvpJyg==",++ "version": "0.71.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.71.0",++ "integrity": "sha512-BLS6gP/MKuwhvTdWO8UYUaOyhmf7ni0cQ8r7150o57R5TrCnzdWQEVESxfMFCGgSbm2wdKtdZwTID69nxD14Tg==","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.70.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.71.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.1","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', 'network.subject.merged'];++const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted', 'network.subject.merged', 'network.account.export_requested', 'network.account.deleted'];const NETWORK_ENDPOINT = 'http://127.0.0.1:3000/internal/network-events';const DEFAULT_LIVE_ENV = '/etc/openvibe/live.env';nnew file mode 100644@@@ -0,0 +1,233 @@++'use strict';++/**++ * Account export and deletion → Live (roadmap WS-B task 7, ADR-033; Contracts 0.71.0).++ *++ * network.account.export_requested Live pushes its part of the person's export to Network++ * (POST /internal/account-exports/:id/parts, network.account.export.contribute):++ * the Live profile and, per table, the rows about the person (secrets, stream keys++ * and tokens never). A person with no Live account sends an empty part, so the++ * export does not wait for Live.++ * network.account.deleted Live erases the person's rows in one transaction, then confirms with counts++ * (POST /internal/account-deletions/:id/confirmations, network.account.deletion.confirm).++ *++ * The tables come from the schema itself: every foreign key to users says what the person owns (ON DELETE CASCADE:++ * the row goes, and with a stream or channel go the rows inside it) and what is shared (SET NULL: the row stays++ * without them, such as their chat lines, canvas tiles and themes). A user_id column without a foreign key is owned.++ * Kept, as ADR-012 and ADR-033 say:++ * - money and ledgers (payment orders, subscriptions, transactions, coin and channel-point logs), with the donation++ * message cleared while Live is the ledger;++ * - moderation records (bans, moderation actions), and staff bookkeeping.++ * The user row stays as a tombstone (username released, email, password and profile cleared, deleted_at), so the++ * kept rows point at nobody.++ *++ * Never touched: the frozen tables (vods, clips, pastes, paste_likes, paste_comments, comments), which Media and++ * Community own and erase themselves. The staged chat tables (chat-tables.js) are left to Chat, which erases its own++ * copy; Live's rows there follow when a table moves.++ *++ * Applied once per export or deletion (account_data_events); a redelivery resends only what did not reach Network.++ */++const db = require('../db/database');++++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const EXPORT_RE = /^exp_[0-9A-HJKMNP-TV-Z]{26}$/;++const DELETION_RE = /^del_[0-9A-HJKMNP-TV-Z]{26}$/;++const FROZEN = new Set(['vods', 'clips', 'pastes', 'paste_likes', 'paste_comments', 'comments']);++const RETAIN = new Set(['payment_orders', 'subscriptions', 'transactions', 'coin_transactions', 'channel_points_log', 'moderation_actions', 'bans',++ 'token_revocations', 'verification_keys', 'subject_merges', 'account_data_events']);++// Rows the schema keeps without the person (SET NULL) that are theirs to erase all the same: their own chat lines,++// sign-in IPs and VPN requests.++const OVERRIDE_DELETE = new Set(['chat_messages.user_id', 'ip_log.user_id', 'vpn_approvals.user_id']);++// Copies of the person's name kept beside a person column: cleared wherever the row stays.++const NAME_COLS = ['username', 'display_name', 'user_name', 'sender_name', 'sender_username', 'author_name', 'author_username', 'avatar_url', 'user_avatar'];++// Person columns the schema declares without a foreign key.++const EXTRA = [['channel_points', 'streamer_id'], ['arena_mic_moments', 'target_user_id'], ['arena_topics', 'created_by'], ['arena_topics', 'target_user_id']];++const SECRET_COL = /(^|_)(token|tokens|secret|hash|password|code|key|keys|p256dh|auth|cookie|cookies|credentials)($|_)/i;++const ROW_LIMIT = 2000;++const PART_BUDGET = 18 * 1024 * 1024;++++function staged() { return new Set(Object.keys(require('../chat/chat-tables').TABLES)); }++++function ensureSchema(d) {++ d.exec(`CREATE TABLE IF NOT EXISTS account_data_events (++ id TEXT PRIMARY KEY,++ kind TEXT NOT NULL,++ subject TEXT NOT NULL,++ outcome TEXT,++ sent_at DATETIME,++ applied_at DATETIME DEFAULT CURRENT_TIMESTAMP++ )`);++ const cols = d.prepare('PRAGMA table_info(users)').all().map((c) => c.name);++ if (!cols.includes('deleted_at')) d.exec('ALTER TABLE users ADD COLUMN deleted_at DATETIME');++}++++const q = (name) => `"${String(name).replace(/"/g, '""')}"`;++++/** table → [{ col, action: 'delete'|'null' }], every person column Live has, frozen tables left out. */++function personColumns(d) {++ const out = new Map();++ const add = (t, col, action) => {++ if (!out.has(t)) out.set(t, []);++ if (!out.get(t).some((c) => c.col === col)) out.get(t).push({ col, action });++ };++ const tables = d.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name").all().map((r) => r.name);++ for (const t of tables) {++ if (t === 'users' || FROZEN.has(t)) continue;++ for (const fk of d.prepare(`PRAGMA foreign_key_list(${q(t)})`).all()) {++ if (fk.table === 'users') add(t, fk.from, fk.on_delete === 'SET NULL' && !OVERRIDE_DELETE.has(`${t}.${fk.from}`) ? 'null' : 'delete');++ }++ if (d.prepare(`PRAGMA table_info(${q(t)})`).all().some((c) => c.name === 'user_id')) add(t, 'user_id', 'delete');++ }++ for (const [t, col] of EXTRA) {++ if (tables.includes(t) && d.prepare(`PRAGMA table_info(${q(t)})`).all().some((c) => c.name === col)) add(t, col, 'delete');++ }++ return out;++}++++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;++};++++// ── Export ─────────────────────────────────────────────────────++++/** Live's part for one person → network.account-export-part@1 (without the subject). */++function exportPart(d, userId) {++ if (!userId) return { files: [], note: 'No Live account.' };++ const files = [];++ const truncated = [];++ let bytes = 0;++ const user = d.prepare('SELECT * FROM users WHERE id = ?').get(userId) || {};++ const profile = {};++ for (const [k, v] of Object.entries(user)) if (!SECRET_COL.test(k)) profile[k] = v;++ files.push({ name: 'profile.json', content: profile });++ bytes += JSON.stringify(profile).length;++ const other = {};++ for (const [t, cols] of personColumns(d)) {++ const keep = d.prepare(`PRAGMA table_info(${q(t)})`).all().map((c) => c.name).filter((c) => !SECRET_COL.test(c));++ if (!keep.length) continue;++ const where = cols.map((c) => `${q(c.col)} = ?`).join(' OR ');++ const rows = d.prepare(`SELECT ${keep.map(q).join(', ')} FROM ${q(t)} WHERE ${where} ORDER BY rowid DESC LIMIT ${ROW_LIMIT + 1}`).all(...cols.map(() => userId));++ if (!rows.length) continue;++ const content = rows.slice(0, ROW_LIMIT);++ const size = JSON.stringify(content).length;++ if (bytes + size > PART_BUDGET) { truncated.push(`${t}.json`); continue; }++ bytes += size;++ if (rows.length > ROW_LIMIT) truncated.push(`${t}.json`);++ // At most 64 files per part: the first 60 tables get their own, the rest share other.json.++ if (files.length < 60) files.push({ name: `${t}.json`.replace(/[^a-z0-9_.-]/g, '_'), content });++ else other[t] = content;++ }++ if (Object.keys(other).length) files.push({ name: 'other.json', content: other });++ return { files, truncated, note: 'Recordings and clips are in the media part, pastes and comments in the community part, chat history in the chat part.' };++}++++// ── Deletion ───────────────────────────────────────────────────++++/** Erase Live's rows of these Live users in one transaction → { erased, retained }. */++function eraseUsers(d, userIds, { now = new Date().toISOString() } = {}) {++ const erased = {};++ const retained = {};++ const bump = (o, k, n) => { if (n) o[k] = (o[k] || 0) + n; };++ const stagedTables = staged();++ const cols = personColumns(d);++ const usersCols = d.prepare('PRAGMA table_info(users)').all().map((c) => c.name);++ const onBilling = (() => { try { return require('../monetization/money-authority').onBilling(); } catch { return false; } })();++ d.transaction(() => {++ for (const uid of userIds) {++ for (const [t, list] of cols) {++ const where = list.map((c) => `${q(c.col)} = ?`).join(' OR ');++ if (RETAIN.has(t)) { bump(retained, t, d.prepare(`SELECT COUNT(*) AS n FROM ${q(t)} WHERE ${where}`).get(...list.map(() => uid)).n); continue; }++ if (stagedTables.has(t)) { bump(retained, 'left_for_chat', d.prepare(`SELECT COUNT(*) AS n FROM ${q(t)} WHERE ${where}`).get(...list.map(() => uid)).n); continue; }++ for (const c of list) {++ if (c.action === 'null') {++ const info = d.prepare(`PRAGMA table_info(${q(t)})`).all().filter((x) => NAME_COLS.includes(x.name));++ if (info.length) d.prepare(`UPDATE ${q(t)} SET ${info.map((x) => `${q(x.name)} = ${x.notnull ? "'deleted'" : 'NULL'}`).join(', ')} WHERE ${q(c.col)} = ?`).run(uid);++ }++ const r = c.action === 'null'++ ? d.prepare(`UPDATE ${q(t)} SET ${q(c.col)} = NULL WHERE ${q(c.col)} = ?`).run(uid)++ : d.prepare(`DELETE FROM ${q(t)} WHERE ${q(c.col)} = ?`).run(uid);++ bump(erased, t, r.changes);++ }++ }++ // A donation's message is the person's words: cleared while Live keeps the ledger (read-only under Billing).++ if (!onBilling && cols.has('transactions')) {++ d.prepare('UPDATE transactions SET message = NULL WHERE from_user_id = ? AND message IS NOT NULL').run(uid);++ }++ // The tombstone: the kept rows point at nobody; the username is free again.++ const set = { username: `deleted-${uid}`, email: null, password_hash: '!deleted', display_name: null, avatar_url: null, bio: null, stream_key: null,++ profile_color: null, theme_id: null, avatar_paste_id: null, ban_reason: null, deleted_at: now };++ const keys = Object.keys(set).filter((k) => usersCols.includes(k));++ d.prepare(`UPDATE users SET ${keys.map((k) => `${q(k)} = ?`).join(', ')} WHERE id = ?`).run(...keys.map((k) => set[k]), uid);++ bump(erased, 'accounts', 1);++ }++ })();++ return { erased, retained };++}++++// ── Events ─────────────────────────────────────────────────────++++async function networkCall(path, body) {++ const principal = require('../net/network-principal');++ const base = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');++ for (let attempt = 0; attempt < 2; attempt++) {++ const headers = await principal.serviceHeaders('openvibe.network');++ const res = await fetch(`${base}${path}`, { method: 'POST', headers: { 'Content-Type': 'application/json', ...headers }, body: JSON.stringify(body), signal: AbortSignal.timeout(30_000) });++ if (res.status === 401 && attempt === 0) { principal.invalidate('openvibe.network'); continue; }++ return res;++ }++ return null;++}++++/**++ * Apply one envelope → 'exported' | 'erased' | 'confirmed' | 'closed' | 'ignored:<why>'. Throws when Network could not++ * be reached (the delivery is answered 500 and redelivered). send(path, body) → Response (tests inject it).++ */++async function apply(ev, { send = networkCall } = {}) {++ const p = ev && ev.payload && typeof ev.payload === 'object' ? ev.payload : {};++ const d = db.getDb();++ ensureSchema(d);++ if (ev.event_type === 'network.account.export_requested') {++ if (!EXPORT_RE.test(String(p.export_id || '')) || !SUBJECT_RE.test(String(p.subject || ''))) return 'ignored:payload';++ const seen = d.prepare('SELECT sent_at FROM account_data_events WHERE id = ?').get(p.export_id);++ if (seen && seen.sent_at) return 'unchanged';++ const part = exportPart(d, liveUserOf(d, p.subject));++ const res = await send(`/internal/account-exports/${p.export_id}/parts`, { subject: p.subject, ...part });++ if (!res) throw new Error('Network unreachable');++ const outcome = res.ok ? 'exported' : (res.status === 409 || res.status === 404 ? 'closed' : null);++ if (!outcome) throw new Error(`export part refused: ${res.status}`);++ d.prepare('INSERT OR REPLACE INTO account_data_events (id, kind, subject, outcome, sent_at) VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)')++ .run(p.export_id, 'export', p.subject, JSON.stringify({ result: outcome, files: part.files.length }));++ console.log(`[AccountData] export ${p.export_id}: ${outcome} (${part.files.length} file(s))`);++ return outcome;++ }++ if (ev.event_type === 'network.account.deleted') {++ if (!DELETION_RE.test(String(p.deletion_id || '')) || !SUBJECT_RE.test(String(p.subject || ''))) return 'ignored:payload';++ const subjects = [p.subject, ...(Array.isArray(p.aliases) ? p.aliases.filter((s) => SUBJECT_RE.test(String(s))) : [])];++ let rec = d.prepare('SELECT * FROM account_data_events WHERE id = ?').get(p.deletion_id);++ let result = 'confirmed';++ if (!rec) {++ const ids = [...new Set(subjects.map((s) => liveUserOf(d, s)).filter(Boolean))];++ for (const id of [...ids]) for (const m of (hasColumn(d, 'users', 'merged_into') ? d.prepare('SELECT id FROM users WHERE merged_into = ?').all(id) : [])) if (!ids.includes(m.id)) ids.push(m.id);++ const counts = eraseUsers(d, ids);++ if (d.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'subject_projection'").get()) {++ d.prepare(`DELETE FROM subject_projection WHERE subject_id IN (${subjects.map(() => '?').join(',')})`).run(...subjects);++ }++ d.prepare('INSERT INTO account_data_events (id, kind, subject, outcome) VALUES (?, ?, ?, ?)').run(p.deletion_id, 'deletion', p.subject, JSON.stringify({ users: ids, ...counts }));++ rec = d.prepare('SELECT * FROM account_data_events WHERE id = ?').get(p.deletion_id);++ result = 'erased';++ console.log(`[AccountData] deletion ${p.deletion_id}: ${JSON.stringify({ users: ids, ...counts })}`);++ }++ if (rec.sent_at) return 'unchanged';++ const o = JSON.parse(rec.outcome || '{}');++ const res = await send(`/internal/account-deletions/${p.deletion_id}/confirmations`, { subject: p.subject, completed_at: new Date(`${String(rec.applied_at).replace(' ', 'T')}Z`).toISOString(), erased: o.erased || {}, retained: o.retained || {} });++ if (!res) throw new Error('Network unreachable');++ if (!res.ok && res.status !== 404) throw new Error(`confirmation refused: ${res.status}`);++ d.prepare('UPDATE account_data_events SET sent_at = CURRENT_TIMESTAMP WHERE id = ?').run(p.deletion_id);++ return result;++ }++ return 'ignored:type';++}++++function hasColumn(d, table, col) { return d.prepare(`PRAGMA table_info(${q(table)})`).all().some((c) => c.name === col); }++++module.exports = { apply, exportPart, eraseUsers, personColumns, ensureSchema, FROZEN, RETAIN };@@@ -12,6 +12,8 @@* 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++ * account.export_requested / account.deleted the person's export part goes to Network, or their Live rows are++ * erased and the erasure confirmed (./account-data.js, ADR-033), once per export or deletion** 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).@@@ -25,10 +27,11 @@ 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>' (a merge: a promise of 'merged' | 'relinked' | 'nothing' | …). */++/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>' (a merge, export or deletion: a promise). */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';++ if (ev.event_type === 'network.account.export_requested' || ev.event_type === 'network.account.deleted') return ev.source === 'network' ? require('./account-data').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';@@@ -50,11 +53,11 @@ 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 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' || out === 'exported' || out === 'erased' || out === 'confirmed') stats.profiles++; else stats.ignored++; };const out = apply(delivery.event);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' }); });++ // A merge, export or deletion: answer after it applied, so a failure is redelivered.++ return out.then((o) => { count(o); res.status(204).end(); }, (e) => { console.error(`[NetworkEvents] ${delivery.event.event_type} failed:`, e.message); res.status(500).json({ error: 'not applied' }); });}count(out);res.status(204).end();nnew file mode 100644@@@ -0,0 +1,126 @@++'use strict';++// Account export and deletion → Live (roadmap WS-B task 7, ADR-033): through POST /internal/network-events (signed),++// network.account.export_requested sends Live's part to Network (the profile and the person's rows per table, no++// stream keys or token hashes; an empty part for someone with no Live account), and network.account.deleted erases the++// person's rows (follows and channel points both ways, streams with what is inside them, their own chat lines,++// managed streams, tokens), keeps money and moderation rows pointing at a tombstone (the donation message cleared),++// keeps shared rows without them, releases the username, and confirms with counts. A redelivery sends nothing twice;++// a failed confirmation is retried without erasing again. The subscription asks for both topics.++// node test/account-data.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-account-data-'));++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 accountData = require('../server/auth/account-data');++const { NETWORK_TOPICS } = require('../scripts/subscribe-media-events');++++const DANA = ids.newId('user'), XENA = ids.newId('user'), NOBODY = ids.newId('user');++const d = db.getDb();++d.prepare(`INSERT INTO users (id, username, email, password_hash, bio) VALUES (20, 'dana', '[email protected]', '$sso$', 'hi'), (21, 'xena', NULL, '$sso$', ''), (22, 'yuri', NULL, '$sso$', '')`).run();++d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, service_username, subject_id) VALUES (20, 'network', '200', 'dana', ?), (21, 'network', '201', 'xena', ?)").run(DANA, XENA);++d.prepare('INSERT INTO follows (follower_id, streamer_id) VALUES (20, 21), (22, 20), (22, 21)').run();++d.prepare('INSERT INTO channel_points (user_id, streamer_id, balance) VALUES (20, 21, 30), (22, 20, 7), (22, 21, 5)').run();++d.prepare("INSERT INTO streams (id, user_id, title) VALUES (1, 20, 'dana live'), (2, 21, 'xena live')").run();++// yuri chats in dana's stream (goes with the stream); dana chats in xena's (her own line goes); yuri in xena's stays.++d.prepare("INSERT INTO chat_messages (stream_id, user_id, username, message) VALUES (1, 22, 'yuri', 'hey dana'), (2, 20, 'dana', 'hey xena'), (2, 22, 'yuri', 'yo')").run();++d.prepare("INSERT INTO managed_streams (user_id, stream_key) VALUES (20, 'sk_live_secret_dana')").run();++d.prepare("INSERT INTO api_tokens (user_id, token_hash) VALUES (20, 'hash-secret-dana')").run();++d.prepare("INSERT INTO transactions (from_user_id, to_user_id, amount, type, message) VALUES (20, 21, 5, 'donation', 'love you xena'), (22, 20, 3, 'donation', 'for dana')").run();++d.prepare("INSERT INTO payment_orders (user_id, provider, provider_ref, kind, amount_cents, currency, bucks, status) VALUES (20, 'paypal', 'PO-1', 'bucks', 500, 'USD', 5, 'completed')").run();++d.prepare("INSERT INTO moderation_actions (action_type, actor_user_id, target_user_id) VALUES ('timeout', 21, 20)").run();++++const envelope = (type, payload) => {++ assert.ok(validate(`${type}@1`, payload).valid, JSON.stringify(validate(`${type}@1`, payload).errors));++ return { event_id: ids.newId('event'), event_type: type, version: 1, source: 'network', actor: { type: 'user', id: payload.subject },++ timestamp: new Date().toISOString(), visibility: 'internal', subject: { type: 'user', id: payload.subject }, payload };++};++const ulid = () => ids.ulid();++++(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 deliver = 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; };++ const sent = [];++ let failNext = false;++ const send = async (p, body) => { if (failNext) { failNext = false; return { ok: false, status: 503 }; } sent.push({ path: p, body }); return { ok: true, status: 200 }; };++ try {++ assert.ok(NETWORK_TOPICS.includes('network.account.export_requested') && NETWORK_TOPICS.includes('network.account.deleted'), 'the subscription asks for both');++ const cols = accountData.personColumns(d);++ for (const t of accountData.FROZEN) assert.ok(!cols.has(t), `${t} is never touched`);++++ // ── Export ──++ const exp = envelope('network.account.export_requested', { export_id: `exp_${ulid()}`, subject: DANA, requested_at: new Date().toISOString(), deadline: new Date(Date.now() + 1800000).toISOString() });++ assert.strictEqual(await accountData.apply(exp, { send }), 'exported');++ assert.strictEqual(sent.length, 1);++ assert.strictEqual(sent[0].path, `/internal/account-exports/${exp.payload.export_id}/parts`);++ const part = sent[0].body;++ assert.ok(validate('network.account-export-part@1', part).valid, JSON.stringify(validate('network.account-export-part@1', part).errors));++ const byName = Object.fromEntries(part.files.map((f) => [f.name, f.content]));++ assert.strictEqual(byName['profile.json'].username, 'dana');++ assert.deepStrictEqual(byName['follows.json'].map((f) => [f.follower_id, f.streamer_id]).sort(), [[20, 21], [22, 20]]);++ assert.ok(byName['transactions.json'] && byName['payment_orders.json'], 'money history is theirs to see');++ const text = JSON.stringify(part);++ for (const secret of ['sk_live_secret_dana', 'hash-secret-dana', '$sso$']) assert.ok(!text.includes(secret), `no ${secret}`);++ assert.strictEqual(await accountData.apply(exp, { send }), 'unchanged', 'a redelivery sends nothing twice');++ assert.strictEqual(sent.length, 1);++ const lone = envelope('network.account.export_requested', { export_id: `exp_${ulid()}`, subject: NOBODY, requested_at: new Date().toISOString(), deadline: new Date(Date.now() + 1800000).toISOString() });++ assert.strictEqual(await accountData.apply(lone, { send }), 'exported');++ assert.deepStrictEqual(sent[1].body.files, [], 'no Live account: an empty part, so the export does not wait');++++ // ── Deletion ──++ const del = envelope('network.account.deleted', { deletion_id: `del_${ulid()}`, subject: DANA, requested_at: new Date(Date.now() - 30 * 86400000).toISOString(), deleted_at: new Date().toISOString() });++ failNext = true;++ await assert.rejects(accountData.apply(del, { send }), /confirmation refused: 503/, 'a failed confirmation is redelivered');++ const n = (sql, ...a) => d.prepare(sql).get(...a).n;++ assert.strictEqual(n('SELECT COUNT(*) AS n FROM follows WHERE follower_id = 20 OR streamer_id = 20'), 0, 'follows both ways');++ assert.strictEqual(n('SELECT COUNT(*) AS n FROM channel_points WHERE user_id = 20 OR streamer_id = 20'), 0, 'channel points both ways');++ assert.strictEqual(n('SELECT COUNT(*) AS n FROM follows'), 1, "others' follows stay");++ assert.strictEqual(n('SELECT COUNT(*) AS n FROM streams WHERE user_id = 20'), 0);++ assert.deepStrictEqual(d.prepare('SELECT stream_id, user_id, message FROM chat_messages ORDER BY id').all(), [{ stream_id: 2, user_id: 22, message: 'yo' }],++ 'the stream took its chat with it; her own line elsewhere went; others stay');++ assert.strictEqual(n('SELECT COUNT(*) AS n FROM managed_streams WHERE user_id = 20') + n('SELECT COUNT(*) AS n FROM api_tokens WHERE user_id = 20') + n("SELECT COUNT(*) AS n FROM linked_accounts WHERE user_id = 20"), 0);++ assert.deepStrictEqual(d.prepare('SELECT from_user_id AS f, message FROM transactions ORDER BY id').all(), [{ f: 20, message: null }, { f: 22, message: 'for dana' }], 'money kept; her own message cleared');++ assert.strictEqual(n("SELECT COUNT(*) AS n FROM payment_orders WHERE user_id = 20"), 1, 'payment orders kept');++ assert.strictEqual(n('SELECT COUNT(*) AS n FROM moderation_actions WHERE target_user_id = 20'), 1, 'moderation records kept');++ const tomb = d.prepare('SELECT username, email, bio, deleted_at FROM users WHERE id = 20').get();++ assert.deepStrictEqual([tomb.username, tomb.email, tomb.bio, !!tomb.deleted_at], ['deleted-20', null, null, true]);++ d.prepare("INSERT INTO users (username, password_hash) VALUES ('dana', '$sso$')").run();++ assert.strictEqual(await accountData.apply(del, { send }), 'confirmed', 'the retry confirms without erasing again');++ const conf = sent[2];++ assert.strictEqual(conf.path, `/internal/account-deletions/${del.payload.deletion_id}/confirmations`);++ assert.ok(validate('network.account-deletion-confirmation@1', conf.body).valid, JSON.stringify(validate('network.account-deletion-confirmation@1', conf.body).errors));++ assert.ok(conf.body.erased.follows >= 2 && conf.body.erased.streams === 1 && conf.body.erased.accounts === 1, JSON.stringify(conf.body.erased));++ assert.ok(conf.body.retained.transactions === 2 && conf.body.retained.payment_orders === 1 && conf.body.retained.moderation_actions === 1, JSON.stringify(conf.body.retained));++ assert.strictEqual(await accountData.apply(del, { send }), 'unchanged');++ assert.strictEqual(sent.length, 3);++++ // Through the signed endpoint: another source is ignored (204, nothing sent), and a bad payload too.++ assert.strictEqual(await deliver({ ...envelope('network.account.deleted', { deletion_id: `del_${ulid()}`, subject: XENA, requested_at: new Date().toISOString(), deleted_at: new Date().toISOString() }), source: 'media' }), 204);++ assert.strictEqual(n('SELECT COUNT(*) AS n FROM users WHERE id = 21 AND deleted_at IS NULL'), 1, 'xena untouched');++ assert.strictEqual(await networkEvents.apply({ event_id: ids.newId('event'), event_type: 'network.account.deleted', source: 'network', payload: { deletion_id: 'x', subject: XENA } }), 'ignored:payload');++ } finally {++ server.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ }++ console.log('account export and deletion: all checks passed');++})().catch((e) => { console.error(e); process.exit(1); });@@@ -353,6 +353,7 @@ async function check(name, fn) {'server/streaming/golive-notify.js': 'OpenVibe.Network (configured)','server/utils/notify.js': 'OpenVibe.Network (configured)','server/auth/identity-sync.js': 'OpenVibe.Network (configured)',++ 'server/auth/account-data.js': 'OpenVibe.Network (configured): export parts and deletion confirmations','server/monetization/billing-client.js': 'OpenVibe.Billing (configured)','server/monetization/wallet-client.js': 'OpenVibe.Network wallet (configured)','server/monetization/payments.js': 'PayPal (fixed host)',
@@ -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.70.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.71.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.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.23.1",
"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.70.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.70.0",
- "integrity": "sha512-NUjKoaP66vsRfyurIXhjzNja/9ELwD8smsMGJ0BuZ0LFpRocTc3yc9rCtHfIztwiuO50sC3E26J3DDlJUvpJyg==",
+ "version": "0.71.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.71.0",
+ "integrity": "sha512-BLS6gP/MKuwhvTdWO8UYUaOyhmf7ni0cQ8r7150o57R5TrCnzdWQEVESxfMFCGgSbm2wdKtdZwTID69nxD14Tg==",
"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.70.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.71.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.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.23.1",
"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', 'network.subject.merged'];
+const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted', 'network.subject.merged', 'network.account.export_requested', 'network.account.deleted'];
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';
new file mode 100644
new file mode 100644
@@ -0,0 +1,233 @@
@@ -0,0 +1,233 @@
+'use strict';
+/**
+ * Account export and deletion → Live (roadmap WS-B task 7, ADR-033; Contracts 0.71.0).
+ *
+ * network.account.export_requested Live pushes its part of the person's export to Network
+ * (POST /internal/account-exports/:id/parts, network.account.export.contribute):
+ * the Live profile and, per table, the rows about the person (secrets, stream keys
+ * and tokens never). A person with no Live account sends an empty part, so the
+ * export does not wait for Live.
+ * network.account.deleted Live erases the person's rows in one transaction, then confirms with counts
+ * (POST /internal/account-deletions/:id/confirmations, network.account.deletion.confirm).
+ *
+ * The tables come from the schema itself: every foreign key to users says what the person owns (ON DELETE CASCADE:
+ * the row goes, and with a stream or channel go the rows inside it) and what is shared (SET NULL: the row stays
+ * without them, such as their chat lines, canvas tiles and themes). A user_id column without a foreign key is owned.
+ * Kept, as ADR-012 and ADR-033 say:
+ * - money and ledgers (payment orders, subscriptions, transactions, coin and channel-point logs), with the donation
+ * message cleared while Live is the ledger;
+ * - moderation records (bans, moderation actions), and staff bookkeeping.
+ * The user row stays as a tombstone (username released, email, password and profile cleared, deleted_at), so the
+ * kept rows point at nobody.
+ *
+ * Never touched: the frozen tables (vods, clips, pastes, paste_likes, paste_comments, comments), which Media and
+ * Community own and erase themselves. The staged chat tables (chat-tables.js) are left to Chat, which erases its own
+ * copy; Live's rows there follow when a table moves.
+ *
+ * Applied once per export or deletion (account_data_events); a redelivery resends only what did not reach Network.
+ */
+const db = require('../db/database');
+
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const EXPORT_RE = /^exp_[0-9A-HJKMNP-TV-Z]{26}$/;
+const DELETION_RE = /^del_[0-9A-HJKMNP-TV-Z]{26}$/;
+const FROZEN = new Set(['vods', 'clips', 'pastes', 'paste_likes', 'paste_comments', 'comments']);
+const RETAIN = new Set(['payment_orders', 'subscriptions', 'transactions', 'coin_transactions', 'channel_points_log', 'moderation_actions', 'bans',
+ 'token_revocations', 'verification_keys', 'subject_merges', 'account_data_events']);
+// Rows the schema keeps without the person (SET NULL) that are theirs to erase all the same: their own chat lines,
+// sign-in IPs and VPN requests.
+const OVERRIDE_DELETE = new Set(['chat_messages.user_id', 'ip_log.user_id', 'vpn_approvals.user_id']);
+// Copies of the person's name kept beside a person column: cleared wherever the row stays.
+const NAME_COLS = ['username', 'display_name', 'user_name', 'sender_name', 'sender_username', 'author_name', 'author_username', 'avatar_url', 'user_avatar'];
+// Person columns the schema declares without a foreign key.
+const EXTRA = [['channel_points', 'streamer_id'], ['arena_mic_moments', 'target_user_id'], ['arena_topics', 'created_by'], ['arena_topics', 'target_user_id']];
+const SECRET_COL = /(^|_)(token|tokens|secret|hash|password|code|key|keys|p256dh|auth|cookie|cookies|credentials)($|_)/i;
+const ROW_LIMIT = 2000;
+const PART_BUDGET = 18 * 1024 * 1024;
+
+function staged() { return new Set(Object.keys(require('../chat/chat-tables').TABLES)); }
+
+function ensureSchema(d) {
+ d.exec(`CREATE TABLE IF NOT EXISTS account_data_events (
+ id TEXT PRIMARY KEY,
+ kind TEXT NOT NULL,
+ subject TEXT NOT NULL,
+ outcome TEXT,
+ sent_at DATETIME,
+ applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
+ )`);
+ const cols = d.prepare('PRAGMA table_info(users)').all().map((c) => c.name);
+ if (!cols.includes('deleted_at')) d.exec('ALTER TABLE users ADD COLUMN deleted_at DATETIME');
+}
+
+const q = (name) => `"${String(name).replace(/"/g, '""')}"`;
+
+/** table → [{ col, action: 'delete'|'null' }], every person column Live has, frozen tables left out. */
+function personColumns(d) {
+ const out = new Map();
+ const add = (t, col, action) => {
+ if (!out.has(t)) out.set(t, []);
+ if (!out.get(t).some((c) => c.col === col)) out.get(t).push({ col, action });
+ };
+ const tables = d.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name").all().map((r) => r.name);
+ for (const t of tables) {
+ if (t === 'users' || FROZEN.has(t)) continue;
+ for (const fk of d.prepare(`PRAGMA foreign_key_list(${q(t)})`).all()) {
+ if (fk.table === 'users') add(t, fk.from, fk.on_delete === 'SET NULL' && !OVERRIDE_DELETE.has(`${t}.${fk.from}`) ? 'null' : 'delete');
+ }
+ if (d.prepare(`PRAGMA table_info(${q(t)})`).all().some((c) => c.name === 'user_id')) add(t, 'user_id', 'delete');
+ }
+ for (const [t, col] of EXTRA) {
+ if (tables.includes(t) && d.prepare(`PRAGMA table_info(${q(t)})`).all().some((c) => c.name === col)) add(t, col, 'delete');
+ }
+ return out;
+}
+
+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;
+};
+
+// ── Export ─────────────────────────────────────────────────────
+
+/** Live's part for one person → network.account-export-part@1 (without the subject). */
+function exportPart(d, userId) {
+ if (!userId) return { files: [], note: 'No Live account.' };
+ const files = [];
+ const truncated = [];
+ let bytes = 0;
+ const user = d.prepare('SELECT * FROM users WHERE id = ?').get(userId) || {};
+ const profile = {};
+ for (const [k, v] of Object.entries(user)) if (!SECRET_COL.test(k)) profile[k] = v;
+ files.push({ name: 'profile.json', content: profile });
+ bytes += JSON.stringify(profile).length;
+ const other = {};
+ for (const [t, cols] of personColumns(d)) {
+ const keep = d.prepare(`PRAGMA table_info(${q(t)})`).all().map((c) => c.name).filter((c) => !SECRET_COL.test(c));
+ if (!keep.length) continue;
+ const where = cols.map((c) => `${q(c.col)} = ?`).join(' OR ');
+ const rows = d.prepare(`SELECT ${keep.map(q).join(', ')} FROM ${q(t)} WHERE ${where} ORDER BY rowid DESC LIMIT ${ROW_LIMIT + 1}`).all(...cols.map(() => userId));
+ if (!rows.length) continue;
+ const content = rows.slice(0, ROW_LIMIT);
+ const size = JSON.stringify(content).length;
+ if (bytes + size > PART_BUDGET) { truncated.push(`${t}.json`); continue; }
+ bytes += size;
+ if (rows.length > ROW_LIMIT) truncated.push(`${t}.json`);
+ // At most 64 files per part: the first 60 tables get their own, the rest share other.json.
+ if (files.length < 60) files.push({ name: `${t}.json`.replace(/[^a-z0-9_.-]/g, '_'), content });
+ else other[t] = content;
+ }
+ if (Object.keys(other).length) files.push({ name: 'other.json', content: other });
+ return { files, truncated, note: 'Recordings and clips are in the media part, pastes and comments in the community part, chat history in the chat part.' };
+}
+
+// ── Deletion ───────────────────────────────────────────────────
+
+/** Erase Live's rows of these Live users in one transaction → { erased, retained }. */
+function eraseUsers(d, userIds, { now = new Date().toISOString() } = {}) {
+ const erased = {};
+ const retained = {};
+ const bump = (o, k, n) => { if (n) o[k] = (o[k] || 0) + n; };
+ const stagedTables = staged();
+ const cols = personColumns(d);
+ const usersCols = d.prepare('PRAGMA table_info(users)').all().map((c) => c.name);
+ const onBilling = (() => { try { return require('../monetization/money-authority').onBilling(); } catch { return false; } })();
+ d.transaction(() => {
+ for (const uid of userIds) {
+ for (const [t, list] of cols) {
+ const where = list.map((c) => `${q(c.col)} = ?`).join(' OR ');
+ if (RETAIN.has(t)) { bump(retained, t, d.prepare(`SELECT COUNT(*) AS n FROM ${q(t)} WHERE ${where}`).get(...list.map(() => uid)).n); continue; }
+ if (stagedTables.has(t)) { bump(retained, 'left_for_chat', d.prepare(`SELECT COUNT(*) AS n FROM ${q(t)} WHERE ${where}`).get(...list.map(() => uid)).n); continue; }
+ for (const c of list) {
+ if (c.action === 'null') {
+ const info = d.prepare(`PRAGMA table_info(${q(t)})`).all().filter((x) => NAME_COLS.includes(x.name));
+ if (info.length) d.prepare(`UPDATE ${q(t)} SET ${info.map((x) => `${q(x.name)} = ${x.notnull ? "'deleted'" : 'NULL'}`).join(', ')} WHERE ${q(c.col)} = ?`).run(uid);
+ }
+ const r = c.action === 'null'
+ ? d.prepare(`UPDATE ${q(t)} SET ${q(c.col)} = NULL WHERE ${q(c.col)} = ?`).run(uid)
+ : d.prepare(`DELETE FROM ${q(t)} WHERE ${q(c.col)} = ?`).run(uid);
+ bump(erased, t, r.changes);
+ }
+ }
+ // A donation's message is the person's words: cleared while Live keeps the ledger (read-only under Billing).
+ if (!onBilling && cols.has('transactions')) {
+ d.prepare('UPDATE transactions SET message = NULL WHERE from_user_id = ? AND message IS NOT NULL').run(uid);
+ }
+ // The tombstone: the kept rows point at nobody; the username is free again.
+ const set = { username: `deleted-${uid}`, email: null, password_hash: '!deleted', display_name: null, avatar_url: null, bio: null, stream_key: null,
+ profile_color: null, theme_id: null, avatar_paste_id: null, ban_reason: null, deleted_at: now };
+ const keys = Object.keys(set).filter((k) => usersCols.includes(k));
+ d.prepare(`UPDATE users SET ${keys.map((k) => `${q(k)} = ?`).join(', ')} WHERE id = ?`).run(...keys.map((k) => set[k]), uid);
+ bump(erased, 'accounts', 1);
+ }
+ })();
+ return { erased, retained };
+}
+
+// ── Events ─────────────────────────────────────────────────────
+
+async function networkCall(path, body) {
+ const principal = require('../net/network-principal');
+ const base = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
+ for (let attempt = 0; attempt < 2; attempt++) {
+ const headers = await principal.serviceHeaders('openvibe.network');
+ const res = await fetch(`${base}${path}`, { method: 'POST', headers: { 'Content-Type': 'application/json', ...headers }, body: JSON.stringify(body), signal: AbortSignal.timeout(30_000) });
+ if (res.status === 401 && attempt === 0) { principal.invalidate('openvibe.network'); continue; }
+ return res;
+ }
+ return null;
+}
+
+/**
+ * Apply one envelope → 'exported' | 'erased' | 'confirmed' | 'closed' | 'ignored:<why>'. Throws when Network could not
+ * be reached (the delivery is answered 500 and redelivered). send(path, body) → Response (tests inject it).
+ */
+async function apply(ev, { send = networkCall } = {}) {
+ const p = ev && ev.payload && typeof ev.payload === 'object' ? ev.payload : {};
+ const d = db.getDb();
+ ensureSchema(d);
+ if (ev.event_type === 'network.account.export_requested') {
+ if (!EXPORT_RE.test(String(p.export_id || '')) || !SUBJECT_RE.test(String(p.subject || ''))) return 'ignored:payload';
+ const seen = d.prepare('SELECT sent_at FROM account_data_events WHERE id = ?').get(p.export_id);
+ if (seen && seen.sent_at) return 'unchanged';
+ const part = exportPart(d, liveUserOf(d, p.subject));
+ const res = await send(`/internal/account-exports/${p.export_id}/parts`, { subject: p.subject, ...part });
+ if (!res) throw new Error('Network unreachable');
+ const outcome = res.ok ? 'exported' : (res.status === 409 || res.status === 404 ? 'closed' : null);
+ if (!outcome) throw new Error(`export part refused: ${res.status}`);
+ d.prepare('INSERT OR REPLACE INTO account_data_events (id, kind, subject, outcome, sent_at) VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)')
+ .run(p.export_id, 'export', p.subject, JSON.stringify({ result: outcome, files: part.files.length }));
+ console.log(`[AccountData] export ${p.export_id}: ${outcome} (${part.files.length} file(s))`);
+ return outcome;
+ }
+ if (ev.event_type === 'network.account.deleted') {
+ if (!DELETION_RE.test(String(p.deletion_id || '')) || !SUBJECT_RE.test(String(p.subject || ''))) return 'ignored:payload';
+ const subjects = [p.subject, ...(Array.isArray(p.aliases) ? p.aliases.filter((s) => SUBJECT_RE.test(String(s))) : [])];
+ let rec = d.prepare('SELECT * FROM account_data_events WHERE id = ?').get(p.deletion_id);
+ let result = 'confirmed';
+ if (!rec) {
+ const ids = [...new Set(subjects.map((s) => liveUserOf(d, s)).filter(Boolean))];
+ for (const id of [...ids]) for (const m of (hasColumn(d, 'users', 'merged_into') ? d.prepare('SELECT id FROM users WHERE merged_into = ?').all(id) : [])) if (!ids.includes(m.id)) ids.push(m.id);
+ const counts = eraseUsers(d, ids);
+ if (d.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'subject_projection'").get()) {
+ d.prepare(`DELETE FROM subject_projection WHERE subject_id IN (${subjects.map(() => '?').join(',')})`).run(...subjects);
+ }
+ d.prepare('INSERT INTO account_data_events (id, kind, subject, outcome) VALUES (?, ?, ?, ?)').run(p.deletion_id, 'deletion', p.subject, JSON.stringify({ users: ids, ...counts }));
+ rec = d.prepare('SELECT * FROM account_data_events WHERE id = ?').get(p.deletion_id);
+ result = 'erased';
+ console.log(`[AccountData] deletion ${p.deletion_id}: ${JSON.stringify({ users: ids, ...counts })}`);
+ }
+ if (rec.sent_at) return 'unchanged';
+ const o = JSON.parse(rec.outcome || '{}');
+ const res = await send(`/internal/account-deletions/${p.deletion_id}/confirmations`, { subject: p.subject, completed_at: new Date(`${String(rec.applied_at).replace(' ', 'T')}Z`).toISOString(), erased: o.erased || {}, retained: o.retained || {} });
+ if (!res) throw new Error('Network unreachable');
+ if (!res.ok && res.status !== 404) throw new Error(`confirmation refused: ${res.status}`);
+ d.prepare('UPDATE account_data_events SET sent_at = CURRENT_TIMESTAMP WHERE id = ?').run(p.deletion_id);
+ return result;
+ }
+ return 'ignored:type';
+}
+
+function hasColumn(d, table, col) { return d.prepare(`PRAGMA table_info(${q(table)})`).all().some((c) => c.name === col); }
+
+module.exports = { apply, exportPart, eraseUsers, personColumns, ensureSchema, FROZEN, RETAIN };
@@ -12,6 +12,8 @@
@@ -12,6 +12,8 @@
* 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
* 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
* move to the survivor's (./subject-merge.js), once per merge
+ * account.export_requested / account.deleted the person's export part goes to Network, or their Live rows are
+ * erased and the erasure confirmed (./account-data.js, ADR-033), once per export or deletion
*
*
* 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).
@@ -25,10 +27,11 @@ const notifyChat = (userId) => { try { const cs = require('../chat/chat-server')
@@ -25,10 +27,11 @@ 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>' (a merge: a promise of 'merged' | 'relinked' | 'nothing' | …). */
+/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>' (a merge, export or deletion: a promise). */
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';
if (ev.event_type === 'network.subject.merged') return ev.source === 'network' ? require('./subject-merge').apply(ev) : 'ignored:source';
+ if (ev.event_type === 'network.account.export_requested' || ev.event_type === 'network.account.deleted') return ev.source === 'network' ? require('./account-data').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';
@@ -50,11 +53,11 @@ function handler(req, res) {
@@ -50,11 +53,11 @@ 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 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' || out === 'exported' || out === 'erased' || out === 'confirmed') stats.profiles++; else stats.ignored++; };
const out = apply(delivery.event);
const out = apply(delivery.event);
if (out && typeof out.then === 'function') {
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' }); });
+ // A merge, export or deletion: answer after it applied, so a failure is redelivered.
+ return out.then((o) => { count(o); res.status(204).end(); }, (e) => { console.error(`[NetworkEvents] ${delivery.event.event_type} failed:`, e.message); res.status(500).json({ error: 'not applied' }); });
}
}
count(out);
count(out);
res.status(204).end();
res.status(204).end();
new file mode 100644
new file mode 100644
@@ -0,0 +1,126 @@
@@ -0,0 +1,126 @@
+'use strict';
+// Account export and deletion → Live (roadmap WS-B task 7, ADR-033): through POST /internal/network-events (signed),
+// network.account.export_requested sends Live's part to Network (the profile and the person's rows per table, no
+// stream keys or token hashes; an empty part for someone with no Live account), and network.account.deleted erases the
+// person's rows (follows and channel points both ways, streams with what is inside them, their own chat lines,
+// managed streams, tokens), keeps money and moderation rows pointing at a tombstone (the donation message cleared),
+// keeps shared rows without them, releases the username, and confirms with counts. A redelivery sends nothing twice;
+// a failed confirmation is retried without erasing again. The subscription asks for both topics.
+// node test/account-data.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-account-data-'));
+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 accountData = require('../server/auth/account-data');
+const { NETWORK_TOPICS } = require('../scripts/subscribe-media-events');
+
+const DANA = ids.newId('user'), XENA = ids.newId('user'), NOBODY = ids.newId('user');
+const d = db.getDb();
+d.prepare(`INSERT INTO users (id, username, email, password_hash, bio) VALUES (20, 'dana', '[email protected]', '$sso$', 'hi'), (21, 'xena', NULL, '$sso$', ''), (22, 'yuri', NULL, '$sso$', '')`).run();
+d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, service_username, subject_id) VALUES (20, 'network', '200', 'dana', ?), (21, 'network', '201', 'xena', ?)").run(DANA, XENA);
+d.prepare('INSERT INTO follows (follower_id, streamer_id) VALUES (20, 21), (22, 20), (22, 21)').run();
+d.prepare('INSERT INTO channel_points (user_id, streamer_id, balance) VALUES (20, 21, 30), (22, 20, 7), (22, 21, 5)').run();
+d.prepare("INSERT INTO streams (id, user_id, title) VALUES (1, 20, 'dana live'), (2, 21, 'xena live')").run();
+// yuri chats in dana's stream (goes with the stream); dana chats in xena's (her own line goes); yuri in xena's stays.
+d.prepare("INSERT INTO chat_messages (stream_id, user_id, username, message) VALUES (1, 22, 'yuri', 'hey dana'), (2, 20, 'dana', 'hey xena'), (2, 22, 'yuri', 'yo')").run();
+d.prepare("INSERT INTO managed_streams (user_id, stream_key) VALUES (20, 'sk_live_secret_dana')").run();
+d.prepare("INSERT INTO api_tokens (user_id, token_hash) VALUES (20, 'hash-secret-dana')").run();
+d.prepare("INSERT INTO transactions (from_user_id, to_user_id, amount, type, message) VALUES (20, 21, 5, 'donation', 'love you xena'), (22, 20, 3, 'donation', 'for dana')").run();
+d.prepare("INSERT INTO payment_orders (user_id, provider, provider_ref, kind, amount_cents, currency, bucks, status) VALUES (20, 'paypal', 'PO-1', 'bucks', 500, 'USD', 5, 'completed')").run();
+d.prepare("INSERT INTO moderation_actions (action_type, actor_user_id, target_user_id) VALUES ('timeout', 21, 20)").run();
+
+const envelope = (type, payload) => {
+ assert.ok(validate(`${type}@1`, payload).valid, JSON.stringify(validate(`${type}@1`, payload).errors));
+ return { event_id: ids.newId('event'), event_type: type, version: 1, source: 'network', actor: { type: 'user', id: payload.subject },
+ timestamp: new Date().toISOString(), visibility: 'internal', subject: { type: 'user', id: payload.subject }, payload };
+};
+const ulid = () => ids.ulid();
+
+(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 deliver = 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; };
+ const sent = [];
+ let failNext = false;
+ const send = async (p, body) => { if (failNext) { failNext = false; return { ok: false, status: 503 }; } sent.push({ path: p, body }); return { ok: true, status: 200 }; };
+ try {
+ assert.ok(NETWORK_TOPICS.includes('network.account.export_requested') && NETWORK_TOPICS.includes('network.account.deleted'), 'the subscription asks for both');
+ const cols = accountData.personColumns(d);
+ for (const t of accountData.FROZEN) assert.ok(!cols.has(t), `${t} is never touched`);
+
+ // ── Export ──
+ const exp = envelope('network.account.export_requested', { export_id: `exp_${ulid()}`, subject: DANA, requested_at: new Date().toISOString(), deadline: new Date(Date.now() + 1800000).toISOString() });
+ assert.strictEqual(await accountData.apply(exp, { send }), 'exported');
+ assert.strictEqual(sent.length, 1);
+ assert.strictEqual(sent[0].path, `/internal/account-exports/${exp.payload.export_id}/parts`);
+ const part = sent[0].body;
+ assert.ok(validate('network.account-export-part@1', part).valid, JSON.stringify(validate('network.account-export-part@1', part).errors));
+ const byName = Object.fromEntries(part.files.map((f) => [f.name, f.content]));
+ assert.strictEqual(byName['profile.json'].username, 'dana');
+ assert.deepStrictEqual(byName['follows.json'].map((f) => [f.follower_id, f.streamer_id]).sort(), [[20, 21], [22, 20]]);
+ assert.ok(byName['transactions.json'] && byName['payment_orders.json'], 'money history is theirs to see');
+ const text = JSON.stringify(part);
+ for (const secret of ['sk_live_secret_dana', 'hash-secret-dana', '$sso$']) assert.ok(!text.includes(secret), `no ${secret}`);
+ assert.strictEqual(await accountData.apply(exp, { send }), 'unchanged', 'a redelivery sends nothing twice');
+ assert.strictEqual(sent.length, 1);
+ const lone = envelope('network.account.export_requested', { export_id: `exp_${ulid()}`, subject: NOBODY, requested_at: new Date().toISOString(), deadline: new Date(Date.now() + 1800000).toISOString() });
+ assert.strictEqual(await accountData.apply(lone, { send }), 'exported');
+ assert.deepStrictEqual(sent[1].body.files, [], 'no Live account: an empty part, so the export does not wait');
+
+ // ── Deletion ──
+ const del = envelope('network.account.deleted', { deletion_id: `del_${ulid()}`, subject: DANA, requested_at: new Date(Date.now() - 30 * 86400000).toISOString(), deleted_at: new Date().toISOString() });
+ failNext = true;
+ await assert.rejects(accountData.apply(del, { send }), /confirmation refused: 503/, 'a failed confirmation is redelivered');
+ const n = (sql, ...a) => d.prepare(sql).get(...a).n;
+ assert.strictEqual(n('SELECT COUNT(*) AS n FROM follows WHERE follower_id = 20 OR streamer_id = 20'), 0, 'follows both ways');
+ assert.strictEqual(n('SELECT COUNT(*) AS n FROM channel_points WHERE user_id = 20 OR streamer_id = 20'), 0, 'channel points both ways');
+ assert.strictEqual(n('SELECT COUNT(*) AS n FROM follows'), 1, "others' follows stay");
+ assert.strictEqual(n('SELECT COUNT(*) AS n FROM streams WHERE user_id = 20'), 0);
+ assert.deepStrictEqual(d.prepare('SELECT stream_id, user_id, message FROM chat_messages ORDER BY id').all(), [{ stream_id: 2, user_id: 22, message: 'yo' }],
+ 'the stream took its chat with it; her own line elsewhere went; others stay');
+ assert.strictEqual(n('SELECT COUNT(*) AS n FROM managed_streams WHERE user_id = 20') + n('SELECT COUNT(*) AS n FROM api_tokens WHERE user_id = 20') + n("SELECT COUNT(*) AS n FROM linked_accounts WHERE user_id = 20"), 0);
+ assert.deepStrictEqual(d.prepare('SELECT from_user_id AS f, message FROM transactions ORDER BY id').all(), [{ f: 20, message: null }, { f: 22, message: 'for dana' }], 'money kept; her own message cleared');
+ assert.strictEqual(n("SELECT COUNT(*) AS n FROM payment_orders WHERE user_id = 20"), 1, 'payment orders kept');
+ assert.strictEqual(n('SELECT COUNT(*) AS n FROM moderation_actions WHERE target_user_id = 20'), 1, 'moderation records kept');
+ const tomb = d.prepare('SELECT username, email, bio, deleted_at FROM users WHERE id = 20').get();
+ assert.deepStrictEqual([tomb.username, tomb.email, tomb.bio, !!tomb.deleted_at], ['deleted-20', null, null, true]);
+ d.prepare("INSERT INTO users (username, password_hash) VALUES ('dana', '$sso$')").run();
+ assert.strictEqual(await accountData.apply(del, { send }), 'confirmed', 'the retry confirms without erasing again');
+ const conf = sent[2];
+ assert.strictEqual(conf.path, `/internal/account-deletions/${del.payload.deletion_id}/confirmations`);
+ assert.ok(validate('network.account-deletion-confirmation@1', conf.body).valid, JSON.stringify(validate('network.account-deletion-confirmation@1', conf.body).errors));
+ assert.ok(conf.body.erased.follows >= 2 && conf.body.erased.streams === 1 && conf.body.erased.accounts === 1, JSON.stringify(conf.body.erased));
+ assert.ok(conf.body.retained.transactions === 2 && conf.body.retained.payment_orders === 1 && conf.body.retained.moderation_actions === 1, JSON.stringify(conf.body.retained));
+ assert.strictEqual(await accountData.apply(del, { send }), 'unchanged');
+ assert.strictEqual(sent.length, 3);
+
+ // Through the signed endpoint: another source is ignored (204, nothing sent), and a bad payload too.
+ assert.strictEqual(await deliver({ ...envelope('network.account.deleted', { deletion_id: `del_${ulid()}`, subject: XENA, requested_at: new Date().toISOString(), deleted_at: new Date().toISOString() }), source: 'media' }), 204);
+ assert.strictEqual(n('SELECT COUNT(*) AS n FROM users WHERE id = 21 AND deleted_at IS NULL'), 1, 'xena untouched');
+ assert.strictEqual(await networkEvents.apply({ event_id: ids.newId('event'), event_type: 'network.account.deleted', source: 'network', payload: { deletion_id: 'x', subject: XENA } }), 'ignored:payload');
+ } finally {
+ server.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ }
+ console.log('account export and deletion: all checks passed');
+})().catch((e) => { console.error(e); process.exit(1); });
@@ -353,6 +353,7 @@ async function check(name, fn) {
@@ -353,6 +353,7 @@ async function check(name, fn) {
'server/streaming/golive-notify.js': 'OpenVibe.Network (configured)',
'server/streaming/golive-notify.js': 'OpenVibe.Network (configured)',
'server/utils/notify.js': 'OpenVibe.Network (configured)',
'server/utils/notify.js': 'OpenVibe.Network (configured)',
'server/auth/identity-sync.js': 'OpenVibe.Network (configured)',
'server/auth/identity-sync.js': 'OpenVibe.Network (configured)',
+ 'server/auth/account-data.js': 'OpenVibe.Network (configured): export parts and deletion confirmations',
'server/monetization/billing-client.js': 'OpenVibe.Billing (configured)',
'server/monetization/billing-client.js': 'OpenVibe.Billing (configured)',
'server/monetization/wallet-client.js': 'OpenVibe.Network wallet (configured)',
'server/monetization/wallet-client.js': 'OpenVibe.Network wallet (configured)',
'server/monetization/payments.js': 'PayPal (fixed host)',
'server/monetization/payments.js': 'PayPal (fixed host)',