openvibers/openvibe.live Commits
Remove account merge handling (#787)
Changes
@@@ -245,7 +245,7 @@ through the SDK inbox (`idempotency_receipts`):|---|---|---|---|| `openre.session.*` | `/internal/openre-events` | `OPENRE_EVENTS_SECRET` | OpenRestream's `scripts/subscribe-live-events.js` || `media.vod.*`, `media.clip.*`, `media.storage.*` | `/internal/media-events` | `MEDIA_EVENTS_SECRET` | `scripts/subscribe-media-events.js` |--| `network.user.*`, `network.follow.*`, `network.subject.merged`, `network.account.*` | `/internal/network-events` | `LIVE_EVENTS_SECRET` | `scripts/subscribe-media-events.js --network` |++| `network.user.*`, `network.follow.*`, `network.account.*` | `/internal/network-events` | `LIVE_EVENTS_SECRET` | `scripts/subscribe-media-events.js --network` || `billing.transaction.settled`, `billing.entitlement.changed` | `/internal/billing-events` | `LIVE_EVENTS_SECRET` | `scripts/subscribe-media-events.js --billing` || `tips.interaction.ready` | `/internal/tips-events` | `LIVE_EVENTS_SECRET` | `scripts/subscribe-media-events.js --tips` |@@@ -36,7 +36,7 @@ const fs = require('fs');const TOPICS = ['media.vod.*', 'media.clip.*', 'media.storage.*', 'media.object.*'];const DEFAULT_ENDPOINT = 'http://127.0.0.1:3000/internal/media-events';--const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted', 'network.subject.merged', 'network.account.export_requested', 'network.account.deleted'];++const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted', 'network.account.export_requested', 'network.account.deleted'];const NETWORK_ENDPOINT = 'http://127.0.0.1:3000/internal/network-events';const BILLING_TOPICS = ['billing.transaction.settled', 'billing.entitlement.changed'];const BILLING_ENDPOINT = 'http://127.0.0.1:3000/internal/billing-events';@@@ -10,8 +10,6 @@* token_valid_after the person's older tokens are refused (./revocations.js)* updated the person's profile, role and ban (./subject-projection.js, WS-B task 2)* follow.created/.deleted Live's follows table as a projection of Network's graph (../social/network-follows.js)-- * subject.merged two accounts became one (ADR-029): the folded-in Live user's follows, channel points and streams-- * move to the survivor's (./subject-merge.js), once per merge* account.export_requested / account.deleted the person's export part goes to Network, or their Live rows are* erased and the erasure confirmed (./account-data.js, ADR-033), once per export or deletion*@@@ -32,15 +30,13 @@ const notifyChat = async (userId) => {function secret() { return process.env.LIVE_EVENTS_SECRET || process.env.MEDIA_EVENTS_SECRET || ''; }--/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>' (a merge, export or deletion: a promise). */++/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>' (an account event: a promise). */async function apply(ev) {if (!ev || !EVENT_ID_RE.test(String(ev.event_id || ''))) return 'ignored:envelope';-- if (ev.event_type === 'network.subject.merged') return ev.source === 'network' ? await require('./subject-merge').apply(ev) : 'ignored:source';if (ev.event_type === 'network.account.export_requested' || ev.event_type === 'network.account.deleted') return ev.source === 'network' ? await require('./account-data').apply(ev) : 'ignored:source';const follow = ev.event_type === 'network.follow.created' || ev.event_type === 'network.follow.deleted';if (ev.event_type !== 'network.user.token_valid_after' && ev.event_type !== 'network.user.updated' && !follow) return 'ignored:type';if (ev.source !== 'network') return 'ignored:source';-- // The follow graph is Network's (ADR-030); Live's follows table is its projection.if (follow) return await require('../social/network-follows').apply(ev);const p = ev.payload && typeof ev.payload === 'object' ? ev.payload : {};if (ev.event_type === 'network.user.updated') return await require('./subject-projection').apply(p, { notify: notifyChat });@@@ -58,10 +54,10 @@ async function handler(req, res) {const delivery = parseDelivery(req.rawBody, req.headers, key, { requireV2: true });if (!delivery) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }stats.received++;-- const count = (out) => { if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated' || out === 'followed' || out === 'unfollowed' || out === 'merged' || out === 'relinked' || out === 'exported' || out === 'erased' || out === 'confirmed') stats.profiles++; else stats.ignored++; };++ const count = (out) => { if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated' || out === 'followed' || out === 'unfollowed' || out === 'relinked' || out === 'exported' || out === 'erased' || out === 'confirmed') stats.profiles++; else stats.ignored++; };const out = await apply(delivery.event);if (out && typeof out.then === 'function') {-- // A merge, export or deletion: answer after it applied, so a failure is redelivered.++ // Account export or deletion: answer after it applied, so a failure is redelivered.return out.then((o) => { count(o); res.status(204).end(); }, (e) => { console.error(`[NetworkEvents] ${delivery.event.event_type} failed:`, e.message); res.status(500).json({ error: 'not applied' }); });}count(out);ddeleted file mode 100644@@@ -1,106 +0,0 @@--'use strict';--/**-- * network.subject.merged → Live (roadmap WS-B task 5, ADR-029; Contracts 0.69.0). Two Network accounts became one:-- * `from` is now an alias of `into`. Live has at most one user linked to each (linked_accounts, service 'network'):-- *-- * neither, or only the survivor's nothing of Live's to move-- * only the folded-in account's that Live user becomes the survivor's: its network link names `into` (and the-- * survivor's Network user id, resolved from Network)-- * both in one transaction, the folded-in user's follows (as follower and as the followed-- * channel), channel points (as viewer and as the channel; balances of the same pair-- * are summed, each move logged in channel_points_log once) and streams move to the-- * survivor's user. A follow the survivor already has, or following oneself, is-- * dropped and counted. The folded-in Live user stays, marked merged_into.-- *-- * Live's follows are a projection of Network's graph (ADR-030), which Network already merged; this keeps the-- * projection in step without waiting for a rebuild. VODs and clips are Media's (it follows the same event), chat is-- * Chat's. Applied once per merge_id (subject_merges).-- */--const db = require('../db/database');----const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;--const MERGE_RE = /^mrg_[0-9A-HJKMNP-TV-Z]{26}$/;----const liveUserOf = async (d, subject) => {-- const r = await d.prepare("SELECT user_id FROM linked_accounts WHERE service = 'network' AND subject_id = ? ORDER BY id LIMIT 1").get(subject);-- return r ? r.user_id : null;--};----/** Move everything of Live user `a` to Live user `b` (one transaction). → counts. */--async function moveUser(d, a, b, mergeId) {-- const out = { follows: 0, followers: 0, follows_dropped: 0, points: 0, points_summed: 0, streams: 0 };-- // Follows: as the follower, then as the followed channel.-- for (const [col, other, key] of [['follower_id', 'streamer_id', 'follows'], ['streamer_id', 'follower_id', 'followers']]) {-- for (const f of await d.prepare(`SELECT id, ${other} AS o FROM follows WHERE ${col} = ?`).all(a)) {-- const clash = f.o === b || await d.prepare(`SELECT 1 FROM follows WHERE ${col} = ? AND ${other} = ?`).get(b, f.o);-- if (clash) { await d.prepare('DELETE FROM follows WHERE id = ?').run(f.id); out.follows_dropped++; }-- else { await d.prepare(`UPDATE follows SET ${col} = ? WHERE id = ?`).run(b, f.id); out[key]++; }-- }-- }-- // Channel points (loyalty, ADR-012): first as the viewer, then as the channel; the same pair twice is summed.-- const logMove = d.prepare('INSERT INTO channel_points_log (idempotency_key, user_id, streamer_id, delta, reason) VALUES (?, ?, ?, ?, ?) ON CONFLICT DO NOTHING');-- const add = async (u, s, bal) => {-- const have = await d.prepare('SELECT 1 FROM channel_points WHERE user_id = ? AND streamer_id = ?').get(u, s);-- if (have) { await d.prepare('UPDATE channel_points SET balance = balance + ?, updated_at = ov_now() WHERE user_id = ? AND streamer_id = ?').run(bal, u, s); out.points_summed++; }-- else await d.prepare('INSERT INTO channel_points (user_id, streamer_id, balance) VALUES (?, ?, ?)').run(u, s, bal);-- };-- const move = async (fromU, fromS, toU, toS, bal, key) => {-- await d.prepare('DELETE FROM channel_points WHERE user_id = ? AND streamer_id = ?').run(fromU, fromS);-- await add(toU, toS, bal);-- if (bal) {-- await logMove.run(`live:cp:merge:${mergeId}:${key}:out`, fromU, fromS, -bal, 'account_merge');-- await logMove.run(`live:cp:merge:${mergeId}:${key}:in`, toU, toS, bal, 'account_merge');-- }-- out.points++;-- };-- for (const p of await d.prepare('SELECT streamer_id AS s, balance FROM channel_points WHERE user_id = ?').all(a)) await move(a, p.s, b, p.s === a ? b : p.s, p.balance, `v:${p.s}`);-- for (const p of await d.prepare('SELECT user_id AS u, balance FROM channel_points WHERE streamer_id = ?').all(a)) await move(p.u, a, p.u, b, p.balance, `c:${p.u}`);-- out.streams = (await d.prepare('UPDATE streams SET user_id = ? WHERE user_id = ?').run(b, a)).changes;-- await d.prepare('UPDATE users SET merged_into = ? WHERE id = ?').run(b, a);-- return out;--}----/**-- * Apply one network.subject.merged envelope → 'merged' | 'relinked' | 'nothing' | 'unchanged' | 'ignored:<why>'.-- * resolveNetworkId(subject) → the survivor's Network user id (injected by tests; defaults to Network's resolve-batch).-- */--async function apply(ev, { resolveNetworkId = defaultResolve } = {}) {-- const p = ev && ev.payload && typeof ev.payload === 'object' ? ev.payload : {};-- if (!MERGE_RE.test(String(p.merge_id || '')) || !SUBJECT_RE.test(String(p.from || '')) || !SUBJECT_RE.test(String(p.into || '')) || p.from === p.into) return 'ignored:payload';-- const d = db.getDb();-- if (await d.prepare('SELECT 1 FROM subject_merges WHERE merge_id = ?').get(p.merge_id)) return 'unchanged';-- const a = await liveUserOf(d, p.from);-- const b = await liveUserOf(d, p.into);-- let result; let outcome;-- if (!a) { result = 'nothing'; outcome = { reason: 'no Live user for the folded-in account' }; }-- else if (!b) {-- const networkId = await resolveNetworkId(p.into);-- try {-- await d.prepare("UPDATE linked_accounts SET subject_id = ?, service_user_id = COALESCE(?::text, service_user_id) WHERE service = 'network' AND user_id = ?")-- .run(p.into, networkId == null ? null : String(networkId), a);-- result = 'relinked'; outcome = { user: a, network_user_id: networkId == null ? null : Number(networkId) };-- } catch (e) {-- // Another Live user already carries the survivor's Network id without its subject: left for staff, logged.-- result = 'conflict'; outcome = { user: a, error: String(e.message).slice(0, 200) };-- }-- } else if (a === b) { result = 'nothing'; outcome = { reason: 'one Live user for both' }; }-- else {-- outcome = await d.tx(async () => await moveUser(d, a, b, p.merge_id));-- result = 'merged';-- }-- await d.prepare('INSERT INTO subject_merges (merge_id, from_subject, into_subject, from_user_id, into_user_id, outcome) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING')-- .run(p.merge_id, p.from, p.into, a, b, JSON.stringify({ result, ...outcome }));-- console.log(`[Merge] ${p.merge_id}: ${result} ${JSON.stringify(outcome)}`);-- return result;--}----async function defaultResolve(subject) {-- try {-- const { networkPost } = require('./identity-sync');-- const r = await networkPost('/internal/identity/resolve-batch', { subject_ids: [subject] });-- const hit = r && r.results && r.results[subject];-- return hit && hit.network_user_id != null ? hit.network_user_id : null;-- } catch { return null; }--}----module.exports = { apply, moveUser };@@@ -51,7 +51,6 @@ const KINDS = Object.freeze([Object.freeze({ id: 'media-request', match: /^Media request: (.+)$/, site: 'live', subject: 'media', text: 'for the stream {name}' }),Object.freeze({ id: 'media-refund', match: /^Refund: (.+)$/, site: 'live', subject: 'media', text: 'for a refund of {name}' }),Object.freeze({ id: 'migration', match: 'migration:hobo-coins', site: 'network', text: 'for OpenCoins moved from the old Live wallet' }),-- Object.freeze({ id: 'account-merge', match: 'account_merge', site: 'network', text: 'for merging your accounts' }),Object.freeze({ id: 'account-deleted', match: 'account_deleted', site: 'network', text: 'for closing your account' }),Object.freeze({ id: 'watch', match: 'watch', site: 'live', text: 'for watching a stream' }),Object.freeze({ id: 'chat', match: 'chat', site: 'live', text: 'for chatting on a stream' }),@@@ -227,7 +227,7 @@ server.listen(0, '127.0.0.1', async () => {// Live subscribes to the follow events.assert.match(fs.readFileSync(path.join(__dirname, '..', 'scripts', 'subscribe-media-events.js'), 'utf8'), /'network\.follow\.created', 'network\.follow\.deleted'/);-- // Only the projection writes follows (and the account merge repoints it); no Live-only path is left.++ // Only the projection writes follows (and Network projects it); no Live-only path is left.const root = path.join(__dirname, '..', 'server');const writers = [];const walk = (dir) => { for (const e of fs.readdirSync(dir, { withFileTypes: true })) {@@@ -236,7 +236,7 @@ server.listen(0, '127.0.0.1', async () => {else if (e.name.endsWith('.js') && /\b(INSERT(\s+OR\s+\w+)?\s+INTO|DELETE\s+FROM|UPDATE)\s+follows\b/i.test(fs.readFileSync(p, 'utf8'))) writers.push(path.relative(root, p));} };walk(root);-- assert.deepStrictEqual(writers.sort(), ['auth/subject-merge.js', 'social/network-follows.js'], 'follows writers');++ assert.deepStrictEqual(writers.sort(), ['social/network-follows.js'], 'follows writers');assert.strictEqual(typeof db.followUser, 'undefined');assert.strictEqual(typeof db.unfollowUser, 'undefined');assert.ok(!/FOLLOWS_AUTHORITY/.test(fs.readFileSync(path.join(root, 'social', 'network-follows.js'), 'utf8')), 'no switch back to Live-only follows');ddeleted file mode 100644@@@ -1,92 +0,0 @@--'use strict';--// network.subject.merged → Live (roadmap WS-B task 5, ADR-029): through POST /internal/network-events (signed), the--// folded-in Live user's follows (both ways), channel points (both ways, a shared pair summed, every move logged once)--// and streams move to the survivor's user; a follow it already has and following oneself are dropped; the folded-in--// user is marked merged_into. With only the folded-in account on Live, that user is relinked to the survivor. A--// redelivery changes nothing; a merge nobody on Live had is acknowledged. The subscription asks for the topic.--// node test/subject-merge.test.js--const assert = require('assert');--const http = require('http');----process.env.NODE_ENV = 'test';--process.env.LIVE_EVENTS_SECRET = 's'.repeat(40);--const quiet = console.log;--console.log = (...a) => { if (!/^\[/.test(String(a[0]))) quiet(...a); };--console.warn = () => {};----const express = require('express');--const { ids, validate } = require('openvibe-contracts');--const { signDeliveryHeaders } = require('openvibe-sdk/events');--const db = require('../server/db/database');----const KEEP = ids.newId('user'), FOLD = ids.newId('user'), X = ids.newId('user'), LONE = ids.newId('user'), LONE_INTO = ids.newId('user');----const envelope = (from, into, mergeId = `mrg_${ids.ulid()}`) => {-- const payload = { merge_id: mergeId, from, into, merged_at: new Date().toISOString(), initiated_by: 'person', split_until: new Date(Date.now() + 30 * 86400000).toISOString() };-- assert.ok(validate('network.subject.merged@1', payload).valid);-- return { event_id: ids.newId('event'), event_type: 'network.subject.merged', version: 1, source: 'network', actor: { type: 'user', id: into },-- timestamp: payload.merged_at, visibility: 'internal', subject: { type: 'user', id: into }, payload };--};----(async () => {-- await db.initDb();-- const networkEvents = require('../server/auth/network-events');-- const subjectMerge = require('../server/auth/subject-merge');-- const { NETWORK_TOPICS } = require('../scripts/subscribe-media-events');-- const d = db.getDb();-- // users.merged_into is added lazily by the old SQLite ensureSchema but is not in migrations/0002_live.sql yet;-- // the owner-only DDL helper stands it up here until the migration includes it.-- await globalThis.__ovLiveDdl('ALTER TABLE users ADD COLUMN IF NOT EXISTS merged_into bigint');-- await d.prepare(`INSERT INTO users (id, username, password_hash) OVERRIDING SYSTEM VALUE VALUES (10, 'keep', '$sso$'), (11, 'fold', '$sso$'), (12, 'xena', '$sso$'), (13, 'yuri', '$sso$'), (14, 'lone', '$sso$')`).run();-- await d.prepare(`INSERT INTO linked_accounts (user_id, service, service_user_id, service_username, subject_id) VALUES-- (10, 'network', '100', 'keep', ?), (11, 'network', '101', 'fold', ?), (12, 'network', '102', 'xena', ?), (14, 'network', '104', 'lone', ?)`).run(KEEP, FOLD, X, LONE);-- // Follows: fold follows xena (keep does too: a clash), fold follows yuri, yuri follows fold's channel, fold follows keep (self after merge).-- await d.prepare('INSERT INTO follows (follower_id, streamer_id) VALUES (11, 12), (10, 12), (11, 13), (13, 11), (11, 10)').run();-- // Channel points: fold in xena's channel (keep has some there: summed), yuri in fold's channel, fold in keep's channel.-- await d.prepare('INSERT INTO channel_points (user_id, streamer_id, balance) VALUES (11, 12, 30), (10, 12, 5), (13, 11, 7), (11, 10, 9)').run();-- await d.prepare("INSERT INTO streams (user_id, title) VALUES (11, 'old stream'), (11, 'another'), (10, 'mine')").run();---- const app = express();-- app.use(express.json({ verify: (req, res, buf) => { req.rawBody = buf; } }));-- app.post('/internal/network-events', networkEvents.handler);-- const server = http.createServer(app);-- await new Promise((r) => server.listen(0, '127.0.0.1', r));-- const url = `http://127.0.0.1:${server.address().port}/internal/network-events`;-- const send = async (ev) => { const raw = JSON.stringify({ event: ev, seq: 1 }); return (await fetch(url, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw, process.env.LIVE_EVENTS_SECRET) }, body: raw })).status; };-- try {-- assert.ok(NETWORK_TOPICS.includes('network.subject.merged'), 'the subscription asks for merges');---- const ev = envelope(FOLD, KEEP);-- assert.strictEqual(await send(ev), 204);-- const follows = (await d.prepare('SELECT follower_id AS a, streamer_id AS b FROM follows ORDER BY a, b').all()).map((f) => `${f.a}>${f.b}`);-- assert.deepStrictEqual(follows, ['10>12', '10>13', '13>10'], 'moved; the clash with keep>xena and fold>keep (now self) dropped');-- const points = (await d.prepare('SELECT user_id AS u, streamer_id AS s, balance FROM channel_points ORDER BY u, s').all()).map((p) => `${p.u}@${p.s}=${p.balance}`);-- assert.deepStrictEqual(points, ['10@10=9', '10@12=35', '13@10=7'], 'summed where both had points; the channel side moved too');-- const log = (await d.prepare("SELECT delta FROM channel_points_log WHERE reason = 'account_merge'").all()).map((l) => l.delta);-- assert.strictEqual(log.reduce((n, x) => n + x, 0), 0, 'every move logged out and in');-- assert.strictEqual(log.length, 6);-- assert.deepStrictEqual((await d.prepare('SELECT user_id FROM streams ORDER BY id').all()).map((s) => s.user_id), [10, 10, 10]);-- assert.strictEqual((await d.prepare('SELECT merged_into FROM users WHERE id = 11').get()).merged_into, 10);-- const rec = await d.prepare('SELECT * FROM subject_merges WHERE merge_id = ?').get(ev.payload.merge_id);-- assert.deepStrictEqual([rec.from_user_id, rec.into_user_id, JSON.parse(rec.outcome).result], [11, 10, 'merged']);---- // A redelivery changes nothing.-- assert.strictEqual(await send(ev), 204);-- assert.strictEqual(await subjectMerge.apply(ev), 'unchanged');-- assert.strictEqual((await d.prepare("SELECT COUNT(*) AS n FROM channel_points_log WHERE reason = 'account_merge'").get()).n, 6);---- // Only the folded-in account had a Live user: it becomes the survivor's.-- const r = await subjectMerge.apply(envelope(LONE, LONE_INTO), { resolveNetworkId: async (s) => (s === LONE_INTO ? 205 : null) });-- assert.strictEqual(r, 'relinked');-- assert.deepStrictEqual(await d.prepare("SELECT subject_id, service_user_id FROM linked_accounts WHERE user_id = 14 AND service = 'network'").get(), { subject_id: LONE_INTO, service_user_id: '205' });---- // Nobody on Live: acknowledged, recorded.-- assert.strictEqual(await subjectMerge.apply(envelope(ids.newId('user'), ids.newId('user'))), 'nothing');-- // Not Network's, or a malformed payload: ignored.-- assert.strictEqual(await networkEvents.apply({ ...envelope(FOLD, KEEP), source: 'media' }), 'ignored:source');-- assert.strictEqual(await subjectMerge.apply({ payload: { merge_id: 'x', from: FOLD, into: KEEP } }), 'ignored:payload');-- } finally {-- server.close();-- }-- console.log('subject merge: all checks passed');--})().catch((e) => { console.error(e); process.exit(1); });
@@ -245,7 +245,7 @@ through the SDK inbox (`idempotency_receipts`):
@@ -245,7 +245,7 @@ through the SDK inbox (`idempotency_receipts`):
|---|---|---|---|
|---|---|---|---|
| `openre.session.*` | `/internal/openre-events` | `OPENRE_EVENTS_SECRET` | OpenRestream's `scripts/subscribe-live-events.js` |
| `openre.session.*` | `/internal/openre-events` | `OPENRE_EVENTS_SECRET` | OpenRestream's `scripts/subscribe-live-events.js` |
| `media.vod.*`, `media.clip.*`, `media.storage.*` | `/internal/media-events` | `MEDIA_EVENTS_SECRET` | `scripts/subscribe-media-events.js` |
| `media.vod.*`, `media.clip.*`, `media.storage.*` | `/internal/media-events` | `MEDIA_EVENTS_SECRET` | `scripts/subscribe-media-events.js` |
-| `network.user.*`, `network.follow.*`, `network.subject.merged`, `network.account.*` | `/internal/network-events` | `LIVE_EVENTS_SECRET` | `scripts/subscribe-media-events.js --network` |
+| `network.user.*`, `network.follow.*`, `network.account.*` | `/internal/network-events` | `LIVE_EVENTS_SECRET` | `scripts/subscribe-media-events.js --network` |
| `billing.transaction.settled`, `billing.entitlement.changed` | `/internal/billing-events` | `LIVE_EVENTS_SECRET` | `scripts/subscribe-media-events.js --billing` |
| `billing.transaction.settled`, `billing.entitlement.changed` | `/internal/billing-events` | `LIVE_EVENTS_SECRET` | `scripts/subscribe-media-events.js --billing` |
| `tips.interaction.ready` | `/internal/tips-events` | `LIVE_EVENTS_SECRET` | `scripts/subscribe-media-events.js --tips` |
| `tips.interaction.ready` | `/internal/tips-events` | `LIVE_EVENTS_SECRET` | `scripts/subscribe-media-events.js --tips` |
@@ -36,7 +36,7 @@ const fs = require('fs');
@@ -36,7 +36,7 @@ const fs = require('fs');
const TOPICS = ['media.vod.*', 'media.clip.*', 'media.storage.*', 'media.object.*'];
const TOPICS = ['media.vod.*', 'media.clip.*', 'media.storage.*', 'media.object.*'];
const DEFAULT_ENDPOINT = 'http://127.0.0.1:3000/internal/media-events';
const DEFAULT_ENDPOINT = 'http://127.0.0.1:3000/internal/media-events';
-const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted', 'network.subject.merged', 'network.account.export_requested', 'network.account.deleted'];
+const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted', 'network.account.export_requested', 'network.account.deleted'];
const NETWORK_ENDPOINT = 'http://127.0.0.1:3000/internal/network-events';
const NETWORK_ENDPOINT = 'http://127.0.0.1:3000/internal/network-events';
const BILLING_TOPICS = ['billing.transaction.settled', 'billing.entitlement.changed'];
const BILLING_TOPICS = ['billing.transaction.settled', 'billing.entitlement.changed'];
const BILLING_ENDPOINT = 'http://127.0.0.1:3000/internal/billing-events';
const BILLING_ENDPOINT = 'http://127.0.0.1:3000/internal/billing-events';
@@ -10,8 +10,6 @@
@@ -10,8 +10,6 @@
* token_valid_after the person's older tokens are refused (./revocations.js)
* token_valid_after the person's older tokens are refused (./revocations.js)
* updated the person's profile, role and ban (./subject-projection.js, WS-B task 2)
* updated the person's profile, role and ban (./subject-projection.js, WS-B task 2)
* follow.created/.deleted Live's follows table as a projection of Network's graph (../social/network-follows.js)
* follow.created/.deleted Live's follows table as a projection of Network's graph (../social/network-follows.js)
- * subject.merged two accounts became one (ADR-029): the folded-in Live user's follows, channel points and streams
- * move to the survivor's (./subject-merge.js), once per merge
* account.export_requested / account.deleted the person's export part goes to Network, or their Live rows are
* account.export_requested / account.deleted the person's export part goes to Network, or their Live rows are
* erased and the erasure confirmed (./account-data.js, ADR-033), once per export or deletion
* erased and the erasure confirmed (./account-data.js, ADR-033), once per export or deletion
*
*
@@ -32,15 +30,13 @@ const notifyChat = async (userId) => {
@@ -32,15 +30,13 @@ const notifyChat = async (userId) => {
function secret() { return process.env.LIVE_EVENTS_SECRET || process.env.MEDIA_EVENTS_SECRET || ''; }
function secret() { return process.env.LIVE_EVENTS_SECRET || process.env.MEDIA_EVENTS_SECRET || ''; }
-/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>' (a merge, export or deletion: a promise). */
+/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>' (an account event: a promise). */
async function apply(ev) {
async function apply(ev) {
if (!ev || !EVENT_ID_RE.test(String(ev.event_id || ''))) return 'ignored:envelope';
if (!ev || !EVENT_ID_RE.test(String(ev.event_id || ''))) return 'ignored:envelope';
- if (ev.event_type === 'network.subject.merged') return ev.source === 'network' ? await require('./subject-merge').apply(ev) : 'ignored:source';
if (ev.event_type === 'network.account.export_requested' || ev.event_type === 'network.account.deleted') return ev.source === 'network' ? await require('./account-data').apply(ev) : 'ignored:source';
if (ev.event_type === 'network.account.export_requested' || ev.event_type === 'network.account.deleted') return ev.source === 'network' ? await require('./account-data').apply(ev) : 'ignored:source';
const follow = ev.event_type === 'network.follow.created' || ev.event_type === 'network.follow.deleted';
const follow = ev.event_type === 'network.follow.created' || ev.event_type === 'network.follow.deleted';
if (ev.event_type !== 'network.user.token_valid_after' && ev.event_type !== 'network.user.updated' && !follow) return 'ignored:type';
if (ev.event_type !== 'network.user.token_valid_after' && ev.event_type !== 'network.user.updated' && !follow) return 'ignored:type';
if (ev.source !== 'network') return 'ignored:source';
if (ev.source !== 'network') return 'ignored:source';
- // The follow graph is Network's (ADR-030); Live's follows table is its projection.
if (follow) return await require('../social/network-follows').apply(ev);
if (follow) return await require('../social/network-follows').apply(ev);
const p = ev.payload && typeof ev.payload === 'object' ? ev.payload : {};
const p = ev.payload && typeof ev.payload === 'object' ? ev.payload : {};
if (ev.event_type === 'network.user.updated') return await require('./subject-projection').apply(p, { notify: notifyChat });
if (ev.event_type === 'network.user.updated') return await require('./subject-projection').apply(p, { notify: notifyChat });
@@ -58,10 +54,10 @@ async function handler(req, res) {
@@ -58,10 +54,10 @@ async function handler(req, res) {
const delivery = parseDelivery(req.rawBody, req.headers, key, { requireV2: true });
const delivery = parseDelivery(req.rawBody, req.headers, key, { requireV2: true });
if (!delivery) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }
if (!delivery) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }
stats.received++;
stats.received++;
- const count = (out) => { if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated' || out === 'followed' || out === 'unfollowed' || out === 'merged' || out === 'relinked' || out === 'exported' || out === 'erased' || out === 'confirmed') stats.profiles++; else stats.ignored++; };
+ const count = (out) => { if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated' || out === 'followed' || out === 'unfollowed' || out === 'relinked' || out === 'exported' || out === 'erased' || out === 'confirmed') stats.profiles++; else stats.ignored++; };
const out = await apply(delivery.event);
const out = await apply(delivery.event);
if (out && typeof out.then === 'function') {
if (out && typeof out.then === 'function') {
- // A merge, export or deletion: answer after it applied, so a failure is redelivered.
+ // Account export or deletion: answer after it applied, so a failure is redelivered.
return out.then((o) => { count(o); res.status(204).end(); }, (e) => { console.error(`[NetworkEvents] ${delivery.event.event_type} failed:`, e.message); res.status(500).json({ error: 'not applied' }); });
return out.then((o) => { count(o); res.status(204).end(); }, (e) => { console.error(`[NetworkEvents] ${delivery.event.event_type} failed:`, e.message); res.status(500).json({ error: 'not applied' }); });
}
}
count(out);
count(out);
deleted file mode 100644
deleted file mode 100644
@@ -1,106 +0,0 @@
@@ -1,106 +0,0 @@
-'use strict';
-/**
- * network.subject.merged → Live (roadmap WS-B task 5, ADR-029; Contracts 0.69.0). Two Network accounts became one:
- * `from` is now an alias of `into`. Live has at most one user linked to each (linked_accounts, service 'network'):
- *
- * neither, or only the survivor's nothing of Live's to move
- * only the folded-in account's that Live user becomes the survivor's: its network link names `into` (and the
- * survivor's Network user id, resolved from Network)
- * both in one transaction, the folded-in user's follows (as follower and as the followed
- * channel), channel points (as viewer and as the channel; balances of the same pair
- * are summed, each move logged in channel_points_log once) and streams move to the
- * survivor's user. A follow the survivor already has, or following oneself, is
- * dropped and counted. The folded-in Live user stays, marked merged_into.
- *
- * Live's follows are a projection of Network's graph (ADR-030), which Network already merged; this keeps the
- * projection in step without waiting for a rebuild. VODs and clips are Media's (it follows the same event), chat is
- * Chat's. Applied once per merge_id (subject_merges).
- */
-const db = require('../db/database');
-
-const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
-const MERGE_RE = /^mrg_[0-9A-HJKMNP-TV-Z]{26}$/;
-
-const liveUserOf = async (d, subject) => {
- const r = await d.prepare("SELECT user_id FROM linked_accounts WHERE service = 'network' AND subject_id = ? ORDER BY id LIMIT 1").get(subject);
- return r ? r.user_id : null;
-};
-
-/** Move everything of Live user `a` to Live user `b` (one transaction). → counts. */
-async function moveUser(d, a, b, mergeId) {
- const out = { follows: 0, followers: 0, follows_dropped: 0, points: 0, points_summed: 0, streams: 0 };
- // Follows: as the follower, then as the followed channel.
- for (const [col, other, key] of [['follower_id', 'streamer_id', 'follows'], ['streamer_id', 'follower_id', 'followers']]) {
- for (const f of await d.prepare(`SELECT id, ${other} AS o FROM follows WHERE ${col} = ?`).all(a)) {
- const clash = f.o === b || await d.prepare(`SELECT 1 FROM follows WHERE ${col} = ? AND ${other} = ?`).get(b, f.o);
- if (clash) { await d.prepare('DELETE FROM follows WHERE id = ?').run(f.id); out.follows_dropped++; }
- else { await d.prepare(`UPDATE follows SET ${col} = ? WHERE id = ?`).run(b, f.id); out[key]++; }
- }
- }
- // Channel points (loyalty, ADR-012): first as the viewer, then as the channel; the same pair twice is summed.
- const logMove = d.prepare('INSERT INTO channel_points_log (idempotency_key, user_id, streamer_id, delta, reason) VALUES (?, ?, ?, ?, ?) ON CONFLICT DO NOTHING');
- const add = async (u, s, bal) => {
- const have = await d.prepare('SELECT 1 FROM channel_points WHERE user_id = ? AND streamer_id = ?').get(u, s);
- if (have) { await d.prepare('UPDATE channel_points SET balance = balance + ?, updated_at = ov_now() WHERE user_id = ? AND streamer_id = ?').run(bal, u, s); out.points_summed++; }
- else await d.prepare('INSERT INTO channel_points (user_id, streamer_id, balance) VALUES (?, ?, ?)').run(u, s, bal);
- };
- const move = async (fromU, fromS, toU, toS, bal, key) => {
- await d.prepare('DELETE FROM channel_points WHERE user_id = ? AND streamer_id = ?').run(fromU, fromS);
- await add(toU, toS, bal);
- if (bal) {
- await logMove.run(`live:cp:merge:${mergeId}:${key}:out`, fromU, fromS, -bal, 'account_merge');
- await logMove.run(`live:cp:merge:${mergeId}:${key}:in`, toU, toS, bal, 'account_merge');
- }
- out.points++;
- };
- for (const p of await d.prepare('SELECT streamer_id AS s, balance FROM channel_points WHERE user_id = ?').all(a)) await move(a, p.s, b, p.s === a ? b : p.s, p.balance, `v:${p.s}`);
- for (const p of await d.prepare('SELECT user_id AS u, balance FROM channel_points WHERE streamer_id = ?').all(a)) await move(p.u, a, p.u, b, p.balance, `c:${p.u}`);
- out.streams = (await d.prepare('UPDATE streams SET user_id = ? WHERE user_id = ?').run(b, a)).changes;
- await d.prepare('UPDATE users SET merged_into = ? WHERE id = ?').run(b, a);
- return out;
-}
-
-/**
- * Apply one network.subject.merged envelope → 'merged' | 'relinked' | 'nothing' | 'unchanged' | 'ignored:<why>'.
- * resolveNetworkId(subject) → the survivor's Network user id (injected by tests; defaults to Network's resolve-batch).
- */
-async function apply(ev, { resolveNetworkId = defaultResolve } = {}) {
- const p = ev && ev.payload && typeof ev.payload === 'object' ? ev.payload : {};
- if (!MERGE_RE.test(String(p.merge_id || '')) || !SUBJECT_RE.test(String(p.from || '')) || !SUBJECT_RE.test(String(p.into || '')) || p.from === p.into) return 'ignored:payload';
- const d = db.getDb();
- if (await d.prepare('SELECT 1 FROM subject_merges WHERE merge_id = ?').get(p.merge_id)) return 'unchanged';
- const a = await liveUserOf(d, p.from);
- const b = await liveUserOf(d, p.into);
- let result; let outcome;
- if (!a) { result = 'nothing'; outcome = { reason: 'no Live user for the folded-in account' }; }
- else if (!b) {
- const networkId = await resolveNetworkId(p.into);
- try {
- await d.prepare("UPDATE linked_accounts SET subject_id = ?, service_user_id = COALESCE(?::text, service_user_id) WHERE service = 'network' AND user_id = ?")
- .run(p.into, networkId == null ? null : String(networkId), a);
- result = 'relinked'; outcome = { user: a, network_user_id: networkId == null ? null : Number(networkId) };
- } catch (e) {
- // Another Live user already carries the survivor's Network id without its subject: left for staff, logged.
- result = 'conflict'; outcome = { user: a, error: String(e.message).slice(0, 200) };
- }
- } else if (a === b) { result = 'nothing'; outcome = { reason: 'one Live user for both' }; }
- else {
- outcome = await d.tx(async () => await moveUser(d, a, b, p.merge_id));
- result = 'merged';
- }
- await d.prepare('INSERT INTO subject_merges (merge_id, from_subject, into_subject, from_user_id, into_user_id, outcome) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING')
- .run(p.merge_id, p.from, p.into, a, b, JSON.stringify({ result, ...outcome }));
- console.log(`[Merge] ${p.merge_id}: ${result} ${JSON.stringify(outcome)}`);
- return result;
-}
-
-async function defaultResolve(subject) {
- try {
- const { networkPost } = require('./identity-sync');
- const r = await networkPost('/internal/identity/resolve-batch', { subject_ids: [subject] });
- const hit = r && r.results && r.results[subject];
- return hit && hit.network_user_id != null ? hit.network_user_id : null;
- } catch { return null; }
-}
-
-module.exports = { apply, moveUser };
@@ -51,7 +51,6 @@ const KINDS = Object.freeze([
@@ -51,7 +51,6 @@ const KINDS = Object.freeze([
Object.freeze({ id: 'media-request', match: /^Media request: (.+)$/, site: 'live', subject: 'media', text: 'for the stream {name}' }),
Object.freeze({ id: 'media-request', match: /^Media request: (.+)$/, site: 'live', subject: 'media', text: 'for the stream {name}' }),
Object.freeze({ id: 'media-refund', match: /^Refund: (.+)$/, site: 'live', subject: 'media', text: 'for a refund of {name}' }),
Object.freeze({ id: 'media-refund', match: /^Refund: (.+)$/, site: 'live', subject: 'media', text: 'for a refund of {name}' }),
Object.freeze({ id: 'migration', match: 'migration:hobo-coins', site: 'network', text: 'for OpenCoins moved from the old Live wallet' }),
Object.freeze({ id: 'migration', match: 'migration:hobo-coins', site: 'network', text: 'for OpenCoins moved from the old Live wallet' }),
- Object.freeze({ id: 'account-merge', match: 'account_merge', site: 'network', text: 'for merging your accounts' }),
Object.freeze({ id: 'account-deleted', match: 'account_deleted', site: 'network', text: 'for closing your account' }),
Object.freeze({ id: 'account-deleted', match: 'account_deleted', site: 'network', text: 'for closing your account' }),
Object.freeze({ id: 'watch', match: 'watch', site: 'live', text: 'for watching a stream' }),
Object.freeze({ id: 'watch', match: 'watch', site: 'live', text: 'for watching a stream' }),
Object.freeze({ id: 'chat', match: 'chat', site: 'live', text: 'for chatting on a stream' }),
Object.freeze({ id: 'chat', match: 'chat', site: 'live', text: 'for chatting on a stream' }),
@@ -227,7 +227,7 @@ server.listen(0, '127.0.0.1', async () => {
@@ -227,7 +227,7 @@ server.listen(0, '127.0.0.1', async () => {
// Live subscribes to the follow events.
// Live subscribes to the follow events.
assert.match(fs.readFileSync(path.join(__dirname, '..', 'scripts', 'subscribe-media-events.js'), 'utf8'), /'network\.follow\.created', 'network\.follow\.deleted'/);
assert.match(fs.readFileSync(path.join(__dirname, '..', 'scripts', 'subscribe-media-events.js'), 'utf8'), /'network\.follow\.created', 'network\.follow\.deleted'/);
- // Only the projection writes follows (and the account merge repoints it); no Live-only path is left.
+ // Only the projection writes follows (and Network projects it); no Live-only path is left.
const root = path.join(__dirname, '..', 'server');
const root = path.join(__dirname, '..', 'server');
const writers = [];
const writers = [];
const walk = (dir) => { for (const e of fs.readdirSync(dir, { withFileTypes: true })) {
const walk = (dir) => { for (const e of fs.readdirSync(dir, { withFileTypes: true })) {
@@ -236,7 +236,7 @@ server.listen(0, '127.0.0.1', async () => {
@@ -236,7 +236,7 @@ server.listen(0, '127.0.0.1', async () => {
else if (e.name.endsWith('.js') && /\b(INSERT(\s+OR\s+\w+)?\s+INTO|DELETE\s+FROM|UPDATE)\s+follows\b/i.test(fs.readFileSync(p, 'utf8'))) writers.push(path.relative(root, p));
else if (e.name.endsWith('.js') && /\b(INSERT(\s+OR\s+\w+)?\s+INTO|DELETE\s+FROM|UPDATE)\s+follows\b/i.test(fs.readFileSync(p, 'utf8'))) writers.push(path.relative(root, p));
} };
} };
walk(root);
walk(root);
- assert.deepStrictEqual(writers.sort(), ['auth/subject-merge.js', 'social/network-follows.js'], 'follows writers');
+ assert.deepStrictEqual(writers.sort(), ['social/network-follows.js'], 'follows writers');
assert.strictEqual(typeof db.followUser, 'undefined');
assert.strictEqual(typeof db.followUser, 'undefined');
assert.strictEqual(typeof db.unfollowUser, 'undefined');
assert.strictEqual(typeof db.unfollowUser, 'undefined');
assert.ok(!/FOLLOWS_AUTHORITY/.test(fs.readFileSync(path.join(root, 'social', 'network-follows.js'), 'utf8')), 'no switch back to Live-only follows');
assert.ok(!/FOLLOWS_AUTHORITY/.test(fs.readFileSync(path.join(root, 'social', 'network-follows.js'), 'utf8')), 'no switch back to Live-only follows');
deleted file mode 100644
deleted file mode 100644
@@ -1,92 +0,0 @@
@@ -1,92 +0,0 @@
-'use strict';
-// network.subject.merged → Live (roadmap WS-B task 5, ADR-029): through POST /internal/network-events (signed), the
-// folded-in Live user's follows (both ways), channel points (both ways, a shared pair summed, every move logged once)
-// and streams move to the survivor's user; a follow it already has and following oneself are dropped; the folded-in
-// user is marked merged_into. With only the folded-in account on Live, that user is relinked to the survivor. A
-// redelivery changes nothing; a merge nobody on Live had is acknowledged. The subscription asks for the topic.
-// node test/subject-merge.test.js
-const assert = require('assert');
-const http = require('http');
-
-process.env.NODE_ENV = 'test';
-process.env.LIVE_EVENTS_SECRET = 's'.repeat(40);
-const quiet = console.log;
-console.log = (...a) => { if (!/^\[/.test(String(a[0]))) quiet(...a); };
-console.warn = () => {};
-
-const express = require('express');
-const { ids, validate } = require('openvibe-contracts');
-const { signDeliveryHeaders } = require('openvibe-sdk/events');
-const db = require('../server/db/database');
-
-const KEEP = ids.newId('user'), FOLD = ids.newId('user'), X = ids.newId('user'), LONE = ids.newId('user'), LONE_INTO = ids.newId('user');
-
-const envelope = (from, into, mergeId = `mrg_${ids.ulid()}`) => {
- const payload = { merge_id: mergeId, from, into, merged_at: new Date().toISOString(), initiated_by: 'person', split_until: new Date(Date.now() + 30 * 86400000).toISOString() };
- assert.ok(validate('network.subject.merged@1', payload).valid);
- return { event_id: ids.newId('event'), event_type: 'network.subject.merged', version: 1, source: 'network', actor: { type: 'user', id: into },
- timestamp: payload.merged_at, visibility: 'internal', subject: { type: 'user', id: into }, payload };
-};
-
-(async () => {
- await db.initDb();
- const networkEvents = require('../server/auth/network-events');
- const subjectMerge = require('../server/auth/subject-merge');
- const { NETWORK_TOPICS } = require('../scripts/subscribe-media-events');
- const d = db.getDb();
- // users.merged_into is added lazily by the old SQLite ensureSchema but is not in migrations/0002_live.sql yet;
- // the owner-only DDL helper stands it up here until the migration includes it.
- await globalThis.__ovLiveDdl('ALTER TABLE users ADD COLUMN IF NOT EXISTS merged_into bigint');
- await d.prepare(`INSERT INTO users (id, username, password_hash) OVERRIDING SYSTEM VALUE VALUES (10, 'keep', '$sso$'), (11, 'fold', '$sso$'), (12, 'xena', '$sso$'), (13, 'yuri', '$sso$'), (14, 'lone', '$sso$')`).run();
- await d.prepare(`INSERT INTO linked_accounts (user_id, service, service_user_id, service_username, subject_id) VALUES
- (10, 'network', '100', 'keep', ?), (11, 'network', '101', 'fold', ?), (12, 'network', '102', 'xena', ?), (14, 'network', '104', 'lone', ?)`).run(KEEP, FOLD, X, LONE);
- // Follows: fold follows xena (keep does too: a clash), fold follows yuri, yuri follows fold's channel, fold follows keep (self after merge).
- await d.prepare('INSERT INTO follows (follower_id, streamer_id) VALUES (11, 12), (10, 12), (11, 13), (13, 11), (11, 10)').run();
- // Channel points: fold in xena's channel (keep has some there: summed), yuri in fold's channel, fold in keep's channel.
- await d.prepare('INSERT INTO channel_points (user_id, streamer_id, balance) VALUES (11, 12, 30), (10, 12, 5), (13, 11, 7), (11, 10, 9)').run();
- await d.prepare("INSERT INTO streams (user_id, title) VALUES (11, 'old stream'), (11, 'another'), (10, 'mine')").run();
-
- const app = express();
- app.use(express.json({ verify: (req, res, buf) => { req.rawBody = buf; } }));
- app.post('/internal/network-events', networkEvents.handler);
- const server = http.createServer(app);
- await new Promise((r) => server.listen(0, '127.0.0.1', r));
- const url = `http://127.0.0.1:${server.address().port}/internal/network-events`;
- const send = async (ev) => { const raw = JSON.stringify({ event: ev, seq: 1 }); return (await fetch(url, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw, process.env.LIVE_EVENTS_SECRET) }, body: raw })).status; };
- try {
- assert.ok(NETWORK_TOPICS.includes('network.subject.merged'), 'the subscription asks for merges');
-
- const ev = envelope(FOLD, KEEP);
- assert.strictEqual(await send(ev), 204);
- const follows = (await d.prepare('SELECT follower_id AS a, streamer_id AS b FROM follows ORDER BY a, b').all()).map((f) => `${f.a}>${f.b}`);
- assert.deepStrictEqual(follows, ['10>12', '10>13', '13>10'], 'moved; the clash with keep>xena and fold>keep (now self) dropped');
- const points = (await d.prepare('SELECT user_id AS u, streamer_id AS s, balance FROM channel_points ORDER BY u, s').all()).map((p) => `${p.u}@${p.s}=${p.balance}`);
- assert.deepStrictEqual(points, ['10@10=9', '10@12=35', '13@10=7'], 'summed where both had points; the channel side moved too');
- const log = (await d.prepare("SELECT delta FROM channel_points_log WHERE reason = 'account_merge'").all()).map((l) => l.delta);
- assert.strictEqual(log.reduce((n, x) => n + x, 0), 0, 'every move logged out and in');
- assert.strictEqual(log.length, 6);
- assert.deepStrictEqual((await d.prepare('SELECT user_id FROM streams ORDER BY id').all()).map((s) => s.user_id), [10, 10, 10]);
- assert.strictEqual((await d.prepare('SELECT merged_into FROM users WHERE id = 11').get()).merged_into, 10);
- const rec = await d.prepare('SELECT * FROM subject_merges WHERE merge_id = ?').get(ev.payload.merge_id);
- assert.deepStrictEqual([rec.from_user_id, rec.into_user_id, JSON.parse(rec.outcome).result], [11, 10, 'merged']);
-
- // A redelivery changes nothing.
- assert.strictEqual(await send(ev), 204);
- assert.strictEqual(await subjectMerge.apply(ev), 'unchanged');
- assert.strictEqual((await d.prepare("SELECT COUNT(*) AS n FROM channel_points_log WHERE reason = 'account_merge'").get()).n, 6);
-
- // Only the folded-in account had a Live user: it becomes the survivor's.
- const r = await subjectMerge.apply(envelope(LONE, LONE_INTO), { resolveNetworkId: async (s) => (s === LONE_INTO ? 205 : null) });
- assert.strictEqual(r, 'relinked');
- assert.deepStrictEqual(await d.prepare("SELECT subject_id, service_user_id FROM linked_accounts WHERE user_id = 14 AND service = 'network'").get(), { subject_id: LONE_INTO, service_user_id: '205' });
-
- // Nobody on Live: acknowledged, recorded.
- assert.strictEqual(await subjectMerge.apply(envelope(ids.newId('user'), ids.newId('user'))), 'nothing');
- // Not Network's, or a malformed payload: ignored.
- assert.strictEqual(await networkEvents.apply({ ...envelope(FOLD, KEEP), source: 'media' }), 'ignored:source');
- assert.strictEqual(await subjectMerge.apply({ payload: { merge_id: 'x', from: FOLD, into: KEEP } }), 'ignored:payload');
- } finally {
- server.close();
- }
- console.log('subject merge: all checks passed');
-})().catch((e) => { console.error(e); process.exit(1); });