openvibers/openvibe.community Commits
Remove account merge handling (#356)
Changes
@@@ -40,7 +40,7 @@"Discord relay (off in production)","server-rendered pages with full SEO","Search documents and community.* events",-- "revocation, blocks, account merge, export and deletion from Network events",++ "revocation, blocks, export and deletion from Network events","N-1 compatibility tests","per-actor limits (openvibe-sdk/limits)"],@@@ -85,5 +85,5 @@ a restore, so prefer the code rollback.## 5. Afterwards- Account deletion erases a person's rows, and `added_by` becomes NULL where they added someone. A-- subject merge moves the rows (`server/identity/account-data.js`, `subject-merge.js`).++- When the forum moves to OpenVibe.Space (T10 step 3), `space_moderators` moves with the forum tables.ddeleted file mode 100644@@@ -1,80 +0,0 @@--'use strict';--/**-- * network.subject.merged → Community (roadmap WS-B task 5, ADR-029; Contracts 0.69.0). Two Network accounts became-- * one: `from` is an alias of `into`. In one transaction Community repoints what it keys by subject:-- * authorship pastes, paste comments, comments, threads, posts, attachments, pulse items-- * one per person per item paste likes, comment and thread votes, post reactions: the survivor's stays where both-- * had one (the other is dropped), then the item's cached count or score is recomputed from its rows-- * game progress (one row per person) and the block projection: the survivor's stays; blocking oneself is dropped-- * space moderators the folded-in account's spaces become the survivor's (one row per space and person)-- * Idempotent: the pulse consumer's inbox applies an event once, and a second run finds nothing under `from`.-- */--const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;--const MERGE_RE = /^mrg_[0-9A-HJKMNP-TV-Z]{26}$/;----const AUTHORSHIP = [-- ['pastes', 'owner_subject'], ['paste_comments', 'author_subject'], ['comments', 'author_subject'], ['threads', 'author_subject'],-- ['posts', 'author_subject'], ['attachments', 'owner_subject'], ['pulse_items', 'actor_subject'],-- ['submissions', 'author_subject'], ['submissions', 'reviewer_subject'], ['space_moderators', 'added_by'],--];--// Per person per item: [table, item column, recompute(db, itemId)]--const PER_ITEM = [-- ['paste_likes', 'paste_id', async (db, id) => await db.prepare('UPDATE pastes SET likes = (SELECT COUNT(*) FROM paste_likes WHERE paste_id = ?) WHERE id = ?').run(id, id)],-- ['comment_votes', 'comment_id', async (db, id) => {-- const a = await db.prepare('SELECT COALESCE(SUM(value), 0)::bigint AS s, COUNT(*) FILTER (WHERE value = 1) AS u, COUNT(*) FILTER (WHERE value = -1) AS d FROM comment_votes WHERE comment_id = ?').get(id);-- await db.prepare('UPDATE comments SET score = ?, upvotes = ?, downvotes = ? WHERE id = ?').run(a.s, a.u, a.d, id);-- }],-- ['thread_votes', 'thread_id', async (db, id) => await db.prepare('UPDATE threads SET score = (SELECT COALESCE(SUM(value), 0)::bigint FROM thread_votes WHERE thread_id = ?) WHERE id = ?').run(id, id)],-- ['post_reactions', 'post_id', null],--];----const hasColumn = async (db, table, col) => !!await db.prepare('SELECT 1 FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = ? AND column_name = ?').get(table, col);----/** The payload, or 'ignored:<why>'. */--function payloadOf(event) {-- if (!event || event.event_type !== 'network.subject.merged') return 'ignored:type';-- if (event.source !== 'network') return 'ignored:source';-- const p = event.payload && typeof event.payload === 'object' ? event.payload : {};-- if (!MERGE_RE.test(String(p.merge_id || '')) || !SUBJECT_RE.test(String(p.from || '')) || !SUBJECT_RE.test(String(p.into || '')) || p.from === p.into) return 'ignored:payload';-- return { merge_id: p.merge_id, from: p.from, into: p.into };--}----/** Apply one merge → 'merge:applied' (with counts logged). */--async function apply(db, { from, into, merge_id: mergeId }, { log = console } = {}) {-- const counts = await db.tx(async () => {-- const c = { authored: 0, moved: 0, dropped: 0 };-- for (const [table, col] of AUTHORSHIP) {-- if (await hasColumn(db, table, col)) c.authored += (await db.prepare(`UPDATE ${table} SET ${col} = ? WHERE ${col} = ?`).run(into, from)).changes;-- }-- for (const [table, item, recompute] of PER_ITEM) {-- if (!await hasColumn(db, table, 'subject_id')) continue;-- for (const r of await db.prepare(`SELECT ${item} AS i FROM ${table} WHERE subject_id = ?`).all(from)) {-- const clash = await db.prepare(`SELECT 1 FROM ${table} WHERE ${item} = ? AND subject_id = ?`).get(r.i, into);-- if (clash) { await db.prepare(`DELETE FROM ${table} WHERE ${item} = ? AND subject_id = ?`).run(r.i, from); c.dropped++; if (recompute) recompute(db, r.i); }-- else { await db.prepare(`UPDATE ${table} SET subject_id = ? WHERE ${item} = ? AND subject_id = ?`).run(into, r.i, from); c.moved++; }-- }-- }-- if (await hasColumn(db, 'game_progress', 'subject_id')) {-- if (await db.prepare('SELECT 1 FROM game_progress WHERE subject_id = ?').get(into)) await db.prepare('DELETE FROM game_progress WHERE subject_id = ?').run(from);-- else await db.prepare('UPDATE game_progress SET subject_id = ? WHERE subject_id = ?').run(into, from);-- }-- if (await hasColumn(db, 'space_moderators', 'subject_id')) {-- await db.prepare('DELETE FROM space_moderators WHERE subject_id = ? AND space_id IN (SELECT space_id FROM space_moderators WHERE subject_id = ?)').run(from, into);-- c.moved += (await db.prepare('UPDATE space_moderators SET subject_id = ? WHERE subject_id = ?').run(into, from)).changes;-- }-- if (await hasColumn(db, 'network_blocks', 'blocker_subject')) {-- for (const [col, other] of [['blocker_subject', 'blocked_subject'], ['blocked_subject', 'blocker_subject']]) {-- for (const b of await db.prepare(`SELECT ${other} AS o FROM network_blocks WHERE ${col} = ?`).all(from)) {-- const clash = b.o === into || await db.prepare(`SELECT 1 FROM network_blocks WHERE ${col} = ? AND ${other} = ?`).get(into, b.o);-- if (clash) await db.prepare(`DELETE FROM network_blocks WHERE ${col} = ? AND ${other} = ?`).run(from, b.o);-- else await db.prepare(`UPDATE network_blocks SET ${col} = ? WHERE ${col} = ? AND ${other} = ?`).run(into, from, b.o);-- }-- }-- }-- return c;-- });-- log.log(`[Merge] ${mergeId}: ${JSON.stringify(counts)}`);-- return 'merge:applied';--}----module.exports = { payloadOf, apply };@@@ -37,7 +37,7 @@ const store = require('./store');const blocks = require('../identity/blocks');const CONSUMER = 'community';--const TOPICS = Object.freeze(['live.stream.started', 'blog.post.published', 'wiki.page.published', 'wiki.page.updated', 'chat.room.message.created', 'news.story.published', 'vip.membership.changed', 'network.user.token_valid_after', 'network.block.changed', 'network.module.updated', 'network.subject.merged', 'network.account.export_requested', 'network.account.deleted']);++const TOPICS = Object.freeze(['live.stream.started', 'blog.post.published', 'wiki.page.published', 'wiki.page.updated', 'chat.room.message.created', 'news.story.published', 'vip.membership.changed', 'network.user.token_valid_after', 'network.block.changed', 'network.module.updated', 'network.account.export_requested', 'network.account.deleted']);const GAME_NAMESPACE = 'games.progress.summary';const GAME_MILESTONE = 5;const GAME_URL = 'https://openvibe.games/';@@@ -186,22 +186,7 @@ function createPulseConsumer({ db, secrets = [], vipCache = null, revocations =problem(500, 'community.event_failed', 'processing failed; it will be retried');});}-- if (event.event_type === 'network.subject.merged') {-- // Two accounts became one (ADR-029): the folded-in subject's rows become the survivor's (../identity/subject-merge.js).-- const merge = require('../identity/subject-merge');-- const p = merge.payloadOf(event);-- if (typeof p === 'string') { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: p }); }-- try {-- const r = await inbox.once(CONSUMER, event.event_id, async () => ({ outcome: await merge.apply(db, p, { log }) }));-- if (r.duplicate) stats.duplicates++; else stats.applied++;-- return res.json({ event_id: event.event_id, duplicate: r.duplicate, outcome: r.duplicate ? null : r.result.outcome });-- } catch (err) {-- stats.failed++;-- log.error(`[Pulse consumer] ${event.event_id} (${event.event_type}) failed:`, err.message);-- return problem(500, 'community.event_failed', 'processing failed; it will be retried');-- }-- }-- if (event.event_type === 'vip.membership.changed') {++ if (event.event_type === 'vip.membership.changed') {if (event.source !== 'vip' || !vipCache) { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: event.source !== 'vip' ? 'ignored:source' : 'ignored:no_gate' }); }const r = await inbox.once(CONSUMER, event.event_id, () => ({ outcome: vipCache.handleEvent(event) ? 'vip:invalidated' : 'vip:unchanged' }));if (r.duplicate) stats.duplicates++; else stats.applied++;@@@ -168,18 +168,12 @@ const { boot, check, done } = require('./helpers/app');assert.deepStrictEqual(await mods('general'), [alex.subject_id]);});-- await check('(f) account deletion erases the row; a subject merge moves it', async () => {++ await check('(f) account deletion erases its rows', async () => {const accountData = require('../server/identity/account-data');-- const merge = require('../server/identity/subject-merge');const ids = async (subject) => (await t.db.prepare('SELECT space_id FROM space_moderators WHERE subject_id = ? ORDER BY space_id').all(subject)).map((r) => Number(r.space_id));await call(`/api/v1/spaces/showcase/moderators/${carol.subject_id}`, { method: 'PUT', who: boss });await call(`/api/v1/spaces/general/moderators/${carol.subject_id}`, { method: 'PUT', who: boss });await call(`/api/v1/spaces/showcase/moderators/${bob.subject_id}`, { method: 'PUT', who: boss });-- // carol folds into bob: general moves over, showcase (both had it) stays bob's alone.-- await merge.apply(t.db, { from: carol.subject_id, into: bob.subject_id, merge_id: 'mrg_01JAB2C3D4E5F6G7H8J9K0MNP4' }, { log: { log() {} } });-- assert.deepStrictEqual(await ids(carol.subject_id), []);-- assert.deepStrictEqual(await ids(bob.subject_id), [1, 3]);-- assert.strictEqual((await call('/api/v1/spaces/general/threads', { who: bob })).json().viewer.can_moderate, true);// alex's account is deleted: the row goes, and the moderators alex added stay without them.await call(`/api/v1/spaces/general/moderators/${sam.subject_id}`, { method: 'PUT', who: alex });const out = await accountData.erase(t.db, [alex.subject_id]);ddeleted file mode 100644@@@ -1,81 +0,0 @@--'use strict';--/**-- * network.subject.merged in Community (roadmap WS-B task 5, ADR-029): a signed delivery makes the folded-in subject's-- * pastes, comments, threads, posts and pulse items the survivor's; per-item likes, votes and-- * reactions move unless the survivor has one there (then the other is dropped and the cached count or score is-- * recomputed); game progress and blocks keep the survivor's, and blocking oneself is dropped. A redelivery changes-- * nothing; another source or a malformed payload is ignored. Community subscribes to the topic.-- */--const assert = require('assert');--const http = require('http');--const express = require('express');--const { signDeliveryHeaders } = require('openvibe-sdk/events');--const { testDb } = require('./helpers/db');--const { createPulseConsumer, TOPICS } = require('../server/pulse/consumer');--let SECRET, db, app, KEEP, FOLD, OTHER, MERGE, evt;----// What the folded-in account left in Community.---- // paste 1: both liked----(async () => {-- SECRET = `whsec_${'ef'.repeat(32)}`;-- db = await testDb();-- app = express();-- app.use('/internal/events', createPulseConsumer({ db, secrets: [SECRET], log: { error() {}, log() {}, warn() {} } }).router);-- KEEP = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP1';-- FOLD = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP2';-- OTHER = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP3';-- MERGE = 'mrg_01JAB2C3D4E5F6G7H8J9K0MNP4';-- evt = (over = {}) => ({ event_id: 'evt_01JAB2C3D4E5F6G7H8J9K0MN09', event_type: 'network.subject.merged', version: 1, source: 'network', visibility: 'internal', timestamp: new Date().toISOString(),-- subject: { type: 'user', id: KEEP }, actor: { type: 'user', id: KEEP },-- payload: { merge_id: MERGE, from: FOLD, into: KEEP, merged_at: new Date().toISOString(), initiated_by: 'person' }, ...over });-- await db.prepare("INSERT INTO pastes (id, slug, owner_subject, likes) OVERRIDING SYSTEM VALUE VALUES (1, 'p1', ?, 2), (2, 'p2', ?, 1)").run(FOLD, OTHER);-- await db.prepare('INSERT INTO paste_likes (paste_id, subject_id) VALUES (1, ?), (1, ?), (2, ?)').run(FOLD, KEEP, FOLD);-- await db.prepare("INSERT INTO comment_threads (id, ref_service, ref_type, ref_id) OVERRIDING SYSTEM VALUE VALUES (1, 'live', 'vod', '9')").run();-- await db.prepare("INSERT INTO comments (id, thread_id, message, author_subject, score, upvotes, downvotes) OVERRIDING SYSTEM VALUE VALUES (1, 1, 'hi', ?, 2, 2, 0)").run(OTHER);-- await db.prepare('INSERT INTO comment_votes (comment_id, subject_id, value) VALUES (1, ?, 1), (1, ?, 1)').run(FOLD, KEEP);-- await db.prepare("INSERT INTO spaces (id, slug, name) OVERRIDING SYSTEM VALUE VALUES (901, 'merge-test', 'Merge test')").run();-- await db.prepare("INSERT INTO threads (id, space_id, slug, title, author_subject, score) OVERRIDING SYSTEM VALUE VALUES (1, 901, 't1', 'T1', ?, 1)").run(FOLD);-- await db.prepare('INSERT INTO thread_votes (thread_id, subject_id, value) VALUES (1, ?, 1)').run(FOLD);-- await db.prepare("INSERT INTO posts (id, thread_id, body_markdown, author_subject) OVERRIDING SYSTEM VALUE VALUES (1, 1, 'hello', ?)").run(FOLD);-- await db.prepare("INSERT INTO post_reactions (post_id, subject_id, reaction) VALUES (1, ?, 'fire'), (1, ?, 'heart')").run(FOLD, KEEP);-- await db.prepare("INSERT INTO game_progress (subject_id, level, updated_at) VALUES (?, 3, '2026-09-27'), (?, 5, '2026-09-27')").run(FOLD, KEEP);-- await db.prepare('INSERT INTO network_blocks (blocker_subject, blocked_subject, active, revision, updated_at) VALUES (?, ?, 1, 1, 0), (?, ?, 1, 1, 0)').run(FOLD, OTHER, FOLD, KEEP);-- assert.ok(TOPICS.includes('network.subject.merged'), 'Community subscribes to merges');-- const srv = await new Promise((r) => { const s = http.createServer(app).listen(0, '127.0.0.1', () => r(s)); });-- const post = (event) => {-- const body = JSON.stringify({ event, seq: 1 });-- return fetch(`http://127.0.0.1:${srv.address().port}/internal/events`, { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, SECRET) } }).then(async (r) => ({ status: r.status, json: await r.json() }));-- };-- try {-- let r = await post(evt({ source: 'live' }));-- assert.deepStrictEqual([r.status, r.json.outcome], [200, 'ignored:source']);-- r = await post(evt({ event_id: 'evt_01JAB2C3D4E5F6G7H8J9K0MN08', payload: { merge_id: 'nope', from: FOLD, into: KEEP } }));-- assert.strictEqual(r.json.outcome, 'ignored:payload');---- r = await post(evt());-- assert.deepStrictEqual([r.status, r.json.outcome], [200, 'merge:applied']);-- const owner = async (sql, ...a) => await db.prepare(sql).get(...a);-- assert.strictEqual((await owner('SELECT owner_subject AS s FROM pastes WHERE id = 1')).s, KEEP);-- assert.strictEqual((await owner('SELECT author_subject AS a FROM threads WHERE id = 1')).a, KEEP);-- assert.strictEqual((await owner('SELECT author_subject AS a FROM posts WHERE id = 1')).a, KEEP);-- // Likes: paste 1 had both (one dropped, count 1); paste 2's like moved.-- assert.deepStrictEqual((await db.prepare('SELECT paste_id AS p, subject_id AS s FROM paste_likes ORDER BY p').all()).map((x) => [x.p, x.s]), [[1, KEEP], [2, KEEP]]);-- assert.strictEqual((await owner('SELECT likes FROM pastes WHERE id = 1')).likes, 1, 'the cached count follows the rows');-- // Votes: comment 1 had both (score recomputed to 1); thread 1's vote moved.-- assert.deepStrictEqual(await owner('SELECT score, upvotes, downvotes FROM comments WHERE id = 1'), { score: 1, upvotes: 1, downvotes: 0 });-- assert.deepStrictEqual((await db.prepare('SELECT subject_id AS s FROM thread_votes').all()).map((x) => x.s), [KEEP]);-- assert.deepStrictEqual(await db.prepare('SELECT subject_id AS s, reaction FROM post_reactions').all(), [{ s: KEEP, reaction: 'heart' }], 'the survivor\'s reaction stays');-- assert.deepStrictEqual(await db.prepare('SELECT subject_id AS s, level FROM game_progress').all(), [{ s: KEEP, level: 5 }]);-- assert.deepStrictEqual(await db.prepare('SELECT blocker_subject AS a, blocked_subject AS b FROM network_blocks').all(), [{ a: KEEP, b: OTHER }], 'moved; blocking oneself dropped');-- assert.strictEqual((await db.prepare('SELECT COUNT(*) AS n FROM pastes WHERE owner_subject = ?').get(FOLD)).n + (await db.prepare('SELECT COUNT(*) AS n FROM paste_likes WHERE subject_id = ?').get(FOLD)).n, 0);---- r = await post(evt());-- assert.deepStrictEqual([r.status, r.json.duplicate], [200, true], 'a redelivery changes nothing');-- } finally {-- srv.close();-- await db.close();-- }-- console.log('community subject merge: all checks passed');--})().catch((e) => { console.error(e); process.exit(1); });
@@ -40,7 +40,7 @@
@@ -40,7 +40,7 @@
"Discord relay (off in production)",
"Discord relay (off in production)",
"server-rendered pages with full SEO",
"server-rendered pages with full SEO",
"Search documents and community.* events",
"Search documents and community.* events",
- "revocation, blocks, account merge, export and deletion from Network events",
+ "revocation, blocks, export and deletion from Network events",
"N-1 compatibility tests",
"N-1 compatibility tests",
"per-actor limits (openvibe-sdk/limits)"
"per-actor limits (openvibe-sdk/limits)"
],
],
@@ -85,5 +85,5 @@ a restore, so prefer the code rollback.
@@ -85,5 +85,5 @@ a restore, so prefer the code rollback.
## 5. Afterwards
## 5. Afterwards
- Account deletion erases a person's rows, and `added_by` becomes NULL where they added someone. A
- Account deletion erases a person's rows, and `added_by` becomes NULL where they added someone. A
- subject merge moves the rows (`server/identity/account-data.js`, `subject-merge.js`).
+
- When the forum moves to OpenVibe.Space (T10 step 3), `space_moderators` moves with the forum tables.
- When the forum moves to OpenVibe.Space (T10 step 3), `space_moderators` moves with the forum tables.
deleted file mode 100644
deleted file mode 100644
@@ -1,80 +0,0 @@
@@ -1,80 +0,0 @@
-'use strict';
-/**
- * network.subject.merged → Community (roadmap WS-B task 5, ADR-029; Contracts 0.69.0). Two Network accounts became
- * one: `from` is an alias of `into`. In one transaction Community repoints what it keys by subject:
- * authorship pastes, paste comments, comments, threads, posts, attachments, pulse items
- * one per person per item paste likes, comment and thread votes, post reactions: the survivor's stays where both
- * had one (the other is dropped), then the item's cached count or score is recomputed from its rows
- * game progress (one row per person) and the block projection: the survivor's stays; blocking oneself is dropped
- * space moderators the folded-in account's spaces become the survivor's (one row per space and person)
- * Idempotent: the pulse consumer's inbox applies an event once, and a second run finds nothing under `from`.
- */
-const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
-const MERGE_RE = /^mrg_[0-9A-HJKMNP-TV-Z]{26}$/;
-
-const AUTHORSHIP = [
- ['pastes', 'owner_subject'], ['paste_comments', 'author_subject'], ['comments', 'author_subject'], ['threads', 'author_subject'],
- ['posts', 'author_subject'], ['attachments', 'owner_subject'], ['pulse_items', 'actor_subject'],
- ['submissions', 'author_subject'], ['submissions', 'reviewer_subject'], ['space_moderators', 'added_by'],
-];
-// Per person per item: [table, item column, recompute(db, itemId)]
-const PER_ITEM = [
- ['paste_likes', 'paste_id', async (db, id) => await db.prepare('UPDATE pastes SET likes = (SELECT COUNT(*) FROM paste_likes WHERE paste_id = ?) WHERE id = ?').run(id, id)],
- ['comment_votes', 'comment_id', async (db, id) => {
- const a = await db.prepare('SELECT COALESCE(SUM(value), 0)::bigint AS s, COUNT(*) FILTER (WHERE value = 1) AS u, COUNT(*) FILTER (WHERE value = -1) AS d FROM comment_votes WHERE comment_id = ?').get(id);
- await db.prepare('UPDATE comments SET score = ?, upvotes = ?, downvotes = ? WHERE id = ?').run(a.s, a.u, a.d, id);
- }],
- ['thread_votes', 'thread_id', async (db, id) => await db.prepare('UPDATE threads SET score = (SELECT COALESCE(SUM(value), 0)::bigint FROM thread_votes WHERE thread_id = ?) WHERE id = ?').run(id, id)],
- ['post_reactions', 'post_id', null],
-];
-
-const hasColumn = async (db, table, col) => !!await db.prepare('SELECT 1 FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = ? AND column_name = ?').get(table, col);
-
-/** The payload, or 'ignored:<why>'. */
-function payloadOf(event) {
- if (!event || event.event_type !== 'network.subject.merged') return 'ignored:type';
- if (event.source !== 'network') return 'ignored:source';
- const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
- if (!MERGE_RE.test(String(p.merge_id || '')) || !SUBJECT_RE.test(String(p.from || '')) || !SUBJECT_RE.test(String(p.into || '')) || p.from === p.into) return 'ignored:payload';
- return { merge_id: p.merge_id, from: p.from, into: p.into };
-}
-
-/** Apply one merge → 'merge:applied' (with counts logged). */
-async function apply(db, { from, into, merge_id: mergeId }, { log = console } = {}) {
- const counts = await db.tx(async () => {
- const c = { authored: 0, moved: 0, dropped: 0 };
- for (const [table, col] of AUTHORSHIP) {
- if (await hasColumn(db, table, col)) c.authored += (await db.prepare(`UPDATE ${table} SET ${col} = ? WHERE ${col} = ?`).run(into, from)).changes;
- }
- for (const [table, item, recompute] of PER_ITEM) {
- if (!await hasColumn(db, table, 'subject_id')) continue;
- for (const r of await db.prepare(`SELECT ${item} AS i FROM ${table} WHERE subject_id = ?`).all(from)) {
- const clash = await db.prepare(`SELECT 1 FROM ${table} WHERE ${item} = ? AND subject_id = ?`).get(r.i, into);
- if (clash) { await db.prepare(`DELETE FROM ${table} WHERE ${item} = ? AND subject_id = ?`).run(r.i, from); c.dropped++; if (recompute) recompute(db, r.i); }
- else { await db.prepare(`UPDATE ${table} SET subject_id = ? WHERE ${item} = ? AND subject_id = ?`).run(into, r.i, from); c.moved++; }
- }
- }
- if (await hasColumn(db, 'game_progress', 'subject_id')) {
- if (await db.prepare('SELECT 1 FROM game_progress WHERE subject_id = ?').get(into)) await db.prepare('DELETE FROM game_progress WHERE subject_id = ?').run(from);
- else await db.prepare('UPDATE game_progress SET subject_id = ? WHERE subject_id = ?').run(into, from);
- }
- if (await hasColumn(db, 'space_moderators', 'subject_id')) {
- await db.prepare('DELETE FROM space_moderators WHERE subject_id = ? AND space_id IN (SELECT space_id FROM space_moderators WHERE subject_id = ?)').run(from, into);
- c.moved += (await db.prepare('UPDATE space_moderators SET subject_id = ? WHERE subject_id = ?').run(into, from)).changes;
- }
- if (await hasColumn(db, 'network_blocks', 'blocker_subject')) {
- for (const [col, other] of [['blocker_subject', 'blocked_subject'], ['blocked_subject', 'blocker_subject']]) {
- for (const b of await db.prepare(`SELECT ${other} AS o FROM network_blocks WHERE ${col} = ?`).all(from)) {
- const clash = b.o === into || await db.prepare(`SELECT 1 FROM network_blocks WHERE ${col} = ? AND ${other} = ?`).get(into, b.o);
- if (clash) await db.prepare(`DELETE FROM network_blocks WHERE ${col} = ? AND ${other} = ?`).run(from, b.o);
- else await db.prepare(`UPDATE network_blocks SET ${col} = ? WHERE ${col} = ? AND ${other} = ?`).run(into, from, b.o);
- }
- }
- }
- return c;
- });
- log.log(`[Merge] ${mergeId}: ${JSON.stringify(counts)}`);
- return 'merge:applied';
-}
-
-module.exports = { payloadOf, apply };
@@ -37,7 +37,7 @@ const store = require('./store');
@@ -37,7 +37,7 @@ const store = require('./store');
const blocks = require('../identity/blocks');
const blocks = require('../identity/blocks');
const CONSUMER = 'community';
const CONSUMER = 'community';
-const TOPICS = Object.freeze(['live.stream.started', 'blog.post.published', 'wiki.page.published', 'wiki.page.updated', 'chat.room.message.created', 'news.story.published', 'vip.membership.changed', 'network.user.token_valid_after', 'network.block.changed', 'network.module.updated', 'network.subject.merged', 'network.account.export_requested', 'network.account.deleted']);
+const TOPICS = Object.freeze(['live.stream.started', 'blog.post.published', 'wiki.page.published', 'wiki.page.updated', 'chat.room.message.created', 'news.story.published', 'vip.membership.changed', 'network.user.token_valid_after', 'network.block.changed', 'network.module.updated', 'network.account.export_requested', 'network.account.deleted']);
const GAME_NAMESPACE = 'games.progress.summary';
const GAME_NAMESPACE = 'games.progress.summary';
const GAME_MILESTONE = 5;
const GAME_MILESTONE = 5;
const GAME_URL = 'https://openvibe.games/';
const GAME_URL = 'https://openvibe.games/';
@@ -186,22 +186,7 @@ function createPulseConsumer({ db, secrets = [], vipCache = null, revocations =
@@ -186,22 +186,7 @@ function createPulseConsumer({ db, secrets = [], vipCache = null, revocations =
problem(500, 'community.event_failed', 'processing failed; it will be retried');
problem(500, 'community.event_failed', 'processing failed; it will be retried');
});
});
}
}
- if (event.event_type === 'network.subject.merged') {
- // Two accounts became one (ADR-029): the folded-in subject's rows become the survivor's (../identity/subject-merge.js).
- const merge = require('../identity/subject-merge');
- const p = merge.payloadOf(event);
- if (typeof p === 'string') { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: p }); }
- try {
- const r = await inbox.once(CONSUMER, event.event_id, async () => ({ outcome: await merge.apply(db, p, { log }) }));
- if (r.duplicate) stats.duplicates++; else stats.applied++;
- return res.json({ event_id: event.event_id, duplicate: r.duplicate, outcome: r.duplicate ? null : r.result.outcome });
- } catch (err) {
- stats.failed++;
- log.error(`[Pulse consumer] ${event.event_id} (${event.event_type}) failed:`, err.message);
- return problem(500, 'community.event_failed', 'processing failed; it will be retried');
- }
- }
- if (event.event_type === 'vip.membership.changed') {
+ if (event.event_type === 'vip.membership.changed') {
if (event.source !== 'vip' || !vipCache) { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: event.source !== 'vip' ? 'ignored:source' : 'ignored:no_gate' }); }
if (event.source !== 'vip' || !vipCache) { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: event.source !== 'vip' ? 'ignored:source' : 'ignored:no_gate' }); }
const r = await inbox.once(CONSUMER, event.event_id, () => ({ outcome: vipCache.handleEvent(event) ? 'vip:invalidated' : 'vip:unchanged' }));
const r = await inbox.once(CONSUMER, event.event_id, () => ({ outcome: vipCache.handleEvent(event) ? 'vip:invalidated' : 'vip:unchanged' }));
if (r.duplicate) stats.duplicates++; else stats.applied++;
if (r.duplicate) stats.duplicates++; else stats.applied++;
@@ -168,18 +168,12 @@ const { boot, check, done } = require('./helpers/app');
@@ -168,18 +168,12 @@ const { boot, check, done } = require('./helpers/app');
assert.deepStrictEqual(await mods('general'), [alex.subject_id]);
assert.deepStrictEqual(await mods('general'), [alex.subject_id]);
});
});
- await check('(f) account deletion erases the row; a subject merge moves it', async () => {
+ await check('(f) account deletion erases its rows', async () => {
const accountData = require('../server/identity/account-data');
const accountData = require('../server/identity/account-data');
- const merge = require('../server/identity/subject-merge');
const ids = async (subject) => (await t.db.prepare('SELECT space_id FROM space_moderators WHERE subject_id = ? ORDER BY space_id').all(subject)).map((r) => Number(r.space_id));
const ids = async (subject) => (await t.db.prepare('SELECT space_id FROM space_moderators WHERE subject_id = ? ORDER BY space_id').all(subject)).map((r) => Number(r.space_id));
await call(`/api/v1/spaces/showcase/moderators/${carol.subject_id}`, { method: 'PUT', who: boss });
await call(`/api/v1/spaces/showcase/moderators/${carol.subject_id}`, { method: 'PUT', who: boss });
await call(`/api/v1/spaces/general/moderators/${carol.subject_id}`, { method: 'PUT', who: boss });
await call(`/api/v1/spaces/general/moderators/${carol.subject_id}`, { method: 'PUT', who: boss });
await call(`/api/v1/spaces/showcase/moderators/${bob.subject_id}`, { method: 'PUT', who: boss });
await call(`/api/v1/spaces/showcase/moderators/${bob.subject_id}`, { method: 'PUT', who: boss });
- // carol folds into bob: general moves over, showcase (both had it) stays bob's alone.
- await merge.apply(t.db, { from: carol.subject_id, into: bob.subject_id, merge_id: 'mrg_01JAB2C3D4E5F6G7H8J9K0MNP4' }, { log: { log() {} } });
- assert.deepStrictEqual(await ids(carol.subject_id), []);
- assert.deepStrictEqual(await ids(bob.subject_id), [1, 3]);
- assert.strictEqual((await call('/api/v1/spaces/general/threads', { who: bob })).json().viewer.can_moderate, true);
// alex's account is deleted: the row goes, and the moderators alex added stay without them.
// alex's account is deleted: the row goes, and the moderators alex added stay without them.
await call(`/api/v1/spaces/general/moderators/${sam.subject_id}`, { method: 'PUT', who: alex });
await call(`/api/v1/spaces/general/moderators/${sam.subject_id}`, { method: 'PUT', who: alex });
const out = await accountData.erase(t.db, [alex.subject_id]);
const out = await accountData.erase(t.db, [alex.subject_id]);
deleted file mode 100644
deleted file mode 100644
@@ -1,81 +0,0 @@
@@ -1,81 +0,0 @@
-'use strict';
-/**
- * network.subject.merged in Community (roadmap WS-B task 5, ADR-029): a signed delivery makes the folded-in subject's
- * pastes, comments, threads, posts and pulse items the survivor's; per-item likes, votes and
- * reactions move unless the survivor has one there (then the other is dropped and the cached count or score is
- * recomputed); game progress and blocks keep the survivor's, and blocking oneself is dropped. A redelivery changes
- * nothing; another source or a malformed payload is ignored. Community subscribes to the topic.
- */
-const assert = require('assert');
-const http = require('http');
-const express = require('express');
-const { signDeliveryHeaders } = require('openvibe-sdk/events');
-const { testDb } = require('./helpers/db');
-const { createPulseConsumer, TOPICS } = require('../server/pulse/consumer');
-let SECRET, db, app, KEEP, FOLD, OTHER, MERGE, evt;
-
-// What the folded-in account left in Community.
-
- // paste 1: both liked
-
-(async () => {
- SECRET = `whsec_${'ef'.repeat(32)}`;
- db = await testDb();
- app = express();
- app.use('/internal/events', createPulseConsumer({ db, secrets: [SECRET], log: { error() {}, log() {}, warn() {} } }).router);
- KEEP = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP1';
- FOLD = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP2';
- OTHER = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP3';
- MERGE = 'mrg_01JAB2C3D4E5F6G7H8J9K0MNP4';
- evt = (over = {}) => ({ event_id: 'evt_01JAB2C3D4E5F6G7H8J9K0MN09', event_type: 'network.subject.merged', version: 1, source: 'network', visibility: 'internal', timestamp: new Date().toISOString(),
- subject: { type: 'user', id: KEEP }, actor: { type: 'user', id: KEEP },
- payload: { merge_id: MERGE, from: FOLD, into: KEEP, merged_at: new Date().toISOString(), initiated_by: 'person' }, ...over });
- await db.prepare("INSERT INTO pastes (id, slug, owner_subject, likes) OVERRIDING SYSTEM VALUE VALUES (1, 'p1', ?, 2), (2, 'p2', ?, 1)").run(FOLD, OTHER);
- await db.prepare('INSERT INTO paste_likes (paste_id, subject_id) VALUES (1, ?), (1, ?), (2, ?)').run(FOLD, KEEP, FOLD);
- await db.prepare("INSERT INTO comment_threads (id, ref_service, ref_type, ref_id) OVERRIDING SYSTEM VALUE VALUES (1, 'live', 'vod', '9')").run();
- await db.prepare("INSERT INTO comments (id, thread_id, message, author_subject, score, upvotes, downvotes) OVERRIDING SYSTEM VALUE VALUES (1, 1, 'hi', ?, 2, 2, 0)").run(OTHER);
- await db.prepare('INSERT INTO comment_votes (comment_id, subject_id, value) VALUES (1, ?, 1), (1, ?, 1)').run(FOLD, KEEP);
- await db.prepare("INSERT INTO spaces (id, slug, name) OVERRIDING SYSTEM VALUE VALUES (901, 'merge-test', 'Merge test')").run();
- await db.prepare("INSERT INTO threads (id, space_id, slug, title, author_subject, score) OVERRIDING SYSTEM VALUE VALUES (1, 901, 't1', 'T1', ?, 1)").run(FOLD);
- await db.prepare('INSERT INTO thread_votes (thread_id, subject_id, value) VALUES (1, ?, 1)').run(FOLD);
- await db.prepare("INSERT INTO posts (id, thread_id, body_markdown, author_subject) OVERRIDING SYSTEM VALUE VALUES (1, 1, 'hello', ?)").run(FOLD);
- await db.prepare("INSERT INTO post_reactions (post_id, subject_id, reaction) VALUES (1, ?, 'fire'), (1, ?, 'heart')").run(FOLD, KEEP);
- await db.prepare("INSERT INTO game_progress (subject_id, level, updated_at) VALUES (?, 3, '2026-09-27'), (?, 5, '2026-09-27')").run(FOLD, KEEP);
- await db.prepare('INSERT INTO network_blocks (blocker_subject, blocked_subject, active, revision, updated_at) VALUES (?, ?, 1, 1, 0), (?, ?, 1, 1, 0)').run(FOLD, OTHER, FOLD, KEEP);
- assert.ok(TOPICS.includes('network.subject.merged'), 'Community subscribes to merges');
- const srv = await new Promise((r) => { const s = http.createServer(app).listen(0, '127.0.0.1', () => r(s)); });
- const post = (event) => {
- const body = JSON.stringify({ event, seq: 1 });
- return fetch(`http://127.0.0.1:${srv.address().port}/internal/events`, { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, SECRET) } }).then(async (r) => ({ status: r.status, json: await r.json() }));
- };
- try {
- let r = await post(evt({ source: 'live' }));
- assert.deepStrictEqual([r.status, r.json.outcome], [200, 'ignored:source']);
- r = await post(evt({ event_id: 'evt_01JAB2C3D4E5F6G7H8J9K0MN08', payload: { merge_id: 'nope', from: FOLD, into: KEEP } }));
- assert.strictEqual(r.json.outcome, 'ignored:payload');
-
- r = await post(evt());
- assert.deepStrictEqual([r.status, r.json.outcome], [200, 'merge:applied']);
- const owner = async (sql, ...a) => await db.prepare(sql).get(...a);
- assert.strictEqual((await owner('SELECT owner_subject AS s FROM pastes WHERE id = 1')).s, KEEP);
- assert.strictEqual((await owner('SELECT author_subject AS a FROM threads WHERE id = 1')).a, KEEP);
- assert.strictEqual((await owner('SELECT author_subject AS a FROM posts WHERE id = 1')).a, KEEP);
- // Likes: paste 1 had both (one dropped, count 1); paste 2's like moved.
- assert.deepStrictEqual((await db.prepare('SELECT paste_id AS p, subject_id AS s FROM paste_likes ORDER BY p').all()).map((x) => [x.p, x.s]), [[1, KEEP], [2, KEEP]]);
- assert.strictEqual((await owner('SELECT likes FROM pastes WHERE id = 1')).likes, 1, 'the cached count follows the rows');
- // Votes: comment 1 had both (score recomputed to 1); thread 1's vote moved.
- assert.deepStrictEqual(await owner('SELECT score, upvotes, downvotes FROM comments WHERE id = 1'), { score: 1, upvotes: 1, downvotes: 0 });
- assert.deepStrictEqual((await db.prepare('SELECT subject_id AS s FROM thread_votes').all()).map((x) => x.s), [KEEP]);
- assert.deepStrictEqual(await db.prepare('SELECT subject_id AS s, reaction FROM post_reactions').all(), [{ s: KEEP, reaction: 'heart' }], 'the survivor\'s reaction stays');
- assert.deepStrictEqual(await db.prepare('SELECT subject_id AS s, level FROM game_progress').all(), [{ s: KEEP, level: 5 }]);
- assert.deepStrictEqual(await db.prepare('SELECT blocker_subject AS a, blocked_subject AS b FROM network_blocks').all(), [{ a: KEEP, b: OTHER }], 'moved; blocking oneself dropped');
- assert.strictEqual((await db.prepare('SELECT COUNT(*) AS n FROM pastes WHERE owner_subject = ?').get(FOLD)).n + (await db.prepare('SELECT COUNT(*) AS n FROM paste_likes WHERE subject_id = ?').get(FOLD)).n, 0);
-
- r = await post(evt());
- assert.deepStrictEqual([r.status, r.json.duplicate], [200, true], 'a redelivery changes nothing');
- } finally {
- srv.close();
- await db.close();
- }
- console.log('community subject merge: all checks passed');
-})().catch((e) => { console.error(e); process.exit(1); });