openvibers/openvibe.chat Commits
Chat takes part in account export and deletion (roadmap WS-B task 7, ADR-033; openvibe-contracts 0.57.0 -> 0.71.0). POST /internal/events takes network.account.export_requested and network.account.deleted (subscribed at boot). server/chat/account-data.js applies each once per export or deletion (account_data_events) and answers after Network took the part or confirmation, so a failure is redelivered without erasing twice.
Changes
@@@ -16,7 +16,7 @@"express": "^4.21.2","express-rate-limit": "^7.5.0","multer": "^2.0.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.57.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.22.0","ws": "^8.18.0"@@@ -1062,9 +1062,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.57.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.57.0",-- "integrity": "sha512-6hXpTo0y1h7YFQgyxkkWIoSx4CqSPDlGAK617MV2SkaO01zVOWKSQ1wor7aFOvrv+7u7a3KsIak3sBznNc39Vw==",++ "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",@@@ -28,7 +28,7 @@"express": "^4.21.2","express-rate-limit": "^7.5.0","multer": "^2.0.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.57.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.22.0","ws": "^8.18.0"nnew file mode 100644@@@ -0,0 +1,212 @@++'use strict';++/**++ * Account export and deletion → Chat (roadmap WS-B task 7, ADR-033; Contracts 0.71.0). Both arrive at POST++ * /internal/events and are applied once per export or deletion (account_data_events); the delivery is answered after++ * Network took the part or the confirmation, so a failure is redelivered without erasing twice. Chat keys people by++ * subject and by Live user id (ctx_users maps the two).++ *++ * network.account.export_requested Chat's part (POST /internal/account-exports/:id/parts with a service token):++ * their chat, room and direct messages, conversations and rooms, calls, blocks,++ * emotes and sounds, and the moderation actions taken on them.++ * network.account.deleted what the subject (and the accounts merged into it) wrote or set up goes:++ * - chat, room and direct messages they sent, their DM memberships (a++ * conversation left with nobody goes), room memberships and attachments;++ * - rooms they own go when nobody else is in them, and otherwise stay without an++ * owner; calls they made or took, and blocks both ways;++ * - their channel's emotes, sounds, moderators, first-chat stats and audio++ * requests, their tags, and the AI summaries and timeline about them;++ * - the channel, stream and user mirrors of their Live account (the user mirror++ * stays as a tombstone).++ * Moderation actions stay as recorded. A staged table (channel_moderators,++ * channel_moderation_settings, emotes, user_tags, chat_ai_summaries,++ * chat_timeline_events) is erased here only while Chat writes it (table_authority);++ * until then Live erases it and its capture keeps this copy current.++ * Chat then confirms with counts.++ */++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 TOPICS = ['network.account.export_requested', 'network.account.deleted'];++const ROW_LIMIT = 5000;++++const cols = (d, t) => { try { return d.prepare(`PRAGMA table_info(${t})`).all().map((c) => c.name); } catch { return []; } };++const has = (d, t, c) => cols(d, t).includes(c);++const inList = (xs) => `(${xs.map(() => '?').join(',')})`;++++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 TEXT,++ applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))++ )`);++}++++/** The Live user ids of these subjects (Chat's ctx_users mirror). */++function liveIds(d, subjects) {++ return d.prepare(`SELECT id FROM ctx_users WHERE subject_id IN ${inList(subjects)}`).all(...subjects).map((r) => r.id);++}++++/** WHERE for a table: its subject column(s) IN subjects, or its Live user column(s) IN ids. → [sql, params] | null. */++function whereFor(d, table, subjectCols, idCols, subjects, ids) {++ const parts = []; const params = [];++ for (const c of subjectCols) if (has(d, table, c)) { parts.push(`${c} IN ${inList(subjects)}`); params.push(...subjects); }++ if (ids.length) for (const c of idCols) if (has(d, table, c)) { parts.push(`${c} IN ${inList(ids)}`); params.push(...ids); }++ return parts.length ? [parts.join(' OR '), params] : null;++}++++// ── Export ─────────────────────────────────────────────────────++++const EXPORTS = [++ ['chat_messages.json', 'chat_messages', ['subject_id'], ['user_id']],++ ['room_messages.json', 'room_messages', ['subject_id'], ['user_id']],++ ['direct_messages.json', 'dm_messages', ['sender_subject_id'], ['sender_id']],++ ['conversations.json', 'dm_participants', ['subject_id'], ['user_id']],++ ['rooms_owned.json', 'rooms', ['owner_subject'], ['owner_id']],++ ['calls.json', 'calls', ['created_by_subject', 'target_subject'], ['created_by', 'target_user_id']],++ ['blocks.json', 'network_blocks', ['blocker_subject'], []],++ ['dm_blocks.json', 'dm_blocks', ['blocker_subject_id'], ['blocker_id']],++ ['emotes.json', 'emotes', [], ['user_id']],++ ['sounds.json', 'channel_sounds', ['created_by_subject_id'], ['created_by']],++ ['moderation_on_you.json', 'moderation_actions', [], ['target_user_id']],++];++++function exportPart(d, subject) {++ const ids = liveIds(d, [subject]);++ const files = []; const truncated = [];++ for (const [name, table, sc, ic] of EXPORTS) {++ const w = whereFor(d, table, sc, ic, [subject], ids);++ if (!w) continue;++ let rows = d.prepare(`SELECT * FROM ${table} WHERE ${w[0]} ORDER BY rowid DESC LIMIT ${ROW_LIMIT + 1}`).all(...w[1]);++ if (table === 'moderation_actions') rows = rows.map(({ actor_user_id, actor_subject_id, ...r }) => r); // who acted is staff's, not theirs++ if (!rows.length) continue;++ if (rows.length > ROW_LIMIT) truncated.push(name);++ files.push({ name, content: rows.slice(0, ROW_LIMIT) });++ }++ return { files, truncated };++}++++// ── Deletion ───────────────────────────────────────────────────++++const STAGED = ['channel_moderators', 'channel_moderation_settings', 'emotes', 'user_tags', 'chat_ai_summaries', 'chat_timeline_events'];++const chatWrites = (d, table) => { try { const r = d.prepare('SELECT authority FROM table_authority WHERE table_name = ?').get(table); return !!r && r.authority === 'chat'; } catch { return false; } };++++function erase(d, subjects) {++ const ids = liveIds(d, subjects);++ const erased = {}; const retained = {};++ const add = (o, k, n) => { if (n) o[k] = (o[k] || 0) + n; };++ const del = (key, table, sc, ic) => {++ const w = whereFor(d, table, sc, ic, subjects, ids);++ if (!w) return;++ // A staged table Live still writes: Live erases it, and its capture brings the change here.++ if (STAGED.includes(table) && !chatWrites(d, table)) { add(retained, 'erased_by_live', d.prepare(`SELECT COUNT(*) AS n FROM ${table} WHERE ${w[0]}`).get(...w[1]).n); return; }++ add(erased, key, d.prepare(`DELETE FROM ${table} WHERE ${w[0]}`).run(...w[1]).changes);++ };++ d.transaction(() => {++ del('messages', 'chat_messages', ['subject_id'], ['user_id']);++ del('room_messages', 'room_messages', ['subject_id'], ['user_id']);++ del('direct_messages', 'dm_messages', ['sender_subject_id'], ['sender_id']);++ del('conversations_left', 'dm_participants', ['subject_id'], ['user_id']);++ if (has(d, 'dm_conversations', 'id')) add(erased, 'conversations', d.prepare('DELETE FROM dm_conversations WHERE id NOT IN (SELECT conversation_id FROM dm_participants)').run().changes);++ del('room_memberships', 'room_members', [], ['user_id']);++ del('room_attachments', 'room_attachments', ['attached_by_subject'], []);++ const owned = whereFor(d, 'rooms', ['owner_subject'], ['owner_id'], subjects, ids);++ if (owned) {++ for (const r of d.prepare(`SELECT id FROM rooms WHERE ${owned[0]}`).all(...owned[1])) {++ const others = has(d, 'room_members', 'room_id') && d.prepare('SELECT 1 FROM room_members WHERE room_id = ? LIMIT 1').get(r.id);++ if (others) { d.prepare('UPDATE rooms SET owner_subject = NULL WHERE id = ?').run(r.id); add(retained, 'rooms_without_owner', 1); }++ else {++ if (has(d, 'room_messages', 'room_id')) d.prepare('DELETE FROM room_messages WHERE room_id = ?').run(r.id);++ d.prepare('DELETE FROM rooms WHERE id = ?').run(r.id);++ add(erased, 'rooms', 1);++ }++ }++ }++ del('calls', 'calls', ['created_by_subject', 'target_subject'], ['created_by', 'target_user_id']);++ del('blocks', 'network_blocks', ['blocker_subject', 'blocked_subject'], []);++ del('blocks', 'dm_blocks', ['blocker_subject_id'], ['blocker_id', 'blocked_id']);++ del('ai_summaries', 'chat_ai_summaries', ['subject_id'], []);++ del('timeline', 'chat_timeline_events', ['subject_id'], []);++ del('tags', 'user_tags', [], ['user_id']);++ del('moderator_roles', 'channel_moderators', [], ['user_id']);++ del('pending_messages', 'pending_ip_messages', [], ['user_id']);++ if (ids.length) {++ // Their channel: emotes, sounds, moderators, stats, audio requests, then the Live mirrors.++ const channels = has(d, 'ctx_channels', 'user_id') ? d.prepare(`SELECT id FROM ctx_channels WHERE user_id IN ${inList(ids)}`).all(...ids).map((r) => r.id) : [];++ if (channels.length && has(d, 'channel_moderators', 'channel_id') && chatWrites(d, 'channel_moderators')) add(erased, 'moderator_roles', d.prepare(`DELETE FROM channel_moderators WHERE channel_id IN ${inList(channels)}`).run(...channels).changes);++ if (channels.length && has(d, 'channel_moderation_settings', 'channel_id') && chatWrites(d, 'channel_moderation_settings')) d.prepare(`DELETE FROM channel_moderation_settings WHERE channel_id IN ${inList(channels)}`).run(...channels);++ del('emotes', 'emotes', [], ['user_id']);++ del('sounds', 'channel_sounds', ['created_by_subject_id'], ['created_by']);++ del('channel_stats', 'stream_first_chats', [], ['channel_user_id']);++ del('audio_requests', 'audio_requests', [], ['channel_user_id']);++ for (const t of ['ctx_channels', 'ctx_streams', 'ctx_managed_streams']) del('live_mirror', t, [], ['user_id']);++ const set = ['username', 'display_name', 'avatar_url', 'profile_color', 'ban_reason'].filter((c) => has(d, 'ctx_users', c));++ for (const id of ids) {++ d.prepare(`UPDATE ctx_users SET ${set.map((c) => `${c} = ${c === 'username' ? '?' : 'NULL'}`).join(', ')} WHERE id = ?`).run(`deleted-${id}`, id);++ }++ }++ const mod = whereFor(d, 'moderation_actions', ['actor_subject_id'], ['target_user_id', 'actor_user_id'], subjects, ids);++ if (mod) add(retained, 'moderation_actions', d.prepare(`SELECT COUNT(*) AS n FROM moderation_actions WHERE ${mod[0]}`).get(...mod[1]).n);++ })();++ return { erased, retained, ids };++}++++// ── Events ─────────────────────────────────────────────────────++++async function networkCall(path, body) {++ const auth = require('../net/service-auth');++ const config = require('../config');++ for (let attempt = 0; attempt < 2; attempt++) {++ const res = await fetch(`${config.networkInternalUrl}${path}`, { method: 'POST', headers: { 'Content-Type': 'application/json', ...(await auth.headers('openvibe.network')) }, body: JSON.stringify(body), signal: AbortSignal.timeout(30_000) });++ if (res.status === 401 && attempt === 0) { auth.invalidate('openvibe.network'); continue; }++ return res;++ }++ return null;++}++++/** One envelope → 'exported' | 'erased' | 'confirmed' | 'closed' | 'unchanged' | 'ignored:<why>'; throws to be redelivered. */++async function apply(ev, { send = networkCall, log = console, onErased = null } = {}) {++ if (!ev || !TOPICS.includes(ev.event_type)) return 'ignored:type';++ if (ev.source !== 'network') return 'ignored:source';++ const p = 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, 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 (?, ?, ?, ?, ?)')++ .run(p.export_id, 'export', p.subject, JSON.stringify({ result: outcome, files: part.files.length }), new Date().toISOString());++ return outcome;++ }++ if (!DELETION_RE.test(String(p.deletion_id || '')) || !SUBJECT_RE.test(String(p.subject || ''))) return 'ignored:payload';++ let rec = d.prepare('SELECT * FROM account_data_events WHERE id = ?').get(p.deletion_id);++ let result = 'confirmed';++ if (!rec) {++ const subjects = [p.subject, ...(Array.isArray(p.aliases) ? p.aliases.filter((s) => SUBJECT_RE.test(String(s))) : [])];++ const { ids, ...counts } = erase(d, subjects);++ d.prepare('INSERT INTO account_data_events (id, kind, subject, outcome) VALUES (?, ?, ?, ?)').run(p.deletion_id, 'deletion', p.subject, JSON.stringify(counts));++ log.log(`[AccountData] deletion ${p.deletion_id}: ${JSON.stringify(counts)}`);++ if (onErased) { try { onErased(subjects, ids); } catch { /* caches refill anyway */ } }++ rec = d.prepare('SELECT * FROM account_data_events WHERE id = ?').get(p.deletion_id);++ result = 'erased';++ }++ 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: rec.applied_at, 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 = ? WHERE id = ?').run(new Date().toISOString(), p.deletion_id);++ return result;++}++++module.exports = { apply, exportPart, erase, ensureSchema, TOPICS };@@@ -47,7 +47,7 @@ const ctx = require('../live-context');const CONSUMER = 'chat';const INBOX_TABLE = 'chat_event_inbox';--const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated', 'network.user.token_valid_after', 'vip.membership.changed', 'network.block.changed', 'network.subject.merged']);++const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated', 'network.user.token_valid_after', 'vip.membership.changed', 'network.block.changed', 'network.subject.merged', 'network.account.export_requested', 'network.account.deleted']);const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;const INBOX_KEEP_MS = 35 * 24 * 3600 * 1000; // Events keeps events 30 days: nothing older can be redelivered@@@ -166,6 +166,21 @@ function createEventsConsumer({ chatServer, secrets = [], now = () => Date.now()stats.received++;stats.last_at = new Date(now()).toISOString();stats.last_type = event.event_type;++ if (event.event_type === 'network.account.export_requested' || event.event_type === 'network.account.deleted') {++ // Account export and deletion (ADR-033, ../chat/account-data.js): once per export or deletion by its own record,++ // answered after Network took the part or confirmation, so a failure is redelivered without erasing twice.++ const onErased = (subjects, ids) => { for (const id of ids) ctx.invalidateUser(id); };++ return require('../chat/account-data').apply(event, { log, onErased }).then((outcome) => {++ if (String(outcome).startsWith('ignored:')) stats.ignored++; else stats.applied++;++ stats.last_outcome = outcome;++ res.status(200).json({ event_id: event.event_id, duplicate: outcome === 'unchanged', outcome });++ }, (err) => {++ stats.failed++;++ stats.last_error = String(err.message || err).slice(0, 200);++ log.error(`[Events consumer] ${event.event_id} (${event.event_type}) failed:`, err.message);++ problem(500, 'chat.event_failed', 'processing failed; it will be retried');++ });++ }let out;try {out = apply(event);nnew file mode 100644@@@ -0,0 +1,80 @@++'use strict';++/**++ * Account export and deletion in Chat (roadmap WS-B task 7, ADR-033):++ * - network.account.export_requested sends Chat's part to Network: their messages, DMs they sent, conversations,++ * blocks, and moderation taken on them without who acted.++ * - network.account.deleted erases what the subject and its merged-in aliases wrote or set up: chat, room and DM++ * messages, DM memberships (a conversation left empty goes), blocks both ways, the channel's first-chat stats, and++ * the Live mirrors (the user mirror as a tombstone). Moderation history stays. A staged table is left to Live++ * while Live writes it.++ * Chat confirms with counts; a failed confirmation is retried without erasing again, and a redelivery sends nothing++ * twice. Chat subscribes to both.++ */++const assert = require('assert');++const { ids, validate } = require('openvibe-contracts');++const { boot, suite } = require('./helpers');++const t = suite('account-data');++let h, dana, otto, old;++const ev = (type, payload) => ({ 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 });++++t('boot', async () => {++ h = await boot();++ dana = h.addUser('dana', { subject: ids.newId('user') });++ old = h.addUser('dana_old', { subject: ids.newId('user') });++ otto = h.addUser('otto', { subject: ids.newId('user') });++ await h.ctx.sync();++ assert.ok(require('../server/events/consumer').TOPICS.includes('network.account.deleted'), 'Chat subscribes');++});++++t('export, then erase and confirm', async () => {++ const accountData = require('../server/chat/account-data');++ const d = require('../server/db/database').getDb();++ d.prepare("INSERT INTO chat_messages (user_id, username, message, channel_user_id, subject_id) VALUES (?, 'dana', 'hi otto', ?, ?), (?, 'otto', 'hi dana', ?, ?), (?, 'dana_old', 'old me', ?, ?)")++ .run(dana.id, otto.id, dana.subject_id, otto.id, dana.id, otto.subject_id, old.id, otto.id, old.subject_id);++ const pair = (await h.http('POST', '/api/dm/conversations', { token: dana.token, body: { user_ids: [otto.id] } })).body.conversation.id;++ assert.ok((await h.http('POST', `/api/dm/conversations/${pair}/messages`, { token: dana.token, body: { message: 'psst' } })).status < 300);++ assert.ok((await h.http('POST', `/api/dm/conversations/${pair}/messages`, { token: otto.token, body: { message: 'what' } })).status < 300);++ require('../server/chat/network-blocks').ensureSchema && require('../server/chat/network-blocks').ensureSchema();++ d.prepare('INSERT INTO network_blocks (blocker_subject, blocked_subject, active, revision, updated_at) VALUES (?, ?, 1, 1, 0), (?, ?, 1, 1, 0)').run(dana.subject_id, otto.subject_id, otto.subject_id, dana.subject_id);++ d.prepare("INSERT INTO moderation_actions (action_type, actor_user_id, target_user_id, actor_subject_id) VALUES ('timeout', ?, ?, ?)").run(otto.id, dana.id, otto.subject_id);++ d.prepare("INSERT INTO user_tags (user_id, tag_id) VALUES (?, 'veteran')").run(dana.id);++++ 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 }; };++++ // ── Export ──++ const exp = ev('network.account.export_requested', { export_id: `exp_${ids.ulid()}`, subject: dana.subject_id, requested_at: new Date().toISOString(), deadline: new Date(Date.now() + 1800000).toISOString() });++ assert.strictEqual(await accountData.apply({ ...exp, source: 'live' }, { send }), 'ignored:source');++ assert.strictEqual(await accountData.apply(exp, { send }), 'exported');++ 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 files = Object.fromEntries(part.files.map((f) => [f.name, f.content]));++ assert.deepStrictEqual(files['chat_messages.json'].map((m) => m.message), ['hi otto']);++ assert.deepStrictEqual(files['direct_messages.json'].map((m) => m.message), ['psst']);++ assert.ok(files['moderation_on_you.json'][0] && !('actor_user_id' in files['moderation_on_you.json'][0]), "moderation on them, without who acted");++ assert.strictEqual(await accountData.apply(exp, { send }), 'unchanged');++ assert.strictEqual(sent.length, 1);++++ // ── Deletion ──++ const del = ev('network.account.deleted', { deletion_id: `del_${ids.ulid()}`, subject: dana.subject_id, aliases: [old.subject_id], requested_at: new Date().toISOString(), deleted_at: new Date().toISOString() });++ failNext = true;++ await assert.rejects(accountData.apply(del, { send }), /confirmation refused: 503/);++ const q = (sql, ...a) => d.prepare(sql).all(...a);++ assert.deepStrictEqual(q('SELECT message FROM chat_messages ORDER BY id').map((m) => m.message), ['hi dana'], "hers and the alias's gone; otto's stays");++ assert.deepStrictEqual(q('SELECT message FROM dm_messages ORDER BY id').map((m) => m.message), ['what'], 'the DM she sent goes, his reply stays');++ assert.deepStrictEqual(q('SELECT user_id FROM dm_participants WHERE conversation_id = ?', pair).map((p) => p.user_id), [otto.id]);++ assert.strictEqual(q('SELECT * FROM network_blocks').length, 0, 'blocks both ways');++ assert.strictEqual(q('SELECT * FROM moderation_actions').length, 1, 'moderation history stays');++ assert.strictEqual(q('SELECT * FROM user_tags').length, 1, 'a staged table Live still writes is left to Live');++ assert.strictEqual(d.prepare('SELECT username FROM ctx_users WHERE id = ?').get(dana.id).username, `deleted-${dana.id}`);++ assert.strictEqual(await accountData.apply(del, { send }), 'confirmed', 'the retry confirms without erasing again');++ const conf = sent[1];++ assert.ok(validate('network.account-deletion-confirmation@1', conf.body).valid, JSON.stringify(validate('network.account-deletion-confirmation@1', conf.body).errors));++ assert.deepStrictEqual([conf.body.erased.messages, conf.body.erased.direct_messages, conf.body.retained.moderation_actions, conf.body.retained.erased_by_live], [2, 1, 1, 1]);++ assert.strictEqual(await accountData.apply(del, { send }), 'unchanged');++ assert.strictEqual(sent.length, 2);++});++++t.run(async () => { if (h && h.close) await h.close(); });@@@ -123,8 +123,8 @@ t('boot with the consumer on and a stub Events', async () => {t('boot created one subscription per topic, to /internal/events, with the consumer’s secret', async () => {const results = await h.subscriptions.done;-- assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created'], ['network.user.token_valid_after', 'created'], ['vip.membership.changed', 'created'], ['network.block.changed', 'created'], ['network.subject.merged', 'created']]);-- assert.strictEqual(stubEvents.subs.length, 6);++ assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created'], ['network.user.token_valid_after', 'created'], ['vip.membership.changed', 'created'], ['network.block.changed', 'created'], ['network.subject.merged', 'created'], ['network.account.export_requested', 'created'], ['network.account.deleted', 'created']]);++ assert.strictEqual(stubEvents.subs.length, 8);for (const s of stubEvents.subs) {assert.strictEqual(s.endpoint, `http://127.0.0.1:${h.port}/internal/events`);assert.strictEqual(s.secret, SECRET, 'the first CHAT_EVENTS_SECRET is handed to Events');@@@ -139,17 +139,17 @@ t('a second boot changes nothing, and never re-enables a subscription an operatoconst endpoint = subs.endpointFor(config, h.port);const posts = stubEvents.posts;let r = await subs.ensure({ config, endpoint });-- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists', 'exists', 'exists']);++ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists', 'exists', 'exists', 'exists', 'exists']);assert.strictEqual(stubEvents.posts, posts, 'nothing was created again');r = await subs.ensure({ config, endpoint, action: 'disable', topics: ['live.release.deployed'] });assert.deepStrictEqual(r.map((x) => [x.result, x.enabled]), [['disabled', false]]);r = await subs.ensure({ config, endpoint });-- assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true], ['network.user.token_valid_after', 'exists', true], ['vip.membership.changed', 'exists', true], ['network.block.changed', 'exists', true], ['network.subject.merged', 'exists', true]]);++ assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true], ['network.user.token_valid_after', 'exists', true], ['vip.membership.changed', 'exists', true], ['network.block.changed', 'exists', true], ['network.subject.merged', 'exists', true], ['network.account.export_requested', 'exists', true], ['network.account.deleted', 'exists', true]]);assert.strictEqual(stubEvents.subs[0].enabled, false, 'still disabled');r = await subs.ensure({ config, endpoint, action: 'enable', topics: ['live.release.deployed'] });assert.strictEqual(stubEvents.subs[0].enabled, true);r = await subs.ensure({ config, endpoint, action: 'list' });-- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists', 'exists', 'exists']);++ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists', 'exists', 'exists', 'exists', 'exists']);});t('signature v2 only: bad signature, stale or future timestamp, v1-only, unsigned → 401', async () => {@@@ -114,6 +114,7 @@ t('DNS rebinding: public at the check, loopback at the download — refused whert('ratchet: every file that makes an outbound request itself is reviewed', () => {const REVIEWED = {'server/net/service-auth.js': 'Network JWKS (configured)',++ 'server/chat/account-data.js': 'Network internal API (configured), fixed paths: export parts and deletion confirmations','server/prefs/network-modules.js': 'Network user modules (configured)','server/chat/routes.js': 'GIF providers (fixed hosts)','server/chat/soundboard-service.js': '101soundboards (host allowlist; audio via safeLookup)',
@@ -16,7 +16,7 @@
@@ -16,7 +16,7 @@
"express": "^4.21.2",
"express": "^4.21.2",
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"multer": "^2.0.2",
"multer": "^2.0.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.57.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.22.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0",
"ws": "^8.18.0"
"ws": "^8.18.0"
@@ -1062,9 +1062,9 @@
@@ -1062,9 +1062,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.57.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.57.0",
- "integrity": "sha512-6hXpTo0y1h7YFQgyxkkWIoSx4CqSPDlGAK617MV2SkaO01zVOWKSQ1wor7aFOvrv+7u7a3KsIak3sBznNc39Vw==",
+ "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",
@@ -28,7 +28,7 @@
@@ -28,7 +28,7 @@
"express": "^4.21.2",
"express": "^4.21.2",
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"multer": "^2.0.2",
"multer": "^2.0.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.57.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.22.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0",
"ws": "^8.18.0"
"ws": "^8.18.0"
new file mode 100644
new file mode 100644
@@ -0,0 +1,212 @@
@@ -0,0 +1,212 @@
+'use strict';
+/**
+ * Account export and deletion → Chat (roadmap WS-B task 7, ADR-033; Contracts 0.71.0). Both arrive at POST
+ * /internal/events and are applied once per export or deletion (account_data_events); the delivery is answered after
+ * Network took the part or the confirmation, so a failure is redelivered without erasing twice. Chat keys people by
+ * subject and by Live user id (ctx_users maps the two).
+ *
+ * network.account.export_requested Chat's part (POST /internal/account-exports/:id/parts with a service token):
+ * their chat, room and direct messages, conversations and rooms, calls, blocks,
+ * emotes and sounds, and the moderation actions taken on them.
+ * network.account.deleted what the subject (and the accounts merged into it) wrote or set up goes:
+ * - chat, room and direct messages they sent, their DM memberships (a
+ * conversation left with nobody goes), room memberships and attachments;
+ * - rooms they own go when nobody else is in them, and otherwise stay without an
+ * owner; calls they made or took, and blocks both ways;
+ * - their channel's emotes, sounds, moderators, first-chat stats and audio
+ * requests, their tags, and the AI summaries and timeline about them;
+ * - the channel, stream and user mirrors of their Live account (the user mirror
+ * stays as a tombstone).
+ * Moderation actions stay as recorded. A staged table (channel_moderators,
+ * channel_moderation_settings, emotes, user_tags, chat_ai_summaries,
+ * chat_timeline_events) is erased here only while Chat writes it (table_authority);
+ * until then Live erases it and its capture keeps this copy current.
+ * Chat then confirms with counts.
+ */
+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 TOPICS = ['network.account.export_requested', 'network.account.deleted'];
+const ROW_LIMIT = 5000;
+
+const cols = (d, t) => { try { return d.prepare(`PRAGMA table_info(${t})`).all().map((c) => c.name); } catch { return []; } };
+const has = (d, t, c) => cols(d, t).includes(c);
+const inList = (xs) => `(${xs.map(() => '?').join(',')})`;
+
+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 TEXT,
+ applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+ )`);
+}
+
+/** The Live user ids of these subjects (Chat's ctx_users mirror). */
+function liveIds(d, subjects) {
+ return d.prepare(`SELECT id FROM ctx_users WHERE subject_id IN ${inList(subjects)}`).all(...subjects).map((r) => r.id);
+}
+
+/** WHERE for a table: its subject column(s) IN subjects, or its Live user column(s) IN ids. → [sql, params] | null. */
+function whereFor(d, table, subjectCols, idCols, subjects, ids) {
+ const parts = []; const params = [];
+ for (const c of subjectCols) if (has(d, table, c)) { parts.push(`${c} IN ${inList(subjects)}`); params.push(...subjects); }
+ if (ids.length) for (const c of idCols) if (has(d, table, c)) { parts.push(`${c} IN ${inList(ids)}`); params.push(...ids); }
+ return parts.length ? [parts.join(' OR '), params] : null;
+}
+
+// ── Export ─────────────────────────────────────────────────────
+
+const EXPORTS = [
+ ['chat_messages.json', 'chat_messages', ['subject_id'], ['user_id']],
+ ['room_messages.json', 'room_messages', ['subject_id'], ['user_id']],
+ ['direct_messages.json', 'dm_messages', ['sender_subject_id'], ['sender_id']],
+ ['conversations.json', 'dm_participants', ['subject_id'], ['user_id']],
+ ['rooms_owned.json', 'rooms', ['owner_subject'], ['owner_id']],
+ ['calls.json', 'calls', ['created_by_subject', 'target_subject'], ['created_by', 'target_user_id']],
+ ['blocks.json', 'network_blocks', ['blocker_subject'], []],
+ ['dm_blocks.json', 'dm_blocks', ['blocker_subject_id'], ['blocker_id']],
+ ['emotes.json', 'emotes', [], ['user_id']],
+ ['sounds.json', 'channel_sounds', ['created_by_subject_id'], ['created_by']],
+ ['moderation_on_you.json', 'moderation_actions', [], ['target_user_id']],
+];
+
+function exportPart(d, subject) {
+ const ids = liveIds(d, [subject]);
+ const files = []; const truncated = [];
+ for (const [name, table, sc, ic] of EXPORTS) {
+ const w = whereFor(d, table, sc, ic, [subject], ids);
+ if (!w) continue;
+ let rows = d.prepare(`SELECT * FROM ${table} WHERE ${w[0]} ORDER BY rowid DESC LIMIT ${ROW_LIMIT + 1}`).all(...w[1]);
+ if (table === 'moderation_actions') rows = rows.map(({ actor_user_id, actor_subject_id, ...r }) => r); // who acted is staff's, not theirs
+ if (!rows.length) continue;
+ if (rows.length > ROW_LIMIT) truncated.push(name);
+ files.push({ name, content: rows.slice(0, ROW_LIMIT) });
+ }
+ return { files, truncated };
+}
+
+// ── Deletion ───────────────────────────────────────────────────
+
+const STAGED = ['channel_moderators', 'channel_moderation_settings', 'emotes', 'user_tags', 'chat_ai_summaries', 'chat_timeline_events'];
+const chatWrites = (d, table) => { try { const r = d.prepare('SELECT authority FROM table_authority WHERE table_name = ?').get(table); return !!r && r.authority === 'chat'; } catch { return false; } };
+
+function erase(d, subjects) {
+ const ids = liveIds(d, subjects);
+ const erased = {}; const retained = {};
+ const add = (o, k, n) => { if (n) o[k] = (o[k] || 0) + n; };
+ const del = (key, table, sc, ic) => {
+ const w = whereFor(d, table, sc, ic, subjects, ids);
+ if (!w) return;
+ // A staged table Live still writes: Live erases it, and its capture brings the change here.
+ if (STAGED.includes(table) && !chatWrites(d, table)) { add(retained, 'erased_by_live', d.prepare(`SELECT COUNT(*) AS n FROM ${table} WHERE ${w[0]}`).get(...w[1]).n); return; }
+ add(erased, key, d.prepare(`DELETE FROM ${table} WHERE ${w[0]}`).run(...w[1]).changes);
+ };
+ d.transaction(() => {
+ del('messages', 'chat_messages', ['subject_id'], ['user_id']);
+ del('room_messages', 'room_messages', ['subject_id'], ['user_id']);
+ del('direct_messages', 'dm_messages', ['sender_subject_id'], ['sender_id']);
+ del('conversations_left', 'dm_participants', ['subject_id'], ['user_id']);
+ if (has(d, 'dm_conversations', 'id')) add(erased, 'conversations', d.prepare('DELETE FROM dm_conversations WHERE id NOT IN (SELECT conversation_id FROM dm_participants)').run().changes);
+ del('room_memberships', 'room_members', [], ['user_id']);
+ del('room_attachments', 'room_attachments', ['attached_by_subject'], []);
+ const owned = whereFor(d, 'rooms', ['owner_subject'], ['owner_id'], subjects, ids);
+ if (owned) {
+ for (const r of d.prepare(`SELECT id FROM rooms WHERE ${owned[0]}`).all(...owned[1])) {
+ const others = has(d, 'room_members', 'room_id') && d.prepare('SELECT 1 FROM room_members WHERE room_id = ? LIMIT 1').get(r.id);
+ if (others) { d.prepare('UPDATE rooms SET owner_subject = NULL WHERE id = ?').run(r.id); add(retained, 'rooms_without_owner', 1); }
+ else {
+ if (has(d, 'room_messages', 'room_id')) d.prepare('DELETE FROM room_messages WHERE room_id = ?').run(r.id);
+ d.prepare('DELETE FROM rooms WHERE id = ?').run(r.id);
+ add(erased, 'rooms', 1);
+ }
+ }
+ }
+ del('calls', 'calls', ['created_by_subject', 'target_subject'], ['created_by', 'target_user_id']);
+ del('blocks', 'network_blocks', ['blocker_subject', 'blocked_subject'], []);
+ del('blocks', 'dm_blocks', ['blocker_subject_id'], ['blocker_id', 'blocked_id']);
+ del('ai_summaries', 'chat_ai_summaries', ['subject_id'], []);
+ del('timeline', 'chat_timeline_events', ['subject_id'], []);
+ del('tags', 'user_tags', [], ['user_id']);
+ del('moderator_roles', 'channel_moderators', [], ['user_id']);
+ del('pending_messages', 'pending_ip_messages', [], ['user_id']);
+ if (ids.length) {
+ // Their channel: emotes, sounds, moderators, stats, audio requests, then the Live mirrors.
+ const channels = has(d, 'ctx_channels', 'user_id') ? d.prepare(`SELECT id FROM ctx_channels WHERE user_id IN ${inList(ids)}`).all(...ids).map((r) => r.id) : [];
+ if (channels.length && has(d, 'channel_moderators', 'channel_id') && chatWrites(d, 'channel_moderators')) add(erased, 'moderator_roles', d.prepare(`DELETE FROM channel_moderators WHERE channel_id IN ${inList(channels)}`).run(...channels).changes);
+ if (channels.length && has(d, 'channel_moderation_settings', 'channel_id') && chatWrites(d, 'channel_moderation_settings')) d.prepare(`DELETE FROM channel_moderation_settings WHERE channel_id IN ${inList(channels)}`).run(...channels);
+ del('emotes', 'emotes', [], ['user_id']);
+ del('sounds', 'channel_sounds', ['created_by_subject_id'], ['created_by']);
+ del('channel_stats', 'stream_first_chats', [], ['channel_user_id']);
+ del('audio_requests', 'audio_requests', [], ['channel_user_id']);
+ for (const t of ['ctx_channels', 'ctx_streams', 'ctx_managed_streams']) del('live_mirror', t, [], ['user_id']);
+ const set = ['username', 'display_name', 'avatar_url', 'profile_color', 'ban_reason'].filter((c) => has(d, 'ctx_users', c));
+ for (const id of ids) {
+ d.prepare(`UPDATE ctx_users SET ${set.map((c) => `${c} = ${c === 'username' ? '?' : 'NULL'}`).join(', ')} WHERE id = ?`).run(`deleted-${id}`, id);
+ }
+ }
+ const mod = whereFor(d, 'moderation_actions', ['actor_subject_id'], ['target_user_id', 'actor_user_id'], subjects, ids);
+ if (mod) add(retained, 'moderation_actions', d.prepare(`SELECT COUNT(*) AS n FROM moderation_actions WHERE ${mod[0]}`).get(...mod[1]).n);
+ })();
+ return { erased, retained, ids };
+}
+
+// ── Events ─────────────────────────────────────────────────────
+
+async function networkCall(path, body) {
+ const auth = require('../net/service-auth');
+ const config = require('../config');
+ for (let attempt = 0; attempt < 2; attempt++) {
+ const res = await fetch(`${config.networkInternalUrl}${path}`, { method: 'POST', headers: { 'Content-Type': 'application/json', ...(await auth.headers('openvibe.network')) }, body: JSON.stringify(body), signal: AbortSignal.timeout(30_000) });
+ if (res.status === 401 && attempt === 0) { auth.invalidate('openvibe.network'); continue; }
+ return res;
+ }
+ return null;
+}
+
+/** One envelope → 'exported' | 'erased' | 'confirmed' | 'closed' | 'unchanged' | 'ignored:<why>'; throws to be redelivered. */
+async function apply(ev, { send = networkCall, log = console, onErased = null } = {}) {
+ if (!ev || !TOPICS.includes(ev.event_type)) return 'ignored:type';
+ if (ev.source !== 'network') return 'ignored:source';
+ const p = 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, 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 (?, ?, ?, ?, ?)')
+ .run(p.export_id, 'export', p.subject, JSON.stringify({ result: outcome, files: part.files.length }), new Date().toISOString());
+ return outcome;
+ }
+ if (!DELETION_RE.test(String(p.deletion_id || '')) || !SUBJECT_RE.test(String(p.subject || ''))) return 'ignored:payload';
+ let rec = d.prepare('SELECT * FROM account_data_events WHERE id = ?').get(p.deletion_id);
+ let result = 'confirmed';
+ if (!rec) {
+ const subjects = [p.subject, ...(Array.isArray(p.aliases) ? p.aliases.filter((s) => SUBJECT_RE.test(String(s))) : [])];
+ const { ids, ...counts } = erase(d, subjects);
+ d.prepare('INSERT INTO account_data_events (id, kind, subject, outcome) VALUES (?, ?, ?, ?)').run(p.deletion_id, 'deletion', p.subject, JSON.stringify(counts));
+ log.log(`[AccountData] deletion ${p.deletion_id}: ${JSON.stringify(counts)}`);
+ if (onErased) { try { onErased(subjects, ids); } catch { /* caches refill anyway */ } }
+ rec = d.prepare('SELECT * FROM account_data_events WHERE id = ?').get(p.deletion_id);
+ result = 'erased';
+ }
+ 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: rec.applied_at, 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 = ? WHERE id = ?').run(new Date().toISOString(), p.deletion_id);
+ return result;
+}
+
+module.exports = { apply, exportPart, erase, ensureSchema, TOPICS };
@@ -47,7 +47,7 @@ const ctx = require('../live-context');
@@ -47,7 +47,7 @@ const ctx = require('../live-context');
const CONSUMER = 'chat';
const CONSUMER = 'chat';
const INBOX_TABLE = 'chat_event_inbox';
const INBOX_TABLE = 'chat_event_inbox';
-const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated', 'network.user.token_valid_after', 'vip.membership.changed', 'network.block.changed', 'network.subject.merged']);
+const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated', 'network.user.token_valid_after', 'vip.membership.changed', 'network.block.changed', 'network.subject.merged', 'network.account.export_requested', 'network.account.deleted']);
const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
const INBOX_KEEP_MS = 35 * 24 * 3600 * 1000; // Events keeps events 30 days: nothing older can be redelivered
const INBOX_KEEP_MS = 35 * 24 * 3600 * 1000; // Events keeps events 30 days: nothing older can be redelivered
@@ -166,6 +166,21 @@ function createEventsConsumer({ chatServer, secrets = [], now = () => Date.now()
@@ -166,6 +166,21 @@ function createEventsConsumer({ chatServer, secrets = [], now = () => Date.now()
stats.received++;
stats.received++;
stats.last_at = new Date(now()).toISOString();
stats.last_at = new Date(now()).toISOString();
stats.last_type = event.event_type;
stats.last_type = event.event_type;
+ if (event.event_type === 'network.account.export_requested' || event.event_type === 'network.account.deleted') {
+ // Account export and deletion (ADR-033, ../chat/account-data.js): once per export or deletion by its own record,
+ // answered after Network took the part or confirmation, so a failure is redelivered without erasing twice.
+ const onErased = (subjects, ids) => { for (const id of ids) ctx.invalidateUser(id); };
+ return require('../chat/account-data').apply(event, { log, onErased }).then((outcome) => {
+ if (String(outcome).startsWith('ignored:')) stats.ignored++; else stats.applied++;
+ stats.last_outcome = outcome;
+ res.status(200).json({ event_id: event.event_id, duplicate: outcome === 'unchanged', outcome });
+ }, (err) => {
+ stats.failed++;
+ stats.last_error = String(err.message || err).slice(0, 200);
+ log.error(`[Events consumer] ${event.event_id} (${event.event_type}) failed:`, err.message);
+ problem(500, 'chat.event_failed', 'processing failed; it will be retried');
+ });
+ }
let out;
let out;
try {
try {
out = apply(event);
out = apply(event);
new file mode 100644
new file mode 100644
@@ -0,0 +1,80 @@
@@ -0,0 +1,80 @@
+'use strict';
+/**
+ * Account export and deletion in Chat (roadmap WS-B task 7, ADR-033):
+ * - network.account.export_requested sends Chat's part to Network: their messages, DMs they sent, conversations,
+ * blocks, and moderation taken on them without who acted.
+ * - network.account.deleted erases what the subject and its merged-in aliases wrote or set up: chat, room and DM
+ * messages, DM memberships (a conversation left empty goes), blocks both ways, the channel's first-chat stats, and
+ * the Live mirrors (the user mirror as a tombstone). Moderation history stays. A staged table is left to Live
+ * while Live writes it.
+ * Chat confirms with counts; a failed confirmation is retried without erasing again, and a redelivery sends nothing
+ * twice. Chat subscribes to both.
+ */
+const assert = require('assert');
+const { ids, validate } = require('openvibe-contracts');
+const { boot, suite } = require('./helpers');
+const t = suite('account-data');
+let h, dana, otto, old;
+const ev = (type, payload) => ({ 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 });
+
+t('boot', async () => {
+ h = await boot();
+ dana = h.addUser('dana', { subject: ids.newId('user') });
+ old = h.addUser('dana_old', { subject: ids.newId('user') });
+ otto = h.addUser('otto', { subject: ids.newId('user') });
+ await h.ctx.sync();
+ assert.ok(require('../server/events/consumer').TOPICS.includes('network.account.deleted'), 'Chat subscribes');
+});
+
+t('export, then erase and confirm', async () => {
+ const accountData = require('../server/chat/account-data');
+ const d = require('../server/db/database').getDb();
+ d.prepare("INSERT INTO chat_messages (user_id, username, message, channel_user_id, subject_id) VALUES (?, 'dana', 'hi otto', ?, ?), (?, 'otto', 'hi dana', ?, ?), (?, 'dana_old', 'old me', ?, ?)")
+ .run(dana.id, otto.id, dana.subject_id, otto.id, dana.id, otto.subject_id, old.id, otto.id, old.subject_id);
+ const pair = (await h.http('POST', '/api/dm/conversations', { token: dana.token, body: { user_ids: [otto.id] } })).body.conversation.id;
+ assert.ok((await h.http('POST', `/api/dm/conversations/${pair}/messages`, { token: dana.token, body: { message: 'psst' } })).status < 300);
+ assert.ok((await h.http('POST', `/api/dm/conversations/${pair}/messages`, { token: otto.token, body: { message: 'what' } })).status < 300);
+ require('../server/chat/network-blocks').ensureSchema && require('../server/chat/network-blocks').ensureSchema();
+ d.prepare('INSERT INTO network_blocks (blocker_subject, blocked_subject, active, revision, updated_at) VALUES (?, ?, 1, 1, 0), (?, ?, 1, 1, 0)').run(dana.subject_id, otto.subject_id, otto.subject_id, dana.subject_id);
+ d.prepare("INSERT INTO moderation_actions (action_type, actor_user_id, target_user_id, actor_subject_id) VALUES ('timeout', ?, ?, ?)").run(otto.id, dana.id, otto.subject_id);
+ d.prepare("INSERT INTO user_tags (user_id, tag_id) VALUES (?, 'veteran')").run(dana.id);
+
+ 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 }; };
+
+ // ── Export ──
+ const exp = ev('network.account.export_requested', { export_id: `exp_${ids.ulid()}`, subject: dana.subject_id, requested_at: new Date().toISOString(), deadline: new Date(Date.now() + 1800000).toISOString() });
+ assert.strictEqual(await accountData.apply({ ...exp, source: 'live' }, { send }), 'ignored:source');
+ assert.strictEqual(await accountData.apply(exp, { send }), 'exported');
+ 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 files = Object.fromEntries(part.files.map((f) => [f.name, f.content]));
+ assert.deepStrictEqual(files['chat_messages.json'].map((m) => m.message), ['hi otto']);
+ assert.deepStrictEqual(files['direct_messages.json'].map((m) => m.message), ['psst']);
+ assert.ok(files['moderation_on_you.json'][0] && !('actor_user_id' in files['moderation_on_you.json'][0]), "moderation on them, without who acted");
+ assert.strictEqual(await accountData.apply(exp, { send }), 'unchanged');
+ assert.strictEqual(sent.length, 1);
+
+ // ── Deletion ──
+ const del = ev('network.account.deleted', { deletion_id: `del_${ids.ulid()}`, subject: dana.subject_id, aliases: [old.subject_id], requested_at: new Date().toISOString(), deleted_at: new Date().toISOString() });
+ failNext = true;
+ await assert.rejects(accountData.apply(del, { send }), /confirmation refused: 503/);
+ const q = (sql, ...a) => d.prepare(sql).all(...a);
+ assert.deepStrictEqual(q('SELECT message FROM chat_messages ORDER BY id').map((m) => m.message), ['hi dana'], "hers and the alias's gone; otto's stays");
+ assert.deepStrictEqual(q('SELECT message FROM dm_messages ORDER BY id').map((m) => m.message), ['what'], 'the DM she sent goes, his reply stays');
+ assert.deepStrictEqual(q('SELECT user_id FROM dm_participants WHERE conversation_id = ?', pair).map((p) => p.user_id), [otto.id]);
+ assert.strictEqual(q('SELECT * FROM network_blocks').length, 0, 'blocks both ways');
+ assert.strictEqual(q('SELECT * FROM moderation_actions').length, 1, 'moderation history stays');
+ assert.strictEqual(q('SELECT * FROM user_tags').length, 1, 'a staged table Live still writes is left to Live');
+ assert.strictEqual(d.prepare('SELECT username FROM ctx_users WHERE id = ?').get(dana.id).username, `deleted-${dana.id}`);
+ assert.strictEqual(await accountData.apply(del, { send }), 'confirmed', 'the retry confirms without erasing again');
+ const conf = sent[1];
+ assert.ok(validate('network.account-deletion-confirmation@1', conf.body).valid, JSON.stringify(validate('network.account-deletion-confirmation@1', conf.body).errors));
+ assert.deepStrictEqual([conf.body.erased.messages, conf.body.erased.direct_messages, conf.body.retained.moderation_actions, conf.body.retained.erased_by_live], [2, 1, 1, 1]);
+ assert.strictEqual(await accountData.apply(del, { send }), 'unchanged');
+ assert.strictEqual(sent.length, 2);
+});
+
+t.run(async () => { if (h && h.close) await h.close(); });
@@ -123,8 +123,8 @@ t('boot with the consumer on and a stub Events', async () => {
@@ -123,8 +123,8 @@ t('boot with the consumer on and a stub Events', async () => {
t('boot created one subscription per topic, to /internal/events, with the consumer’s secret', async () => {
t('boot created one subscription per topic, to /internal/events, with the consumer’s secret', async () => {
const results = await h.subscriptions.done;
const results = await h.subscriptions.done;
- assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created'], ['network.user.token_valid_after', 'created'], ['vip.membership.changed', 'created'], ['network.block.changed', 'created'], ['network.subject.merged', 'created']]);
- assert.strictEqual(stubEvents.subs.length, 6);
+ assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created'], ['network.user.token_valid_after', 'created'], ['vip.membership.changed', 'created'], ['network.block.changed', 'created'], ['network.subject.merged', 'created'], ['network.account.export_requested', 'created'], ['network.account.deleted', 'created']]);
+ assert.strictEqual(stubEvents.subs.length, 8);
for (const s of stubEvents.subs) {
for (const s of stubEvents.subs) {
assert.strictEqual(s.endpoint, `http://127.0.0.1:${h.port}/internal/events`);
assert.strictEqual(s.endpoint, `http://127.0.0.1:${h.port}/internal/events`);
assert.strictEqual(s.secret, SECRET, 'the first CHAT_EVENTS_SECRET is handed to Events');
assert.strictEqual(s.secret, SECRET, 'the first CHAT_EVENTS_SECRET is handed to Events');
@@ -139,17 +139,17 @@ t('a second boot changes nothing, and never re-enables a subscription an operato
@@ -139,17 +139,17 @@ t('a second boot changes nothing, and never re-enables a subscription an operato
const endpoint = subs.endpointFor(config, h.port);
const endpoint = subs.endpointFor(config, h.port);
const posts = stubEvents.posts;
const posts = stubEvents.posts;
let r = await subs.ensure({ config, endpoint });
let r = await subs.ensure({ config, endpoint });
- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists', 'exists', 'exists']);
+ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists', 'exists', 'exists', 'exists', 'exists']);
assert.strictEqual(stubEvents.posts, posts, 'nothing was created again');
assert.strictEqual(stubEvents.posts, posts, 'nothing was created again');
r = await subs.ensure({ config, endpoint, action: 'disable', topics: ['live.release.deployed'] });
r = await subs.ensure({ config, endpoint, action: 'disable', topics: ['live.release.deployed'] });
assert.deepStrictEqual(r.map((x) => [x.result, x.enabled]), [['disabled', false]]);
assert.deepStrictEqual(r.map((x) => [x.result, x.enabled]), [['disabled', false]]);
r = await subs.ensure({ config, endpoint });
r = await subs.ensure({ config, endpoint });
- assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true], ['network.user.token_valid_after', 'exists', true], ['vip.membership.changed', 'exists', true], ['network.block.changed', 'exists', true], ['network.subject.merged', 'exists', true]]);
+ assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true], ['network.user.token_valid_after', 'exists', true], ['vip.membership.changed', 'exists', true], ['network.block.changed', 'exists', true], ['network.subject.merged', 'exists', true], ['network.account.export_requested', 'exists', true], ['network.account.deleted', 'exists', true]]);
assert.strictEqual(stubEvents.subs[0].enabled, false, 'still disabled');
assert.strictEqual(stubEvents.subs[0].enabled, false, 'still disabled');
r = await subs.ensure({ config, endpoint, action: 'enable', topics: ['live.release.deployed'] });
r = await subs.ensure({ config, endpoint, action: 'enable', topics: ['live.release.deployed'] });
assert.strictEqual(stubEvents.subs[0].enabled, true);
assert.strictEqual(stubEvents.subs[0].enabled, true);
r = await subs.ensure({ config, endpoint, action: 'list' });
r = await subs.ensure({ config, endpoint, action: 'list' });
- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists', 'exists', 'exists']);
+ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists', 'exists', 'exists', 'exists', 'exists']);
});
});
t('signature v2 only: bad signature, stale or future timestamp, v1-only, unsigned → 401', async () => {
t('signature v2 only: bad signature, stale or future timestamp, v1-only, unsigned → 401', async () => {
@@ -114,6 +114,7 @@ t('DNS rebinding: public at the check, loopback at the download — refused wher
@@ -114,6 +114,7 @@ t('DNS rebinding: public at the check, loopback at the download — refused wher
t('ratchet: every file that makes an outbound request itself is reviewed', () => {
t('ratchet: every file that makes an outbound request itself is reviewed', () => {
const REVIEWED = {
const REVIEWED = {
'server/net/service-auth.js': 'Network JWKS (configured)',
'server/net/service-auth.js': 'Network JWKS (configured)',
+ 'server/chat/account-data.js': 'Network internal API (configured), fixed paths: export parts and deletion confirmations',
'server/prefs/network-modules.js': 'Network user modules (configured)',
'server/prefs/network-modules.js': 'Network user modules (configured)',
'server/chat/routes.js': 'GIF providers (fixed hosts)',
'server/chat/routes.js': 'GIF providers (fixed hosts)',
'server/chat/soundboard-service.js': '101soundboards (host allowlist; audio via safeLookup)',
'server/chat/soundboard-service.js': '101soundboards (host allowlist; audio via safeLookup)',