openvibers/openvibe.billing Commits
Cutover guards: a Live-credited order never settles twice, later imports never undo Billing's own movements, host freeze CLI
Changes
@@@ -9,6 +9,7 @@"dev": "NODE_ENV=development node server/index.js","test": "node test/run.js","reconcile": "node scripts/reconcile.js",++ "freeze": "node scripts/freeze.js","import-live": "node scripts/import-live.js"},"keywords": ["openvibe", "billing", "ledger", "double-entry"],nnew file mode 100644@@@ -0,0 +1,27 @@++#!/usr/bin/env node++'use strict';++/**++ * The economy freeze from the host, without a service token (the same switch as++ * POST /api/v1/admin/freeze; ADR-012 rule 11).++ *++ * node scripts/freeze.js status++ * node scripts/freeze.js on "Live cutover" writes refused (503 billing.frozen), reads served,++ * webhooks stored (202) and held++ * node scripts/freeze.js off the running service processes the held webhooks in++ * arrival order on its next retry tick (≤ BILLING_WEBHOOK_RETRY_MS)++ */++const { loadConfig } = require('../server/config');++const { openDb } = require('../server/db');++const admin = require('../server/ops/admin');++++const [cmd = 'status', reason] = process.argv.slice(2);++if (!['status', 'on', 'off'].includes(cmd)) { console.error('usage: freeze.js status | on "<reason>" | off'); process.exit(2); }++const config = loadConfig();++const db = openDb(config.dbPath);++const os = require('os');++const state = cmd === 'status'++ ? admin.freezeState(db)++ : admin.setFreeze({ db, now: () => Date.now() }, { on: cmd === 'on', reason: reason || null, actor: { principal: 'operator:cli', host: os.hostname(), user: os.userInfo().username } });++const pending = db.prepare('SELECT COUNT(*) AS n FROM provider_events WHERE processed_at IS NULL').get().n;++console.log(JSON.stringify({ ...state, held_webhooks: pending }));++db.close();@@@ -14,9 +14,13 @@* first order's ref; the others are reported.* 3. history every Live `transactions` row becomes a historical journal transaction (type* import, status imported, test=1 before site_settings.stats_vibes_reset_at).-- * 4. opening per user, an import transaction makes user_credit = users.openvibe_bucks_balance-- * and creator_payable = users.openvibe_bucks_cashout_balance; the difference against-- * the replayed history is booked against import_adjustment and listed.++ * 4. opening per user, an import transaction makes the IMPORTED part of user_credit =++ * users.openvibe_bucks_balance and of creator_payable = users.openvibe_bucks_cashout_balance;++ * the difference against the replayed history is booked against import_adjustment and++ * listed. Billing-native movements (a legacy checkout settled here, everything after the++ * cutover) are never adjusted away by a later run.++ * Orders Live credited are imported as intents settled-in-Live (no settled_txn); a later run follows++ * an order Live credited or failed since, and settlement refuses a delivery for such an intent.* 5. subscriptions → Billing subscriptions + an entitlement for the current paid period.** Idempotent: every write is keyed (import:live:txn:<id>, legacy ids, target-state keys), so a@@@ -92,9 +96,20 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c// ── 2. payment_orders → intents + receipts ───────────const refSeen = new Map(db.prepare('SELECT provider, provider_ref, legacy_order_id FROM payment_intents WHERE provider_ref IS NOT NULL').all().map((r) => [`${r.provider}|${r.provider_ref}`, r.legacy_order_id]));-- let intentsNew = 0; let receiptsNew = 0;++ let intentsNew = 0; let receiptsNew = 0; let intentsUpdated = 0;for (const o of orders) {-- const existing = db.prepare('SELECT id FROM payment_intents WHERE legacy_order_id = ?').get(o.id);++ const existing = db.prepare('SELECT id, status, settled_txn FROM payment_intents WHERE legacy_order_id = ?').get(o.id);++ if (existing && !existing.settled_txn) {++ // Live moved the order on since an earlier run (credited or failed after the shadow++ // import): follow it, so a later delivery of a Live-credited order is recognised as++ // settled-in-Live (intents.settledInLive) instead of settling a second time. An intent++ // Billing settled itself (settled_txn) is never rewound.++ const want = o.status === 'credited' ? 'settled' : (o.status === 'failed' && ['created', 'expired'].includes(existing.status) ? 'failed' : null);++ if (want && existing.status !== want && existing.status !== 'settled') {++ db.prepare('UPDATE payment_intents SET status = ?, updated_at = ? WHERE id = ?').run(want, at, existing.id);++ intentsUpdated++;++ }++ }const legacy = { status: o.status, provider_ref: o.provider_ref, bucks: o.bucks, amount_cents: o.amount_cents, user_id: o.user_id, streamer_id: o.streamer_id };if (!existing) {let ref = o.provider_ref || null;@@@ -135,6 +150,7 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c}}report.counts.intents_new = intentsNew;++ report.counts.intents_updated = intentsUpdated;report.counts.receipts_new = receiptsNew;// ── 3. history ───────────────────────────────────────@@@ -191,6 +207,14 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = cconst upsertHold = db.prepare(`INSERT INTO import_holds (live_user_id, owner, reason, credit_bits, payable_bits, first_seen_at, updated_at)VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (live_user_id) DO UPDATE SET credit_bits = excluded.credit_bits, payable_bits = excluded.payable_bits, updated_at = excluded.updated_at`);const byId = new Map(users.map((u) => [String(u.id), u]));++ // What the imports themselves put in an account (history, earlier opening balances, released++ // holds) — the part the Live column describes. Anything Billing did natively (a legacy checkout++ // that settled here, and everything after the cutover) is not Live's to overwrite, so a later++ // run (a hold release, a late snapshot) never books an adjustment against it.++ const importedStmt = db.prepare(`SELECT COALESCE(SUM(e.amount), 0) AS n FROM ledger_entries e++ JOIN accounts a ON a.id = e.account_id JOIN transactions t ON t.id = e.txn_id++ WHERE a.kind = ? AND a.owner_subject IS ? AND a.currency = ? AND t.type = 'import'`);++ const imported = (acct) => importedStmt.get(acct.kind, acct.owner, acct.currency).n;const openingIds = new Set([...ids].filter((i) => byId.has(i)));for (const i of openingIds) {const u = byId.get(i);@@@ -211,7 +235,7 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = cfor (const [acct, name, raw] of [[A.credit, 'user_credit', u.credit], [A.payable, 'creator_payable', u.payable]]) {const target = Math.round(Number(raw) || 0);if (target !== Number(raw)) report.anomalies.push({ kind: 'fractional_balance', live_user_id: u.id, account: name, value: raw, imported_as: target });-- const current = balance(db, acct(own));++ const current = imported(acct(own));const d = target - current;if (d !== 0) {importTxn({@@@ -59,6 +59,21 @@ function create(ctx, input) {return find(db, row.id);}++/**++ * An intent imported from a Live payment order that Live had already credited before the cutover.++ * Its money effect lives in the imported history/opening balances, not in a Billing settlement, so++ * a provider delivery for it that reaches Billing afterwards (a retry of a delivery Live had++ * acknowledged, a manual resend) must not settle it again.++ */++function settledInLive(i) {++ return !!(i && i.legacy_order_id != null && i.status === 'settled' && !i.settled_txn);++}++function refuseIfSettledInLive(i) {++ if (settledInLive(i)) {++ fail(409, 'billing.intent_settled_in_live', `intent ${i.id} (Live order ${i.legacy_order_id}) was already credited by Live before the cutover; this delivery is held for review, not settled again`);++ }++}++function setProviderRef(ctx, id, ref) {ctx.db.prepare('UPDATE payment_intents SET provider_ref = ?, updated_at = ? WHERE id = ?').run(ref, iso(ctx.now()), id);}@@@ -85,4 +100,4 @@ function present(i) {};}--module.exports = { find, findByProviderRef, create, setProviderRef, mergeMetadata, setStatus, markSettled, present, parse };++module.exports = { find, findByProviderRef, create, setProviderRef, mergeMetadata, setStatus, markSettled, present, parse, settledInLive, refuseIfSettledInLive };@@@ -33,6 +33,7 @@ function settle(ctx, input) {if (intent.provider !== input.provider) fail(422, 'billing.intent_mismatch', `intent ${intent.id} belongs to ${intent.provider}, not ${input.provider}`);if (input.subject && input.subject !== intent.subject) fail(422, 'billing.intent_mismatch', 'the intent belongs to another subject');if (intent.status === 'settled' && intent.settled_txn) return { txn: getTxn(db, intent.settled_txn), replay: true };++ intents.refuseIfSettledInLive(intent);}const subject = intent ? intent.subject : input.subject;if (!subject) fail(422, 'billing.invalid_subject', 'a purchase needs a subject or an intent');@@@ -80,6 +80,7 @@ function pay(ctx, input) {return db.transaction(() => {const existingTxn = db.prepare('SELECT id FROM transactions WHERE idempotency_key = ?').get(input.idempotencyKey);if (existingTxn) return result(db, getTxn(db, existingTxn.id), ctx, true);++ if (input.intentId) intents.refuseIfSettledInLive(intents.find(db, input.intentId));const r = input.receipt;if (r) {const dup = db.prepare('SELECT id FROM transactions WHERE receipt_ref = ?').get(r.receiptRef);@@@ -109,6 +109,9 @@ function createPowerchat(cfg, { network } = {}) {const intent = intents.find(ctx.db, m[1]);if (!intent || intent.kind !== 'subscription' || !intent.streamer_subject) return { effect: 'none', reason: `subscription checkout ${ref} has no subscription intent` };const route = intent.route === 'direct' ? 'direct' : 'site';++ if (cents + 1 < intent.amount_cents && intents.settledInLive(intent)) {++ return { effect: 'none', reason: `underpaid delivery for ${ref}, which Live already credited before the cutover — review`, review: true };++ }if (cents + 1 < intent.amount_cents) {if (route === 'direct') return { effect: 'none', reason: 'underpaid direct subscription: an EXTERNAL tip to the streamer' };return { effect: 'tip', args: { provider: 'powerchat', receiptRef, paidCents: cents, from: intent.subject, to: intent.streamer_subject, test: isTest, idempotencyKey: key, actor, metadata: { ...meta, underpaid_subscription_intent: intent.id } } };nnew file mode 100644@@@ -0,0 +1,150 @@++'use strict';++/**++ * Cutover guards (docs/live-cutover.md): a Live payment order that Live already credited must++ * never be credited again by Billing when its PowerChat delivery reaches Billing after the webhook++ * is re-pointed (a PowerChat retry of a delivery Live acknowledged, or a manual resend). The import++ * marks such intents settled-in-Live — also when Live credited the order AFTER an earlier shadow++ * import — and settlement refuses them (rejected, listed by reconciliation for review). A legacy++ * order Live never credited still settles exactly once (a checkout that was in flight), and a later++ * import run never books an opening-balance adjustment against what Billing settled itself.++ */++const assert = require('assert');++const path = require('path');++const Database = require('better-sqlite3');++const { boot, check, done } = require('./helpers/app');++++function fabricateLive(file) {++ const live = new Database(file);++ const recent = new Date(Date.now() - 30 * 60_000).toISOString().replace('T', ' ').slice(0, 19);++ live.exec(`++ CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, openvibe_bucks_balance REAL DEFAULT 0.00,++ openvibe_bucks_cashout_balance REAL DEFAULT 0.00);++ CREATE TABLE transactions (id INTEGER PRIMARY KEY AUTOINCREMENT, from_user_id INTEGER, to_user_id INTEGER, stream_id INTEGER,++ amount INTEGER NOT NULL, type TEXT NOT NULL, status TEXT DEFAULT 'completed', message TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP);++ CREATE TABLE payment_orders (id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, provider TEXT NOT NULL, provider_ref TEXT,++ kind TEXT NOT NULL DEFAULT 'bucks', amount_cents INTEGER NOT NULL DEFAULT 0, currency TEXT DEFAULT 'usd', bucks INTEGER DEFAULT 0,++ streamer_id INTEGER, status TEXT DEFAULT 'pending', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP);++ CREATE TABLE subscriptions (id INTEGER PRIMARY KEY AUTOINCREMENT, subscriber_id INTEGER NOT NULL, streamer_id INTEGER NOT NULL, tier INTEGER DEFAULT 1,++ provider TEXT, provider_ref TEXT, price_cents INTEGER DEFAULT 0, status TEXT DEFAULT 'active', auto_renew INTEGER DEFAULT 0,++ current_period_end DATETIME, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP);++ INSERT INTO users (id, username, openvibe_bucks_balance, openvibe_bucks_cashout_balance) VALUES (1, 'alice', 500, 0), (2, 'bob', 0, 349), (3, 'carol', 0, 0);++ INSERT INTO transactions (id, from_user_id, to_user_id, amount, type, status, created_at) VALUES (1, NULL, 1, 500, 'purchase', 'completed', '${recent}');++ `);++ live.prepare(`INSERT INTO payment_orders (id, user_id, provider, provider_ref, kind, amount_cents, bucks, streamer_id, status, created_at) VALUES++ (1, 1, 'powerchat', NULL, 'bucks', 700, 500, NULL, 'credited', ?),++ (2, 3, 'powerchat', NULL, 'bucks', 700, 500, NULL, 'pending', ?),++ (3, 1, 'powerchat', 'site:fee=50', 'subscription', 549, 0, 2, 'credited', ?),++ (4, 3, 'powerchat', NULL, 'bucks', 1300, 1000, NULL, 'pending', ?)`).run(recent, recent, recent, recent);++ const end = new Date(Date.now() + 20 * 86_400_000).toISOString().replace('T', ' ').slice(0, 19);++ live.prepare(`INSERT INTO subscriptions (id, subscriber_id, streamer_id, provider, price_cents, status, current_period_end, created_at)++ VALUES (1, 1, 2, 'powerchat', 549, 'active', ?, ?)`).run(end, recent);++ live.close();++}++++(async () => {++ const t = await boot();++ const alice = t.network.addUser(1);++ const bob = t.network.addUser(2);++ const carol = t.network.addUser(3);++ const file = path.join(t.dir, 'live-snapshot.db');++ fabricateLive(file);++ const { importLive } = require('../server/importer/live');++ const { createIdentity } = require('../server/network');++ const identity = createIdentity(t.config);++ const run = async () => {++ const live = new Database(file, { readonly: true, fileMustExist: true });++ try { return await importLive(t.ctx, { live, resolveLiveUsers: identity.resolveLiveUsers, log: { log() {} } }); } finally { live.close(); }++ };++ const bal = (kind, owner) => { const { balance } = require('../server/ledger'); return balance(t.db, { kind, owner, currency: 'vibes-bits' }); };++ const intent = (order) => t.db.prepare('SELECT * FROM payment_intents WHERE legacy_order_id = ?').get(order);++ let n = 0;++ const tip = (ref, cents) => t.powerchat({ type: 'donation.completed', streamer: { username: 'openvibe' }, data: { eventId: `ev_cutover_${++n}`, appExternalRef: ref, amountUsdCents: cents } });++ const result = (r) => t.db.prepare('SELECT result FROM provider_events WHERE id = ?').get(r.json.event).result;++ console.log('cutover guards');++++ await run();++++ await check('an order Live credited is imported as settled-in-Live', async () => {++ assert.strictEqual(intent(1).status, 'settled');++ assert.strictEqual(intent(1).settled_txn, null);++ assert.strictEqual(bal('user_credit', alice), 500);++ });++++ await check('its PowerChat delivery reaching Billing is refused (rejected, for review), never credited twice', async () => {++ const r = await tip('pcorder:1', 700);++ assert.strictEqual(r.status, 200, JSON.stringify(r.json));++ const res = JSON.parse(result(r));++ assert.strictEqual(res.effect, 'rejected');++ assert.strictEqual(res.code, 'billing.intent_settled_in_live');++ assert.strictEqual(bal('user_credit', alice), 500);++ const rec = t.assertReconciled('after a refused legacy delivery');++ assert.ok(rec.warnings.rejected_events.some((e) => e.id === r.json.event), 'listed for review');++ });++++ await check('a subscription order Live credited is not paid (or granted) again either', async () => {++ const ents = t.db.prepare('SELECT COUNT(*) AS n FROM entitlements').get().n;++ const r = await tip('pcsub:3', 549);++ assert.strictEqual(JSON.parse(result(r)).code, 'billing.intent_settled_in_live');++ assert.strictEqual(t.db.prepare('SELECT COUNT(*) AS n FROM entitlements').get().n, ents);++ assert.strictEqual(bal('creator_payable', bob), 349);++ });++++ await check('an order Live credited AFTER the shadow import is settled-in-Live after the final import', async () => {++ assert.strictEqual(intent(4).status, 'created');++ const live = new Database(file);++ live.exec(`UPDATE payment_orders SET status = 'credited' WHERE id = 4;++ UPDATE users SET openvibe_bucks_balance = 1000 WHERE id = 3;++ INSERT INTO transactions (from_user_id, to_user_id, amount, type, status) VALUES (NULL, 3, 1000, 'purchase', 'completed');`);++ live.close();++ const report = await run();++ assert.ok(report.counts.intents_updated >= 1, JSON.stringify(report.counts));++ assert.strictEqual(intent(4).status, 'settled');++ assert.strictEqual(bal('user_credit', carol), 1000);++ const r = await tip('pcorder:4', 1300);++ assert.strictEqual(JSON.parse(result(r)).code, 'billing.intent_settled_in_live');++ assert.strictEqual(bal('user_credit', carol), 1000);++ });++++ await check('a legacy order Live never credited (a checkout in flight) settles exactly once', async () => {++ const r = await tip('pcorder:2', 700);++ assert.strictEqual(JSON.parse(result(r)).effect, 'settled');++ assert.strictEqual(bal('user_credit', carol), 1500);++ assert.strictEqual(intent(2).status, 'settled');++ const again = await t.powerchat({ type: 'donation.completed', streamer: { username: 'openvibe' }, data: { eventId: `ev_cutover_${n}`, appExternalRef: 'pcorder:2', amountUsdCents: 700 } });++ assert.strictEqual(JSON.parse(result(again)).effect, 'duplicate_receipt');++ assert.strictEqual(bal('user_credit', carol), 1500);++ const report = await run();++ assert.strictEqual(intent(2).status, 'settled', 'a re-import never rewinds what Billing settled');++ assert.strictEqual(report.new_transactions, 0);++ assert.deepStrictEqual(report.adjustments, [], 'no adjustment against a Billing-native settlement');++ assert.strictEqual(bal('user_credit', carol), 1500);++ t.assertReconciled('after the final import');++ });++++ await check('the host freeze CLI holds webhooks and refuses writes; after "off" the held delivery settles once', async () => {++ const { spawnSync } = require('child_process');++ const cli = (...args) => {++ const r = spawnSync(process.execPath, [path.join(__dirname, '..', 'scripts', 'freeze.js'), ...args], { env: { ...process.env, BILLING_DB_PATH: t.config.dbPath }, encoding: 'utf8' });++ assert.strictEqual(r.status, 0, r.stderr);++ return JSON.parse(r.stdout);++ };++ assert.strictEqual(cli('on', 'Live cutover').frozen, true);++ const w = await t.call('POST', '/api/v1/transfers', { body: { from: { type: 'user', id: carol }, to: { type: 'user', id: bob }, amount: 5 } });++ assert.strictEqual(w.status, 503); assert.strictEqual(w.json.code, 'billing.frozen');++ const live = new Database(file);++ live.exec("INSERT INTO payment_orders (id, user_id, provider, kind, amount_cents, bucks, status) VALUES (5, 3, 'powerchat', 'bucks', 700, 500, 'pending')");++ live.close();++ await run(); // the final import, while Billing is frozen++ const held = await tip('pcorder:5', 700);++ assert.strictEqual(held.status, 202, JSON.stringify(held.json));++ assert.strictEqual(cli('status').held_webhooks, 1);++ assert.strictEqual(cli('off').frozen, false);++ await require('../server/providers').processPending(t.ctx, t.app.locals.adapters); // the service's retry tick++ assert.strictEqual(JSON.parse(result(held)).effect, 'settled');++ assert.strictEqual(bal('user_credit', carol), 2000);++ t.assertReconciled('after the freeze window');++ });++++ await t.close();++ done();++})();
@@ -9,6 +9,7 @@
@@ -9,6 +9,7 @@
"dev": "NODE_ENV=development node server/index.js",
"dev": "NODE_ENV=development node server/index.js",
"test": "node test/run.js",
"test": "node test/run.js",
"reconcile": "node scripts/reconcile.js",
"reconcile": "node scripts/reconcile.js",
+ "freeze": "node scripts/freeze.js",
"import-live": "node scripts/import-live.js"
"import-live": "node scripts/import-live.js"
},
},
"keywords": ["openvibe", "billing", "ledger", "double-entry"],
"keywords": ["openvibe", "billing", "ledger", "double-entry"],
new file mode 100644
new file mode 100644
@@ -0,0 +1,27 @@
@@ -0,0 +1,27 @@
+#!/usr/bin/env node
+'use strict';
+/**
+ * The economy freeze from the host, without a service token (the same switch as
+ * POST /api/v1/admin/freeze; ADR-012 rule 11).
+ *
+ * node scripts/freeze.js status
+ * node scripts/freeze.js on "Live cutover" writes refused (503 billing.frozen), reads served,
+ * webhooks stored (202) and held
+ * node scripts/freeze.js off the running service processes the held webhooks in
+ * arrival order on its next retry tick (≤ BILLING_WEBHOOK_RETRY_MS)
+ */
+const { loadConfig } = require('../server/config');
+const { openDb } = require('../server/db');
+const admin = require('../server/ops/admin');
+
+const [cmd = 'status', reason] = process.argv.slice(2);
+if (!['status', 'on', 'off'].includes(cmd)) { console.error('usage: freeze.js status | on "<reason>" | off'); process.exit(2); }
+const config = loadConfig();
+const db = openDb(config.dbPath);
+const os = require('os');
+const state = cmd === 'status'
+ ? admin.freezeState(db)
+ : admin.setFreeze({ db, now: () => Date.now() }, { on: cmd === 'on', reason: reason || null, actor: { principal: 'operator:cli', host: os.hostname(), user: os.userInfo().username } });
+const pending = db.prepare('SELECT COUNT(*) AS n FROM provider_events WHERE processed_at IS NULL').get().n;
+console.log(JSON.stringify({ ...state, held_webhooks: pending }));
+db.close();
@@ -14,9 +14,13 @@
@@ -14,9 +14,13 @@
* first order's ref; the others are reported.
* first order's ref; the others are reported.
* 3. history every Live `transactions` row becomes a historical journal transaction (type
* 3. history every Live `transactions` row becomes a historical journal transaction (type
* import, status imported, test=1 before site_settings.stats_vibes_reset_at).
* import, status imported, test=1 before site_settings.stats_vibes_reset_at).
- * 4. opening per user, an import transaction makes user_credit = users.openvibe_bucks_balance
- * and creator_payable = users.openvibe_bucks_cashout_balance; the difference against
- * the replayed history is booked against import_adjustment and listed.
+ * 4. opening per user, an import transaction makes the IMPORTED part of user_credit =
+ * users.openvibe_bucks_balance and of creator_payable = users.openvibe_bucks_cashout_balance;
+ * the difference against the replayed history is booked against import_adjustment and
+ * listed. Billing-native movements (a legacy checkout settled here, everything after the
+ * cutover) are never adjusted away by a later run.
+ * Orders Live credited are imported as intents settled-in-Live (no settled_txn); a later run follows
+ * an order Live credited or failed since, and settlement refuses a delivery for such an intent.
* 5. subscriptions → Billing subscriptions + an entitlement for the current paid period.
* 5. subscriptions → Billing subscriptions + an entitlement for the current paid period.
*
*
* Idempotent: every write is keyed (import:live:txn:<id>, legacy ids, target-state keys), so a
* Idempotent: every write is keyed (import:live:txn:<id>, legacy ids, target-state keys), so a
@@ -92,9 +96,20 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c
@@ -92,9 +96,20 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c
// ── 2. payment_orders → intents + receipts ───────────
// ── 2. payment_orders → intents + receipts ───────────
const refSeen = new Map(db.prepare('SELECT provider, provider_ref, legacy_order_id FROM payment_intents WHERE provider_ref IS NOT NULL').all()
const refSeen = new Map(db.prepare('SELECT provider, provider_ref, legacy_order_id FROM payment_intents WHERE provider_ref IS NOT NULL').all()
.map((r) => [`${r.provider}|${r.provider_ref}`, r.legacy_order_id]));
.map((r) => [`${r.provider}|${r.provider_ref}`, r.legacy_order_id]));
- let intentsNew = 0; let receiptsNew = 0;
+ let intentsNew = 0; let receiptsNew = 0; let intentsUpdated = 0;
for (const o of orders) {
for (const o of orders) {
- const existing = db.prepare('SELECT id FROM payment_intents WHERE legacy_order_id = ?').get(o.id);
+ const existing = db.prepare('SELECT id, status, settled_txn FROM payment_intents WHERE legacy_order_id = ?').get(o.id);
+ if (existing && !existing.settled_txn) {
+ // Live moved the order on since an earlier run (credited or failed after the shadow
+ // import): follow it, so a later delivery of a Live-credited order is recognised as
+ // settled-in-Live (intents.settledInLive) instead of settling a second time. An intent
+ // Billing settled itself (settled_txn) is never rewound.
+ const want = o.status === 'credited' ? 'settled' : (o.status === 'failed' && ['created', 'expired'].includes(existing.status) ? 'failed' : null);
+ if (want && existing.status !== want && existing.status !== 'settled') {
+ db.prepare('UPDATE payment_intents SET status = ?, updated_at = ? WHERE id = ?').run(want, at, existing.id);
+ intentsUpdated++;
+ }
+ }
const legacy = { status: o.status, provider_ref: o.provider_ref, bucks: o.bucks, amount_cents: o.amount_cents, user_id: o.user_id, streamer_id: o.streamer_id };
const legacy = { status: o.status, provider_ref: o.provider_ref, bucks: o.bucks, amount_cents: o.amount_cents, user_id: o.user_id, streamer_id: o.streamer_id };
if (!existing) {
if (!existing) {
let ref = o.provider_ref || null;
let ref = o.provider_ref || null;
@@ -135,6 +150,7 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c
@@ -135,6 +150,7 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c
}
}
}
}
report.counts.intents_new = intentsNew;
report.counts.intents_new = intentsNew;
+ report.counts.intents_updated = intentsUpdated;
report.counts.receipts_new = receiptsNew;
report.counts.receipts_new = receiptsNew;
// ── 3. history ───────────────────────────────────────
// ── 3. history ───────────────────────────────────────
@@ -191,6 +207,14 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c
@@ -191,6 +207,14 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c
const upsertHold = db.prepare(`INSERT INTO import_holds (live_user_id, owner, reason, credit_bits, payable_bits, first_seen_at, updated_at)
const upsertHold = db.prepare(`INSERT INTO import_holds (live_user_id, owner, reason, credit_bits, payable_bits, first_seen_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (live_user_id) DO UPDATE SET credit_bits = excluded.credit_bits, payable_bits = excluded.payable_bits, updated_at = excluded.updated_at`);
VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (live_user_id) DO UPDATE SET credit_bits = excluded.credit_bits, payable_bits = excluded.payable_bits, updated_at = excluded.updated_at`);
const byId = new Map(users.map((u) => [String(u.id), u]));
const byId = new Map(users.map((u) => [String(u.id), u]));
+ // What the imports themselves put in an account (history, earlier opening balances, released
+ // holds) — the part the Live column describes. Anything Billing did natively (a legacy checkout
+ // that settled here, and everything after the cutover) is not Live's to overwrite, so a later
+ // run (a hold release, a late snapshot) never books an adjustment against it.
+ const importedStmt = db.prepare(`SELECT COALESCE(SUM(e.amount), 0) AS n FROM ledger_entries e
+ JOIN accounts a ON a.id = e.account_id JOIN transactions t ON t.id = e.txn_id
+ WHERE a.kind = ? AND a.owner_subject IS ? AND a.currency = ? AND t.type = 'import'`);
+ const imported = (acct) => importedStmt.get(acct.kind, acct.owner, acct.currency).n;
const openingIds = new Set([...ids].filter((i) => byId.has(i)));
const openingIds = new Set([...ids].filter((i) => byId.has(i)));
for (const i of openingIds) {
for (const i of openingIds) {
const u = byId.get(i);
const u = byId.get(i);
@@ -211,7 +235,7 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c
@@ -211,7 +235,7 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c
for (const [acct, name, raw] of [[A.credit, 'user_credit', u.credit], [A.payable, 'creator_payable', u.payable]]) {
for (const [acct, name, raw] of [[A.credit, 'user_credit', u.credit], [A.payable, 'creator_payable', u.payable]]) {
const target = Math.round(Number(raw) || 0);
const target = Math.round(Number(raw) || 0);
if (target !== Number(raw)) report.anomalies.push({ kind: 'fractional_balance', live_user_id: u.id, account: name, value: raw, imported_as: target });
if (target !== Number(raw)) report.anomalies.push({ kind: 'fractional_balance', live_user_id: u.id, account: name, value: raw, imported_as: target });
- const current = balance(db, acct(own));
+ const current = imported(acct(own));
const d = target - current;
const d = target - current;
if (d !== 0) {
if (d !== 0) {
importTxn({
importTxn({
@@ -59,6 +59,21 @@ function create(ctx, input) {
@@ -59,6 +59,21 @@ function create(ctx, input) {
return find(db, row.id);
return find(db, row.id);
}
}
+/**
+ * An intent imported from a Live payment order that Live had already credited before the cutover.
+ * Its money effect lives in the imported history/opening balances, not in a Billing settlement, so
+ * a provider delivery for it that reaches Billing afterwards (a retry of a delivery Live had
+ * acknowledged, a manual resend) must not settle it again.
+ */
+function settledInLive(i) {
+ return !!(i && i.legacy_order_id != null && i.status === 'settled' && !i.settled_txn);
+}
+function refuseIfSettledInLive(i) {
+ if (settledInLive(i)) {
+ fail(409, 'billing.intent_settled_in_live', `intent ${i.id} (Live order ${i.legacy_order_id}) was already credited by Live before the cutover; this delivery is held for review, not settled again`);
+ }
+}
+
function setProviderRef(ctx, id, ref) {
function setProviderRef(ctx, id, ref) {
ctx.db.prepare('UPDATE payment_intents SET provider_ref = ?, updated_at = ? WHERE id = ?').run(ref, iso(ctx.now()), id);
ctx.db.prepare('UPDATE payment_intents SET provider_ref = ?, updated_at = ? WHERE id = ?').run(ref, iso(ctx.now()), id);
}
}
@@ -85,4 +100,4 @@ function present(i) {
@@ -85,4 +100,4 @@ function present(i) {
};
};
}
}
-module.exports = { find, findByProviderRef, create, setProviderRef, mergeMetadata, setStatus, markSettled, present, parse };
+module.exports = { find, findByProviderRef, create, setProviderRef, mergeMetadata, setStatus, markSettled, present, parse, settledInLive, refuseIfSettledInLive };
@@ -33,6 +33,7 @@ function settle(ctx, input) {
@@ -33,6 +33,7 @@ function settle(ctx, input) {
if (intent.provider !== input.provider) fail(422, 'billing.intent_mismatch', `intent ${intent.id} belongs to ${intent.provider}, not ${input.provider}`);
if (intent.provider !== input.provider) fail(422, 'billing.intent_mismatch', `intent ${intent.id} belongs to ${intent.provider}, not ${input.provider}`);
if (input.subject && input.subject !== intent.subject) fail(422, 'billing.intent_mismatch', 'the intent belongs to another subject');
if (input.subject && input.subject !== intent.subject) fail(422, 'billing.intent_mismatch', 'the intent belongs to another subject');
if (intent.status === 'settled' && intent.settled_txn) return { txn: getTxn(db, intent.settled_txn), replay: true };
if (intent.status === 'settled' && intent.settled_txn) return { txn: getTxn(db, intent.settled_txn), replay: true };
+ intents.refuseIfSettledInLive(intent);
}
}
const subject = intent ? intent.subject : input.subject;
const subject = intent ? intent.subject : input.subject;
if (!subject) fail(422, 'billing.invalid_subject', 'a purchase needs a subject or an intent');
if (!subject) fail(422, 'billing.invalid_subject', 'a purchase needs a subject or an intent');
@@ -80,6 +80,7 @@ function pay(ctx, input) {
@@ -80,6 +80,7 @@ function pay(ctx, input) {
return db.transaction(() => {
return db.transaction(() => {
const existingTxn = db.prepare('SELECT id FROM transactions WHERE idempotency_key = ?').get(input.idempotencyKey);
const existingTxn = db.prepare('SELECT id FROM transactions WHERE idempotency_key = ?').get(input.idempotencyKey);
if (existingTxn) return result(db, getTxn(db, existingTxn.id), ctx, true);
if (existingTxn) return result(db, getTxn(db, existingTxn.id), ctx, true);
+ if (input.intentId) intents.refuseIfSettledInLive(intents.find(db, input.intentId));
const r = input.receipt;
const r = input.receipt;
if (r) {
if (r) {
const dup = db.prepare('SELECT id FROM transactions WHERE receipt_ref = ?').get(r.receiptRef);
const dup = db.prepare('SELECT id FROM transactions WHERE receipt_ref = ?').get(r.receiptRef);
@@ -109,6 +109,9 @@ function createPowerchat(cfg, { network } = {}) {
@@ -109,6 +109,9 @@ function createPowerchat(cfg, { network } = {}) {
const intent = intents.find(ctx.db, m[1]);
const intent = intents.find(ctx.db, m[1]);
if (!intent || intent.kind !== 'subscription' || !intent.streamer_subject) return { effect: 'none', reason: `subscription checkout ${ref} has no subscription intent` };
if (!intent || intent.kind !== 'subscription' || !intent.streamer_subject) return { effect: 'none', reason: `subscription checkout ${ref} has no subscription intent` };
const route = intent.route === 'direct' ? 'direct' : 'site';
const route = intent.route === 'direct' ? 'direct' : 'site';
+ if (cents + 1 < intent.amount_cents && intents.settledInLive(intent)) {
+ return { effect: 'none', reason: `underpaid delivery for ${ref}, which Live already credited before the cutover — review`, review: true };
+ }
if (cents + 1 < intent.amount_cents) {
if (cents + 1 < intent.amount_cents) {
if (route === 'direct') return { effect: 'none', reason: 'underpaid direct subscription: an EXTERNAL tip to the streamer' };
if (route === 'direct') return { effect: 'none', reason: 'underpaid direct subscription: an EXTERNAL tip to the streamer' };
return { effect: 'tip', args: { provider: 'powerchat', receiptRef, paidCents: cents, from: intent.subject, to: intent.streamer_subject, test: isTest, idempotencyKey: key, actor, metadata: { ...meta, underpaid_subscription_intent: intent.id } } };
return { effect: 'tip', args: { provider: 'powerchat', receiptRef, paidCents: cents, from: intent.subject, to: intent.streamer_subject, test: isTest, idempotencyKey: key, actor, metadata: { ...meta, underpaid_subscription_intent: intent.id } } };
new file mode 100644
new file mode 100644
@@ -0,0 +1,150 @@
@@ -0,0 +1,150 @@
+'use strict';
+/**
+ * Cutover guards (docs/live-cutover.md): a Live payment order that Live already credited must
+ * never be credited again by Billing when its PowerChat delivery reaches Billing after the webhook
+ * is re-pointed (a PowerChat retry of a delivery Live acknowledged, or a manual resend). The import
+ * marks such intents settled-in-Live — also when Live credited the order AFTER an earlier shadow
+ * import — and settlement refuses them (rejected, listed by reconciliation for review). A legacy
+ * order Live never credited still settles exactly once (a checkout that was in flight), and a later
+ * import run never books an opening-balance adjustment against what Billing settled itself.
+ */
+const assert = require('assert');
+const path = require('path');
+const Database = require('better-sqlite3');
+const { boot, check, done } = require('./helpers/app');
+
+function fabricateLive(file) {
+ const live = new Database(file);
+ const recent = new Date(Date.now() - 30 * 60_000).toISOString().replace('T', ' ').slice(0, 19);
+ live.exec(`
+ CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, openvibe_bucks_balance REAL DEFAULT 0.00,
+ openvibe_bucks_cashout_balance REAL DEFAULT 0.00);
+ CREATE TABLE transactions (id INTEGER PRIMARY KEY AUTOINCREMENT, from_user_id INTEGER, to_user_id INTEGER, stream_id INTEGER,
+ amount INTEGER NOT NULL, type TEXT NOT NULL, status TEXT DEFAULT 'completed', message TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP);
+ CREATE TABLE payment_orders (id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, provider TEXT NOT NULL, provider_ref TEXT,
+ kind TEXT NOT NULL DEFAULT 'bucks', amount_cents INTEGER NOT NULL DEFAULT 0, currency TEXT DEFAULT 'usd', bucks INTEGER DEFAULT 0,
+ streamer_id INTEGER, status TEXT DEFAULT 'pending', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP);
+ CREATE TABLE subscriptions (id INTEGER PRIMARY KEY AUTOINCREMENT, subscriber_id INTEGER NOT NULL, streamer_id INTEGER NOT NULL, tier INTEGER DEFAULT 1,
+ provider TEXT, provider_ref TEXT, price_cents INTEGER DEFAULT 0, status TEXT DEFAULT 'active', auto_renew INTEGER DEFAULT 0,
+ current_period_end DATETIME, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP);
+ INSERT INTO users (id, username, openvibe_bucks_balance, openvibe_bucks_cashout_balance) VALUES (1, 'alice', 500, 0), (2, 'bob', 0, 349), (3, 'carol', 0, 0);
+ INSERT INTO transactions (id, from_user_id, to_user_id, amount, type, status, created_at) VALUES (1, NULL, 1, 500, 'purchase', 'completed', '${recent}');
+ `);
+ live.prepare(`INSERT INTO payment_orders (id, user_id, provider, provider_ref, kind, amount_cents, bucks, streamer_id, status, created_at) VALUES
+ (1, 1, 'powerchat', NULL, 'bucks', 700, 500, NULL, 'credited', ?),
+ (2, 3, 'powerchat', NULL, 'bucks', 700, 500, NULL, 'pending', ?),
+ (3, 1, 'powerchat', 'site:fee=50', 'subscription', 549, 0, 2, 'credited', ?),
+ (4, 3, 'powerchat', NULL, 'bucks', 1300, 1000, NULL, 'pending', ?)`).run(recent, recent, recent, recent);
+ const end = new Date(Date.now() + 20 * 86_400_000).toISOString().replace('T', ' ').slice(0, 19);
+ live.prepare(`INSERT INTO subscriptions (id, subscriber_id, streamer_id, provider, price_cents, status, current_period_end, created_at)
+ VALUES (1, 1, 2, 'powerchat', 549, 'active', ?, ?)`).run(end, recent);
+ live.close();
+}
+
+(async () => {
+ const t = await boot();
+ const alice = t.network.addUser(1);
+ const bob = t.network.addUser(2);
+ const carol = t.network.addUser(3);
+ const file = path.join(t.dir, 'live-snapshot.db');
+ fabricateLive(file);
+ const { importLive } = require('../server/importer/live');
+ const { createIdentity } = require('../server/network');
+ const identity = createIdentity(t.config);
+ const run = async () => {
+ const live = new Database(file, { readonly: true, fileMustExist: true });
+ try { return await importLive(t.ctx, { live, resolveLiveUsers: identity.resolveLiveUsers, log: { log() {} } }); } finally { live.close(); }
+ };
+ const bal = (kind, owner) => { const { balance } = require('../server/ledger'); return balance(t.db, { kind, owner, currency: 'vibes-bits' }); };
+ const intent = (order) => t.db.prepare('SELECT * FROM payment_intents WHERE legacy_order_id = ?').get(order);
+ let n = 0;
+ const tip = (ref, cents) => t.powerchat({ type: 'donation.completed', streamer: { username: 'openvibe' }, data: { eventId: `ev_cutover_${++n}`, appExternalRef: ref, amountUsdCents: cents } });
+ const result = (r) => t.db.prepare('SELECT result FROM provider_events WHERE id = ?').get(r.json.event).result;
+ console.log('cutover guards');
+
+ await run();
+
+ await check('an order Live credited is imported as settled-in-Live', async () => {
+ assert.strictEqual(intent(1).status, 'settled');
+ assert.strictEqual(intent(1).settled_txn, null);
+ assert.strictEqual(bal('user_credit', alice), 500);
+ });
+
+ await check('its PowerChat delivery reaching Billing is refused (rejected, for review), never credited twice', async () => {
+ const r = await tip('pcorder:1', 700);
+ assert.strictEqual(r.status, 200, JSON.stringify(r.json));
+ const res = JSON.parse(result(r));
+ assert.strictEqual(res.effect, 'rejected');
+ assert.strictEqual(res.code, 'billing.intent_settled_in_live');
+ assert.strictEqual(bal('user_credit', alice), 500);
+ const rec = t.assertReconciled('after a refused legacy delivery');
+ assert.ok(rec.warnings.rejected_events.some((e) => e.id === r.json.event), 'listed for review');
+ });
+
+ await check('a subscription order Live credited is not paid (or granted) again either', async () => {
+ const ents = t.db.prepare('SELECT COUNT(*) AS n FROM entitlements').get().n;
+ const r = await tip('pcsub:3', 549);
+ assert.strictEqual(JSON.parse(result(r)).code, 'billing.intent_settled_in_live');
+ assert.strictEqual(t.db.prepare('SELECT COUNT(*) AS n FROM entitlements').get().n, ents);
+ assert.strictEqual(bal('creator_payable', bob), 349);
+ });
+
+ await check('an order Live credited AFTER the shadow import is settled-in-Live after the final import', async () => {
+ assert.strictEqual(intent(4).status, 'created');
+ const live = new Database(file);
+ live.exec(`UPDATE payment_orders SET status = 'credited' WHERE id = 4;
+ UPDATE users SET openvibe_bucks_balance = 1000 WHERE id = 3;
+ INSERT INTO transactions (from_user_id, to_user_id, amount, type, status) VALUES (NULL, 3, 1000, 'purchase', 'completed');`);
+ live.close();
+ const report = await run();
+ assert.ok(report.counts.intents_updated >= 1, JSON.stringify(report.counts));
+ assert.strictEqual(intent(4).status, 'settled');
+ assert.strictEqual(bal('user_credit', carol), 1000);
+ const r = await tip('pcorder:4', 1300);
+ assert.strictEqual(JSON.parse(result(r)).code, 'billing.intent_settled_in_live');
+ assert.strictEqual(bal('user_credit', carol), 1000);
+ });
+
+ await check('a legacy order Live never credited (a checkout in flight) settles exactly once', async () => {
+ const r = await tip('pcorder:2', 700);
+ assert.strictEqual(JSON.parse(result(r)).effect, 'settled');
+ assert.strictEqual(bal('user_credit', carol), 1500);
+ assert.strictEqual(intent(2).status, 'settled');
+ const again = await t.powerchat({ type: 'donation.completed', streamer: { username: 'openvibe' }, data: { eventId: `ev_cutover_${n}`, appExternalRef: 'pcorder:2', amountUsdCents: 700 } });
+ assert.strictEqual(JSON.parse(result(again)).effect, 'duplicate_receipt');
+ assert.strictEqual(bal('user_credit', carol), 1500);
+ const report = await run();
+ assert.strictEqual(intent(2).status, 'settled', 'a re-import never rewinds what Billing settled');
+ assert.strictEqual(report.new_transactions, 0);
+ assert.deepStrictEqual(report.adjustments, [], 'no adjustment against a Billing-native settlement');
+ assert.strictEqual(bal('user_credit', carol), 1500);
+ t.assertReconciled('after the final import');
+ });
+
+ await check('the host freeze CLI holds webhooks and refuses writes; after "off" the held delivery settles once', async () => {
+ const { spawnSync } = require('child_process');
+ const cli = (...args) => {
+ const r = spawnSync(process.execPath, [path.join(__dirname, '..', 'scripts', 'freeze.js'), ...args], { env: { ...process.env, BILLING_DB_PATH: t.config.dbPath }, encoding: 'utf8' });
+ assert.strictEqual(r.status, 0, r.stderr);
+ return JSON.parse(r.stdout);
+ };
+ assert.strictEqual(cli('on', 'Live cutover').frozen, true);
+ const w = await t.call('POST', '/api/v1/transfers', { body: { from: { type: 'user', id: carol }, to: { type: 'user', id: bob }, amount: 5 } });
+ assert.strictEqual(w.status, 503); assert.strictEqual(w.json.code, 'billing.frozen');
+ const live = new Database(file);
+ live.exec("INSERT INTO payment_orders (id, user_id, provider, kind, amount_cents, bucks, status) VALUES (5, 3, 'powerchat', 'bucks', 700, 500, 'pending')");
+ live.close();
+ await run(); // the final import, while Billing is frozen
+ const held = await tip('pcorder:5', 700);
+ assert.strictEqual(held.status, 202, JSON.stringify(held.json));
+ assert.strictEqual(cli('status').held_webhooks, 1);
+ assert.strictEqual(cli('off').frozen, false);
+ await require('../server/providers').processPending(t.ctx, t.app.locals.adapters); // the service's retry tick
+ assert.strictEqual(JSON.parse(result(held)).effect, 'settled');
+ assert.strictEqual(bal('user_credit', carol), 2000);
+ t.assertReconciled('after the freeze window');
+ });
+
+ await t.close();
+ done();
+})();