openvibers/openvibe.inventory Commits
Inventory consumes network.subject.merged (ADR-029): the folded-in account's items, equipped set and Workshop credit move to the survivor once per merge, the oldest of a duplicate unique item is kept, the survivor's equipped set wins and the ledger gains a merged row per instance. (#63)
Changes
@@@ -4,6 +4,10 @@ What changed in OpenVibe.Inventory, newest first. Each site also publishes its p## Unreleased++- Account merge (ADR-029, ADR-054 §7): Inventory consumes `network.subject.merged`. The folded-in account's items,++ equipped set and Workshop credit become the survivor's in one transaction, once per merge id++ (`inventory_subject_merges`). Where both own the same non-stackable item the oldest is kept and the other is revoked;++ the survivor's equipped set wins; the ledger keeps the old rows and gains one `merged` row per instance.- Workshop images reach OpenVibe.Media with Inventory's Network service token (audience `openvibe.media`, grant`media.object.upload` on namespace `inventory`). `INVENTORY_MEDIA_APP_KEY` is gone (compatibility register C-71).nnew file mode 100644@@@ -0,0 +1,13 @@++-- phase: expand++-- Account merges (ADR-029, network.subject.merged; ADR-054 §7): one row per merge Inventory has folded, so a redelivery++-- moves nothing twice (server/inventory/subject-merge.js). The counts say what became the survivor's and what was dropped.++CREATE TABLE inventory_subject_merges (++ merge_id text COLLATE "C" PRIMARY KEY, -- mrg_<ULID>++ from_subject text COLLATE "C" NOT NULL, -- the folded-in account (now an alias of into_subject)++ into_subject text COLLATE "C" NOT NULL, -- the survivor++ instances integer NOT NULL DEFAULT 0, -- instances that moved to the survivor++ duplicates integer NOT NULL DEFAULT 0, -- duplicates of a non-stackable item, revoked (the oldest is kept)++ equipped integer NOT NULL DEFAULT 0, -- equipped slots that moved (the survivor had none there)++ equipped_dropped integer NOT NULL DEFAULT 0, -- equipped slots dropped (the survivor's set wins)++ merged_at text COLLATE "C" NOT NULL -- ISO 8601 UTC++);@@@ -40,6 +40,7 @@ const { createNetworkSender } = require('openvibe-sdk/account-data');const { createInventory } = require('./inventory/store');const { seed } = require('./inventory/seed');const accountDataLib = require('./inventory/account-data');++const subjectMerge = require('./inventory/subject-merge');const { assetVersion, send } = require('./render/layout');const { html } = require('./render/html');@@@ -138,7 +139,9 @@ async function createApp(opts = {}) {// ── OpenVibe.Events → Inventory (loopback only: nginx answers 404 for /internal/) ──// network.account.export_requested and network.account.deleted, answered by openvibe-sdk/account-data's consumer,// which reads the raw body itself (the v2 signature covers it) and refuses a request that came through a proxy.-- app.post('/internal/events', accountData.consumer({ secrets: config.events.secrets, send: accountSend, log }));++ // network.subject.merged (ADR-029) is handed to ./inventory/subject-merge.js through onEvent.++ app.post('/internal/events', accountData.consumer({ secrets: config.events.secrets, send: accountSend, log,++ onEvent: (ev) => (ev.event_type === subjectMerge.TOPIC ? subjectMerge.apply(ev, { s, log }) : 'ignored:type') }));// ── Machine endpoints ───────────────────────────────────app.get('/api/health', (_req, res, next) => outbox.status().then((events) => res.json({ status: 'ok', service: 'openvibe-inventory', version: VERSION, events }), next));@@@ -7,7 +7,8 @@const { createApp } = require('./app');const { gracefulStop } = require('openvibe-sdk/service');const { startTelemetryRetention } = require('openvibe-sdk/telemetry');--const { startSubscriptions } = require('openvibe-sdk/account-data');++const { startSubscriptions, TOPICS: ACCOUNT_TOPICS } = require('openvibe-sdk/account-data');++const subjectMerge = require('./inventory/subject-merge');/*** The process stop (openvibe-sdk/service): the HTTP drain runs, then the timers clear, the Events subscriptions and the@@@ -42,6 +43,7 @@ async function start() {// INVENTORY_EVENTS_SECRET or the client secret.const subscriptions = startSubscriptions({eventsUrl: config.events.url, endpoint: `http://127.0.0.1:${config.port}/internal/events`, secret: config.events.secrets[0],++ topics: [...ACCOUNT_TOPICS, subjectMerge.TOPIC],networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,});// Sent events older than a week go (openvibe-sdk outbox prune), every six hours.nnew file mode 100644@@@ -0,0 +1,78 @@++'use strict';++++/**++ * network.subject.merged → Inventory (ADR-029, ADR-054 §7). Two Network accounts became one: `from` is now an alias of++ * `into`, and every item of `from` becomes `into`'s, in one transaction, once per merge_id (inventory_subject_merges):++ *++ * instances every instance moves to the survivor. Where both own the same item of a non-stackable kind, the++ * oldest is kept and the other is revoked (reason account_merge), so nobody holds two of a unique item++ * equipped the survivor's equipped set wins: a slot only `from` filled moves, a slot both filled keeps the++ * survivor's; a slot pointing at a revoked duplicate now points at the kept one++ * Workshop items `from` made are credited to, and managed by, the survivor++ * ledger append-only: `from`'s rows stay as written; each moved or revoked instance gets a `merged` row++ *++ * Delivered to POST /internal/events beside the account topics (openvibe-sdk/account-data's consumer, onEvent).++ */++const TOPIC = 'network.subject.merged';++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const MERGE_RE = /^mrg_[0-9A-HJKMNP-TV-Z]{26}$/;++const ACTOR = 'service:network';++++/** One envelope → 'merged' | 'unchanged' | 'ignored:<why>'. */++async function apply(ev, { s, log = console }) {++ if (!ev || ev.event_type !== TOPIC) return 'ignored:type';++ if (ev.source !== 'network') return 'ignored:source';++ const p = ev.payload && typeof ev.payload === 'object' ? ev.payload : {};++ const mergeId = String(p.merge_id || ''), from = String(p.from || ''), into = String(p.into || '');++ if (!MERGE_RE.test(mergeId) || !SUBJECT_RE.test(from) || !SUBJECT_RE.test(into) || from === into) return 'ignored:payload';++ const r = await s.tx(async () => {++ const db = s.db;++ if (await db.maybe('SELECT 1 FROM inventory_subject_merges WHERE merge_id = $1', [mergeId])) return null;++ const at = s.iso();++ const detail = JSON.stringify({ merge_id: mergeId, from, into });++ // Both own the same non-stackable item: the oldest is kept, every other copy is revoked.++ const dupes = await db.many(`WITH owned AS (++ SELECT i.id, i.definition_id, i.owner_subject,++ row_number() OVER (PARTITION BY i.definition_id ORDER BY i.acquired_at, i.id) AS rn,++ first_value(i.id) OVER (PARTITION BY i.definition_id ORDER BY i.acquired_at, i.id) AS keeper,++ min(i.owner_subject) OVER (PARTITION BY i.definition_id) AS lo,++ max(i.owner_subject) OVER (PARTITION BY i.definition_id) AS hi++ FROM inventory_instances i++ JOIN inventory_definitions d ON d.id = i.definition_id++ JOIN inventory_kinds k ON k.id = d.kind++ WHERE i.owner_subject IN ($1, $2) AND i.state = 'owned' AND NOT k.stackable)++ SELECT id, definition_id, keeper FROM owned WHERE rn > 1 AND lo <> hi ORDER BY id`, [from, into]);++ for (const d of dupes) {++ await db.exec('UPDATE inventory_equipped SET instance_id = $2 WHERE instance_id = $1', [d.id, d.keeper]);++ await db.exec("UPDATE inventory_instances SET state = 'revoked', updated_at = $2 WHERE id = $1", [d.id, at]);++ }++ // The survivor's equipped set wins; a slot only `from` filled moves.++ const equippedDropped = Number(await db.exec(`DELETE FROM inventory_equipped f USING inventory_equipped k++ WHERE f.owner_subject = $1 AND k.owner_subject = $2 AND k.kind = f.kind AND k.slot = f.slot`, [from, into])) || 0;++ const equipped = Number(await db.exec('UPDATE inventory_equipped SET owner_subject = $2 WHERE owner_subject = $1', [from, into])) || 0;++ // Two equipped rows may now point at one kept instance (both wore their copy in different slots): fine, an++ // instance can only fill the slot its kind allows, and both rows are the survivor's.++ const moved = await db.many('UPDATE inventory_instances SET owner_subject = $2, updated_at = $3 WHERE owner_subject = $1 RETURNING id, definition_id', [from, into, at]);++ const dupeIds = new Set(dupes.map((d) => d.id));++ for (const m of moved) {++ await db.exec(`INSERT INTO inventory_ledger (at, action, owner_subject, instance_id, definition_id, actor, reason, detail)++ VALUES ($1, 'merged', $2, $3, $4, $5, $6, $7)`, [at, into, m.id, m.definition_id, ACTOR, dupeIds.has(m.id) ? 'account_merge_duplicate_revoked' : 'account_merge', detail]);++ }++ for (const d of dupes) {++ if (moved.some((m) => m.id === d.id)) continue;++ await db.exec(`INSERT INTO inventory_ledger (at, action, owner_subject, instance_id, definition_id, actor, reason, detail)++ VALUES ($1, 'merged', $2, $3, $4, $5, 'account_merge_duplicate_revoked', $6)`, [at, into, d.id, d.definition_id, ACTOR, detail]);++ }++ // Workshop: what `from` made is the survivor's to manage, with the survivor's credit.++ await db.exec('UPDATE inventory_definitions SET credit_subject = $2 WHERE credit_subject = $1', [from, into]);++ await db.exec('UPDATE inventory_definitions SET issuer = $2 WHERE issuer = $1', [`user:${from}`, `user:${into}`]);++ await db.exec(`INSERT INTO inventory_subject_merges (merge_id, from_subject, into_subject, instances, duplicates, equipped, equipped_dropped, merged_at)++ VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`, [mergeId, from, into, moved.length, dupes.length, equipped, equippedDropped, at]);++ return { instances: moved.length, duplicates: dupes.length, equipped, equippedDropped };++ });++ if (!r) return 'unchanged';++ log.log(`[Merge] ${mergeId}: ${r.instances} instance(s) and ${r.equipped} equipped slot(s) now the survivor's; ${r.duplicates} duplicate(s) revoked, ${r.equippedDropped} equipped slot(s) dropped`);++ return 'merged';++}++++module.exports = { apply, TOPIC };nnew file mode 100644@@@ -0,0 +1,102 @@++'use strict';++/**++ * ADR-029 / ADR-054 §7: network.subject.merged folds one account's items into the survivor's, through the signed++ * loopback route POST /internal/events. Every instance moves; where both own the same non-stackable item the oldest is++ * kept and the other is revoked; the survivor's equipped set wins; the ledger keeps the folded account's rows and gains++ * one `merged` row per instance; a redelivery (same merge_id) moves nothing.++ */++const assert = require('assert');++const { signDeliveryHeaders } = require('openvibe-sdk/events');++const { boot, check, done } = require('./helpers/boot');++++const SECRET = `whsec_${'fixture'.repeat(6)}`;++const MERGE = 'mrg_01JZ0000000000000000000MRG';++++(async () => {++ const t = await boot({ env: { INVENTORY_EVENTS_SECRET: SECRET } });++ const keep = t.network.addUser('keep');++ const gone = t.network.addUser('gone');++ const bob = t.network.addUser('bob');++ const db = t.ctx.s.db;++ const inv = t.ctx.inv;++ const count = async (sql, args) => Number(await db.value(sql, args));++ const deliver = async (event) => {++ const body = JSON.stringify({ event, seq: 1 });++ const res = await fetch(`${t.base}/internal/events`, { method: 'POST', body, headers: { 'Content-Type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });++ return { status: res.status, json: await res.json().catch(() => null) };++ };++ const merged = (id, payload = {}, source = 'network') => ({ event_id: id, event_type: 'network.subject.merged', source, version: 1, timestamp: new Date().toISOString(),++ payload: { merge_id: MERGE, from: gone.subject, into: keep.subject, merged_at: new Date().toISOString(), initiated_by: 'person', ...payload } });++ const own = async (who, alias, key, acquiredAt) => {++ const def = await inv.byAlias('service:live', alias);++ const r = await inv.grant('service:live', { definition_id: def.id, subject: who.subject, idempotency_key: `test-merge-${key}` });++ await db.exec('UPDATE inventory_instances SET acquired_at = $2 WHERE id = $1', [r.instance.id, acquiredAt]);++ return r.instance.id;++ };++ let goneCrown, keepCrown, goneFire, bobCrown, goneLedger;++++ try {++ await check('both accounts own a crown (the folded one older) and wear it; the folded one also wears a name effect', async () => {++ goneCrown = await own(gone, 'hat_crown', 'g1', '2026-01-01T00:00:00.000Z');++ goneFire = await own(gone, 'fx_fire', 'g2', '2026-02-01T00:00:00.000Z');++ keepCrown = await own(keep, 'hat_crown', 'k1', '2026-06-01T00:00:00.000Z');++ bobCrown = await own(bob, 'hat_crown', 'b1', '2025-01-01T00:00:00.000Z');++ await inv.equip(gone.subject, `user:${gone.subject}`, { kind: 'live.hat', slot: 'hat', instance_id: goneCrown });++ await inv.equip(gone.subject, `user:${gone.subject}`, { kind: 'live.name_effect', slot: 'name_effect', instance_id: goneFire });++ await inv.equip(keep.subject, `user:${keep.subject}`, { kind: 'live.hat', slot: 'hat', instance_id: keepCrown });++ goneLedger = await count('SELECT count(*) FROM inventory_ledger WHERE owner_subject = $1', [gone.subject]);++ assert.ok(goneLedger >= 4);++ });++++ await check('a payload that is not a merge, or not from Network, is acknowledged and ignored', async () => {++ const bad = await deliver(merged('evt_01JZ0000000000000000000M00', { from: keep.subject }));++ assert.deepStrictEqual([bad.status, bad.json.outcome], [200, 'ignored:payload']);++ const spoof = await deliver(merged('evt_01JZ0000000000000000000M09', {}, 'live'));++ assert.deepStrictEqual([spoof.status, spoof.json.outcome], [200, 'ignored:source']);++ assert.strictEqual(await count('SELECT count(*) FROM inventory_instances WHERE owner_subject = $1', [gone.subject]), 2);++ });++++ await check('the merge moves every item, keeps the oldest crown, and the survivor\'s equipped set wins', async () => {++ const r = await deliver(merged('evt_01JZ0000000000000000000M01'));++ assert.deepStrictEqual([r.status, r.json && r.json.outcome], [200, 'merged'], JSON.stringify(r.json));++ assert.strictEqual(await count('SELECT count(*) FROM inventory_instances WHERE owner_subject = $1', [gone.subject]), 0, 'no copy under the folded subject');++ assert.strictEqual(await count('SELECT count(*) FROM inventory_equipped WHERE owner_subject = $1', [gone.subject]), 0);++ const rows = await db.many('SELECT id, state FROM inventory_instances WHERE owner_subject = $1 ORDER BY id', [keep.subject]);++ const state = Object.fromEntries(rows.map((x) => [x.id, x.state]));++ assert.deepStrictEqual(state, { [goneCrown]: 'owned', [goneFire]: 'owned', [keepCrown]: 'revoked' }, 'both instances kept as records; the older crown is the one owned');++ const eq = await db.many('SELECT kind, slot, instance_id FROM inventory_equipped WHERE owner_subject = $1 ORDER BY kind', [keep.subject]);++ assert.deepStrictEqual(eq, [++ { kind: 'live.hat', slot: 'hat', instance_id: goneCrown }, // the survivor's slot, now on the crown kept++ { kind: 'live.name_effect', slot: 'name_effect', instance_id: goneFire }, // a slot only the folded account filled++ ]);++ const rec = await db.maybe('SELECT from_subject, into_subject, instances, duplicates, equipped, equipped_dropped FROM inventory_subject_merges WHERE merge_id = $1', [MERGE]);++ assert.deepStrictEqual(rec, { from_subject: gone.subject, into_subject: keep.subject, instances: 2, duplicates: 1, equipped: 1, equipped_dropped: 1 });++ });++++ await check('the ledger keeps the folded account\'s rows and gains one merged row per instance', async () => {++ assert.strictEqual(await count('SELECT count(*) FROM inventory_ledger WHERE owner_subject = $1', [gone.subject]), goneLedger, 'history is never rewritten');++ const lines = await db.many("SELECT instance_id, reason, actor, detail->>'merge_id' AS merge_id FROM inventory_ledger WHERE action = 'merged' ORDER BY instance_id");++ assert.strictEqual(lines.length, 3);++ assert.ok(lines.every((l) => l.merge_id === MERGE && l.actor === 'service:network'));++ assert.strictEqual(lines.find((l) => l.instance_id === keepCrown).reason, 'account_merge_duplicate_revoked');++ assert.strictEqual(lines.find((l) => l.instance_id === goneFire).reason, 'account_merge');++ });++++ await check('a redelivery, or the same merge under another event id, changes nothing; bob is untouched', async () => {++ const before = await count('SELECT count(*) FROM inventory_ledger');++ assert.strictEqual((await deliver(merged('evt_01JZ0000000000000000000M01'))).json.outcome, 'unchanged');++ assert.strictEqual((await deliver(merged('evt_01JZ0000000000000000000M02'))).json.outcome, 'unchanged');++ assert.strictEqual(await count('SELECT count(*) FROM inventory_ledger'), before);++ assert.strictEqual(await count("SELECT count(*) FROM inventory_instances WHERE owner_subject = $1 AND state = 'owned'", [keep.subject]), 2);++ assert.deepStrictEqual(await db.maybe('SELECT owner_subject, state FROM inventory_instances WHERE id = $1', [bobCrown]), { owner_subject: bob.subject, state: 'owned' });++ });++++ await check('the merge topic is subscribed beside the account topics', async () => {++ const src = require('fs').readFileSync(require('path').join(__dirname, '../server/index.js'), 'utf8');++ assert.ok(/topics: \[\.\.\.ACCOUNT_TOPICS, subjectMerge\.TOPIC\]/.test(src));++ });++ } finally {++ await t.close();++ }++ done();++})();
@@ -4,6 +4,10 @@ What changed in OpenVibe.Inventory, newest first. Each site also publishes its p
@@ -4,6 +4,10 @@ What changed in OpenVibe.Inventory, newest first. Each site also publishes its p
## Unreleased
## Unreleased
+- Account merge (ADR-029, ADR-054 §7): Inventory consumes `network.subject.merged`. The folded-in account's items,
+ equipped set and Workshop credit become the survivor's in one transaction, once per merge id
+ (`inventory_subject_merges`). Where both own the same non-stackable item the oldest is kept and the other is revoked;
+ the survivor's equipped set wins; the ledger keeps the old rows and gains one `merged` row per instance.
- Workshop images reach OpenVibe.Media with Inventory's Network service token (audience `openvibe.media`, grant
- Workshop images reach OpenVibe.Media with Inventory's Network service token (audience `openvibe.media`, grant
`media.object.upload` on namespace `inventory`). `INVENTORY_MEDIA_APP_KEY` is gone (compatibility register C-71).
`media.object.upload` on namespace `inventory`). `INVENTORY_MEDIA_APP_KEY` is gone (compatibility register C-71).
new file mode 100644
new file mode 100644
@@ -0,0 +1,13 @@
@@ -0,0 +1,13 @@
+-- phase: expand
+-- Account merges (ADR-029, network.subject.merged; ADR-054 §7): one row per merge Inventory has folded, so a redelivery
+-- moves nothing twice (server/inventory/subject-merge.js). The counts say what became the survivor's and what was dropped.
+CREATE TABLE inventory_subject_merges (
+ merge_id text COLLATE "C" PRIMARY KEY, -- mrg_<ULID>
+ from_subject text COLLATE "C" NOT NULL, -- the folded-in account (now an alias of into_subject)
+ into_subject text COLLATE "C" NOT NULL, -- the survivor
+ instances integer NOT NULL DEFAULT 0, -- instances that moved to the survivor
+ duplicates integer NOT NULL DEFAULT 0, -- duplicates of a non-stackable item, revoked (the oldest is kept)
+ equipped integer NOT NULL DEFAULT 0, -- equipped slots that moved (the survivor had none there)
+ equipped_dropped integer NOT NULL DEFAULT 0, -- equipped slots dropped (the survivor's set wins)
+ merged_at text COLLATE "C" NOT NULL -- ISO 8601 UTC
+);
@@ -40,6 +40,7 @@ const { createNetworkSender } = require('openvibe-sdk/account-data');
@@ -40,6 +40,7 @@ const { createNetworkSender } = require('openvibe-sdk/account-data');
const { createInventory } = require('./inventory/store');
const { createInventory } = require('./inventory/store');
const { seed } = require('./inventory/seed');
const { seed } = require('./inventory/seed');
const accountDataLib = require('./inventory/account-data');
const accountDataLib = require('./inventory/account-data');
+const subjectMerge = require('./inventory/subject-merge');
const { assetVersion, send } = require('./render/layout');
const { assetVersion, send } = require('./render/layout');
const { html } = require('./render/html');
const { html } = require('./render/html');
@@ -138,7 +139,9 @@ async function createApp(opts = {}) {
@@ -138,7 +139,9 @@ async function createApp(opts = {}) {
// ── OpenVibe.Events → Inventory (loopback only: nginx answers 404 for /internal/) ──
// ── OpenVibe.Events → Inventory (loopback only: nginx answers 404 for /internal/) ──
// network.account.export_requested and network.account.deleted, answered by openvibe-sdk/account-data's consumer,
// network.account.export_requested and network.account.deleted, answered by openvibe-sdk/account-data's consumer,
// which reads the raw body itself (the v2 signature covers it) and refuses a request that came through a proxy.
// which reads the raw body itself (the v2 signature covers it) and refuses a request that came through a proxy.
- app.post('/internal/events', accountData.consumer({ secrets: config.events.secrets, send: accountSend, log }));
+ // network.subject.merged (ADR-029) is handed to ./inventory/subject-merge.js through onEvent.
+ app.post('/internal/events', accountData.consumer({ secrets: config.events.secrets, send: accountSend, log,
+ onEvent: (ev) => (ev.event_type === subjectMerge.TOPIC ? subjectMerge.apply(ev, { s, log }) : 'ignored:type') }));
// ── Machine endpoints ───────────────────────────────────
// ── Machine endpoints ───────────────────────────────────
app.get('/api/health', (_req, res, next) => outbox.status().then((events) => res.json({ status: 'ok', service: 'openvibe-inventory', version: VERSION, events }), next));
app.get('/api/health', (_req, res, next) => outbox.status().then((events) => res.json({ status: 'ok', service: 'openvibe-inventory', version: VERSION, events }), next));
@@ -7,7 +7,8 @@
@@ -7,7 +7,8 @@
const { createApp } = require('./app');
const { createApp } = require('./app');
const { gracefulStop } = require('openvibe-sdk/service');
const { gracefulStop } = require('openvibe-sdk/service');
const { startTelemetryRetention } = require('openvibe-sdk/telemetry');
const { startTelemetryRetention } = require('openvibe-sdk/telemetry');
-const { startSubscriptions } = require('openvibe-sdk/account-data');
+const { startSubscriptions, TOPICS: ACCOUNT_TOPICS } = require('openvibe-sdk/account-data');
+const subjectMerge = require('./inventory/subject-merge');
/**
/**
* The process stop (openvibe-sdk/service): the HTTP drain runs, then the timers clear, the Events subscriptions and the
* The process stop (openvibe-sdk/service): the HTTP drain runs, then the timers clear, the Events subscriptions and the
@@ -42,6 +43,7 @@ async function start() {
@@ -42,6 +43,7 @@ async function start() {
// INVENTORY_EVENTS_SECRET or the client secret.
// INVENTORY_EVENTS_SECRET or the client secret.
const subscriptions = startSubscriptions({
const subscriptions = startSubscriptions({
eventsUrl: config.events.url, endpoint: `http://127.0.0.1:${config.port}/internal/events`, secret: config.events.secrets[0],
eventsUrl: config.events.url, endpoint: `http://127.0.0.1:${config.port}/internal/events`, secret: config.events.secrets[0],
+ topics: [...ACCOUNT_TOPICS, subjectMerge.TOPIC],
networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
});
});
// Sent events older than a week go (openvibe-sdk outbox prune), every six hours.
// Sent events older than a week go (openvibe-sdk outbox prune), every six hours.
new file mode 100644
new file mode 100644
@@ -0,0 +1,78 @@
@@ -0,0 +1,78 @@
+'use strict';
+
+/**
+ * network.subject.merged → Inventory (ADR-029, ADR-054 §7). Two Network accounts became one: `from` is now an alias of
+ * `into`, and every item of `from` becomes `into`'s, in one transaction, once per merge_id (inventory_subject_merges):
+ *
+ * instances every instance moves to the survivor. Where both own the same item of a non-stackable kind, the
+ * oldest is kept and the other is revoked (reason account_merge), so nobody holds two of a unique item
+ * equipped the survivor's equipped set wins: a slot only `from` filled moves, a slot both filled keeps the
+ * survivor's; a slot pointing at a revoked duplicate now points at the kept one
+ * Workshop items `from` made are credited to, and managed by, the survivor
+ * ledger append-only: `from`'s rows stay as written; each moved or revoked instance gets a `merged` row
+ *
+ * Delivered to POST /internal/events beside the account topics (openvibe-sdk/account-data's consumer, onEvent).
+ */
+const TOPIC = 'network.subject.merged';
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const MERGE_RE = /^mrg_[0-9A-HJKMNP-TV-Z]{26}$/;
+const ACTOR = 'service:network';
+
+/** One envelope → 'merged' | 'unchanged' | 'ignored:<why>'. */
+async function apply(ev, { s, log = console }) {
+ if (!ev || ev.event_type !== TOPIC) return 'ignored:type';
+ if (ev.source !== 'network') return 'ignored:source';
+ const p = ev.payload && typeof ev.payload === 'object' ? ev.payload : {};
+ const mergeId = String(p.merge_id || ''), from = String(p.from || ''), into = String(p.into || '');
+ if (!MERGE_RE.test(mergeId) || !SUBJECT_RE.test(from) || !SUBJECT_RE.test(into) || from === into) return 'ignored:payload';
+ const r = await s.tx(async () => {
+ const db = s.db;
+ if (await db.maybe('SELECT 1 FROM inventory_subject_merges WHERE merge_id = $1', [mergeId])) return null;
+ const at = s.iso();
+ const detail = JSON.stringify({ merge_id: mergeId, from, into });
+ // Both own the same non-stackable item: the oldest is kept, every other copy is revoked.
+ const dupes = await db.many(`WITH owned AS (
+ SELECT i.id, i.definition_id, i.owner_subject,
+ row_number() OVER (PARTITION BY i.definition_id ORDER BY i.acquired_at, i.id) AS rn,
+ first_value(i.id) OVER (PARTITION BY i.definition_id ORDER BY i.acquired_at, i.id) AS keeper,
+ min(i.owner_subject) OVER (PARTITION BY i.definition_id) AS lo,
+ max(i.owner_subject) OVER (PARTITION BY i.definition_id) AS hi
+ FROM inventory_instances i
+ JOIN inventory_definitions d ON d.id = i.definition_id
+ JOIN inventory_kinds k ON k.id = d.kind
+ WHERE i.owner_subject IN ($1, $2) AND i.state = 'owned' AND NOT k.stackable)
+ SELECT id, definition_id, keeper FROM owned WHERE rn > 1 AND lo <> hi ORDER BY id`, [from, into]);
+ for (const d of dupes) {
+ await db.exec('UPDATE inventory_equipped SET instance_id = $2 WHERE instance_id = $1', [d.id, d.keeper]);
+ await db.exec("UPDATE inventory_instances SET state = 'revoked', updated_at = $2 WHERE id = $1", [d.id, at]);
+ }
+ // The survivor's equipped set wins; a slot only `from` filled moves.
+ const equippedDropped = Number(await db.exec(`DELETE FROM inventory_equipped f USING inventory_equipped k
+ WHERE f.owner_subject = $1 AND k.owner_subject = $2 AND k.kind = f.kind AND k.slot = f.slot`, [from, into])) || 0;
+ const equipped = Number(await db.exec('UPDATE inventory_equipped SET owner_subject = $2 WHERE owner_subject = $1', [from, into])) || 0;
+ // Two equipped rows may now point at one kept instance (both wore their copy in different slots): fine, an
+ // instance can only fill the slot its kind allows, and both rows are the survivor's.
+ const moved = await db.many('UPDATE inventory_instances SET owner_subject = $2, updated_at = $3 WHERE owner_subject = $1 RETURNING id, definition_id', [from, into, at]);
+ const dupeIds = new Set(dupes.map((d) => d.id));
+ for (const m of moved) {
+ await db.exec(`INSERT INTO inventory_ledger (at, action, owner_subject, instance_id, definition_id, actor, reason, detail)
+ VALUES ($1, 'merged', $2, $3, $4, $5, $6, $7)`, [at, into, m.id, m.definition_id, ACTOR, dupeIds.has(m.id) ? 'account_merge_duplicate_revoked' : 'account_merge', detail]);
+ }
+ for (const d of dupes) {
+ if (moved.some((m) => m.id === d.id)) continue;
+ await db.exec(`INSERT INTO inventory_ledger (at, action, owner_subject, instance_id, definition_id, actor, reason, detail)
+ VALUES ($1, 'merged', $2, $3, $4, $5, 'account_merge_duplicate_revoked', $6)`, [at, into, d.id, d.definition_id, ACTOR, detail]);
+ }
+ // Workshop: what `from` made is the survivor's to manage, with the survivor's credit.
+ await db.exec('UPDATE inventory_definitions SET credit_subject = $2 WHERE credit_subject = $1', [from, into]);
+ await db.exec('UPDATE inventory_definitions SET issuer = $2 WHERE issuer = $1', [`user:${from}`, `user:${into}`]);
+ await db.exec(`INSERT INTO inventory_subject_merges (merge_id, from_subject, into_subject, instances, duplicates, equipped, equipped_dropped, merged_at)
+ VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`, [mergeId, from, into, moved.length, dupes.length, equipped, equippedDropped, at]);
+ return { instances: moved.length, duplicates: dupes.length, equipped, equippedDropped };
+ });
+ if (!r) return 'unchanged';
+ log.log(`[Merge] ${mergeId}: ${r.instances} instance(s) and ${r.equipped} equipped slot(s) now the survivor's; ${r.duplicates} duplicate(s) revoked, ${r.equippedDropped} equipped slot(s) dropped`);
+ return 'merged';
+}
+
+module.exports = { apply, TOPIC };
new file mode 100644
new file mode 100644
@@ -0,0 +1,102 @@
@@ -0,0 +1,102 @@
+'use strict';
+/**
+ * ADR-029 / ADR-054 §7: network.subject.merged folds one account's items into the survivor's, through the signed
+ * loopback route POST /internal/events. Every instance moves; where both own the same non-stackable item the oldest is
+ * kept and the other is revoked; the survivor's equipped set wins; the ledger keeps the folded account's rows and gains
+ * one `merged` row per instance; a redelivery (same merge_id) moves nothing.
+ */
+const assert = require('assert');
+const { signDeliveryHeaders } = require('openvibe-sdk/events');
+const { boot, check, done } = require('./helpers/boot');
+
+const SECRET = `whsec_${'fixture'.repeat(6)}`;
+const MERGE = 'mrg_01JZ0000000000000000000MRG';
+
+(async () => {
+ const t = await boot({ env: { INVENTORY_EVENTS_SECRET: SECRET } });
+ const keep = t.network.addUser('keep');
+ const gone = t.network.addUser('gone');
+ const bob = t.network.addUser('bob');
+ const db = t.ctx.s.db;
+ const inv = t.ctx.inv;
+ const count = async (sql, args) => Number(await db.value(sql, args));
+ const deliver = async (event) => {
+ const body = JSON.stringify({ event, seq: 1 });
+ const res = await fetch(`${t.base}/internal/events`, { method: 'POST', body, headers: { 'Content-Type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });
+ return { status: res.status, json: await res.json().catch(() => null) };
+ };
+ const merged = (id, payload = {}, source = 'network') => ({ event_id: id, event_type: 'network.subject.merged', source, version: 1, timestamp: new Date().toISOString(),
+ payload: { merge_id: MERGE, from: gone.subject, into: keep.subject, merged_at: new Date().toISOString(), initiated_by: 'person', ...payload } });
+ const own = async (who, alias, key, acquiredAt) => {
+ const def = await inv.byAlias('service:live', alias);
+ const r = await inv.grant('service:live', { definition_id: def.id, subject: who.subject, idempotency_key: `test-merge-${key}` });
+ await db.exec('UPDATE inventory_instances SET acquired_at = $2 WHERE id = $1', [r.instance.id, acquiredAt]);
+ return r.instance.id;
+ };
+ let goneCrown, keepCrown, goneFire, bobCrown, goneLedger;
+
+ try {
+ await check('both accounts own a crown (the folded one older) and wear it; the folded one also wears a name effect', async () => {
+ goneCrown = await own(gone, 'hat_crown', 'g1', '2026-01-01T00:00:00.000Z');
+ goneFire = await own(gone, 'fx_fire', 'g2', '2026-02-01T00:00:00.000Z');
+ keepCrown = await own(keep, 'hat_crown', 'k1', '2026-06-01T00:00:00.000Z');
+ bobCrown = await own(bob, 'hat_crown', 'b1', '2025-01-01T00:00:00.000Z');
+ await inv.equip(gone.subject, `user:${gone.subject}`, { kind: 'live.hat', slot: 'hat', instance_id: goneCrown });
+ await inv.equip(gone.subject, `user:${gone.subject}`, { kind: 'live.name_effect', slot: 'name_effect', instance_id: goneFire });
+ await inv.equip(keep.subject, `user:${keep.subject}`, { kind: 'live.hat', slot: 'hat', instance_id: keepCrown });
+ goneLedger = await count('SELECT count(*) FROM inventory_ledger WHERE owner_subject = $1', [gone.subject]);
+ assert.ok(goneLedger >= 4);
+ });
+
+ await check('a payload that is not a merge, or not from Network, is acknowledged and ignored', async () => {
+ const bad = await deliver(merged('evt_01JZ0000000000000000000M00', { from: keep.subject }));
+ assert.deepStrictEqual([bad.status, bad.json.outcome], [200, 'ignored:payload']);
+ const spoof = await deliver(merged('evt_01JZ0000000000000000000M09', {}, 'live'));
+ assert.deepStrictEqual([spoof.status, spoof.json.outcome], [200, 'ignored:source']);
+ assert.strictEqual(await count('SELECT count(*) FROM inventory_instances WHERE owner_subject = $1', [gone.subject]), 2);
+ });
+
+ await check('the merge moves every item, keeps the oldest crown, and the survivor\'s equipped set wins', async () => {
+ const r = await deliver(merged('evt_01JZ0000000000000000000M01'));
+ assert.deepStrictEqual([r.status, r.json && r.json.outcome], [200, 'merged'], JSON.stringify(r.json));
+ assert.strictEqual(await count('SELECT count(*) FROM inventory_instances WHERE owner_subject = $1', [gone.subject]), 0, 'no copy under the folded subject');
+ assert.strictEqual(await count('SELECT count(*) FROM inventory_equipped WHERE owner_subject = $1', [gone.subject]), 0);
+ const rows = await db.many('SELECT id, state FROM inventory_instances WHERE owner_subject = $1 ORDER BY id', [keep.subject]);
+ const state = Object.fromEntries(rows.map((x) => [x.id, x.state]));
+ assert.deepStrictEqual(state, { [goneCrown]: 'owned', [goneFire]: 'owned', [keepCrown]: 'revoked' }, 'both instances kept as records; the older crown is the one owned');
+ const eq = await db.many('SELECT kind, slot, instance_id FROM inventory_equipped WHERE owner_subject = $1 ORDER BY kind', [keep.subject]);
+ assert.deepStrictEqual(eq, [
+ { kind: 'live.hat', slot: 'hat', instance_id: goneCrown }, // the survivor's slot, now on the crown kept
+ { kind: 'live.name_effect', slot: 'name_effect', instance_id: goneFire }, // a slot only the folded account filled
+ ]);
+ const rec = await db.maybe('SELECT from_subject, into_subject, instances, duplicates, equipped, equipped_dropped FROM inventory_subject_merges WHERE merge_id = $1', [MERGE]);
+ assert.deepStrictEqual(rec, { from_subject: gone.subject, into_subject: keep.subject, instances: 2, duplicates: 1, equipped: 1, equipped_dropped: 1 });
+ });
+
+ await check('the ledger keeps the folded account\'s rows and gains one merged row per instance', async () => {
+ assert.strictEqual(await count('SELECT count(*) FROM inventory_ledger WHERE owner_subject = $1', [gone.subject]), goneLedger, 'history is never rewritten');
+ const lines = await db.many("SELECT instance_id, reason, actor, detail->>'merge_id' AS merge_id FROM inventory_ledger WHERE action = 'merged' ORDER BY instance_id");
+ assert.strictEqual(lines.length, 3);
+ assert.ok(lines.every((l) => l.merge_id === MERGE && l.actor === 'service:network'));
+ assert.strictEqual(lines.find((l) => l.instance_id === keepCrown).reason, 'account_merge_duplicate_revoked');
+ assert.strictEqual(lines.find((l) => l.instance_id === goneFire).reason, 'account_merge');
+ });
+
+ await check('a redelivery, or the same merge under another event id, changes nothing; bob is untouched', async () => {
+ const before = await count('SELECT count(*) FROM inventory_ledger');
+ assert.strictEqual((await deliver(merged('evt_01JZ0000000000000000000M01'))).json.outcome, 'unchanged');
+ assert.strictEqual((await deliver(merged('evt_01JZ0000000000000000000M02'))).json.outcome, 'unchanged');
+ assert.strictEqual(await count('SELECT count(*) FROM inventory_ledger'), before);
+ assert.strictEqual(await count("SELECT count(*) FROM inventory_instances WHERE owner_subject = $1 AND state = 'owned'", [keep.subject]), 2);
+ assert.deepStrictEqual(await db.maybe('SELECT owner_subject, state FROM inventory_instances WHERE id = $1', [bobCrown]), { owner_subject: bob.subject, state: 'owned' });
+ });
+
+ await check('the merge topic is subscribed beside the account topics', async () => {
+ const src = require('fs').readFileSync(require('path').join(__dirname, '../server/index.js'), 'utf8');
+ assert.ok(/topics: \[\.\.\.ACCOUNT_TOPICS, subjectMerge\.TOPIC\]/.test(src));
+ });
+ } finally {
+ await t.close();
+ }
+ done();
+})();