openvibers/openvibe.billing Commits

EXTERNAL PowerChat tips reach Tips; hourly reconciliation; /metrics; overflow and race hardening

OP OpenVibers committed 23 Sept 2026 · b774c5c62f · 29 files, +1262 −56

- EXTERNAL receipts: a tip on a streamer's own PowerChat is stored once per payment in external_receipts (never journaled) and, under BILLING_AUTHORITY=billing, announced once as billing.receipt.external (streamer subject, amount_cents, value_bits, donor name or anonymous, message, provider event id, app_purpose for the goal, test). Under BILLING_AUTHORITY=live (the default, shadow) it is recorded only: Live's own webhook announces it, so nothing is celebrated twice. The creator comes from provider_accounts (the importer copies Live's powerchat_connections, --accounts-only runs that step alone; GET|POST /api/v1/admin/provider-accounts); a tip on an unmapped account is held for review and announced when reprocessed. An underpaid direct subscription is an EXTERNAL tip too.
- Reconciliation runs hourly (BILLING_RECONCILE_INTERVAL_MS, also while frozen) and on demand, stored with its trigger; passing scheduled runs are pruned after 30 days. New checks of the ledger against provider receipts: receipts.ledger (clearing moved exactly the receipt's cents), receipts.external (never journaled, announced once), receipts.stale (a receipt unprocessed past BILLING_RECONCILE_STALE_RECEIPT_MIN while open). Exposed at GET /api/v1/admin/reconciliations[/:id|latest], the console and /metrics.
- /metrics with openvibe-shared/metrics (v1.3.0): HTTP golden signals, process, release_info and billing_* gauges (freeze, authority, receipts by state, oldest pending receipt, EXTERNAL receipts, outbox backlog, cashouts, latest reconciliation); loopback only, 404 on the public vhost.
- Overflow: the journal refuses amounts and resulting balances outside the safe-integer range as a whole; every provider receipt path caps at $1,000,000.
- Races: a stored receipt is applied in an IMMEDIATE transaction, so several processes settling it at once wait for the lock instead of failing with "database is locked"; it still settles once.
- test/external.test.js, test/hygiene.test.js (huge amounts, 20 concurrent deliveries, 20 redeliveries, four processes on one receipt, reprocess vs retry job, Network down mid-settlement, a failure inside the settlement transaction, Events down, scheduled reconciliation and /metrics).

Changes

ddiff --git a/.env.example b/.env.example
iindex 3921cb7..e00565f 100644
---- a/.env.example
++++ b/.env.example
@@@ -14,6 +14,13 @@ TRUST_PROXY=1
# SQLite ledger (ADR-007). The systemd unit sets /var/lib/openvibe-billing/billing.db.
BILLING_DB_PATH=./data/billing.db
++# Who keeps the money books: live (default — OpenVibe.Live is the ledger and receives the PowerChat
++# webhook; Billing runs in shadow) or billing (after the cutover, docs/live-cutover.md). Only EXTERNAL
++# tips depend on it today: under billing they are announced as billing.receipt.external for
++# OpenVibe.Tips; under live they are recorded only, because Live's own webhook announces them.
++# Any other value stops Billing at boot.
++#BILLING_AUTHORITY=live
++
# ── Identity: OpenVibe.Network ──────────────────────────────
OV_NETWORK_URL=https://openvibe.network
OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000
@@@ -92,3 +99,8 @@ POWERCHAT_SITE_USERNAME=
#BILLING_JOBS=on
#BILLING_SWEEP_INTERVAL_MS=3600000
#BILLING_WEBHOOK_RETRY_MS=60000
++# Scheduled reconciliation (0 = off); passing scheduled runs older than KEEP_DAYS are pruned.
++#BILLING_RECONCILE_INTERVAL_MS=3600000
++#BILLING_RECONCILE_KEEP_DAYS=30
++# A provider receipt unprocessed this long (economy open) fails reconciliation (receipts.stale).
++#BILLING_RECONCILE_STALE_RECEIPT_MIN=60
ddiff --git a/README.md b/README.md
iindex 717962e..1236d6e 100644
---- a/README.md
++++ b/README.md
@@@ -24,7 +24,7 @@ webhooks land here and nowhere else.
| CREDIT (Vibes bought, spendable, never withdrawable by the buyer) | `user_credit:<subject>` (vibes-bits) |
| MONEY (creator payable, payouts, receipts, platform revenue) | `creator_payable:<subject>`, `payouts_pending:<subject>`, `provider_clearing:<provider>`, `platform_revenue`, `refunds`, `chargeback_loss` |
| ENTITLEMENT (channel subscriptions) | `subscriptions` + `entitlements` (one row per paid period) |
--| EXTERNAL (tips on a streamer's own PowerChat) | stored receipt, no journal entry; a direct-route subscription grants its entitlement only |
++| EXTERNAL (tips on a streamer's own PowerChat) | stored receipt (`external_receipts`), no journal entry; announced as `billing.receipt.external` once Billing is the authority; a direct-route subscription grants its entitlement only |
| LOYALTY, COSMETIC, GAME-STATE, LEGACY | not here (rule 5–8) |
## Run it
@@@ -33,10 +33,11 @@ webhooks land here and nowhere else.
fnm exec --using=22.22.1 npm install
cp .env.example .env # set OV_OAUTH_CLIENT_SECRET, POWERCHAT_WEBHOOK_SECRET, …
npm run dev # http://localhost:4600
--npm test # 10 test files: stub Network/Events/providers, temp DBs, random ports
++npm test # 13 test files: stub Network/Events/providers, temp DBs, random ports
npm run reconcile # reconciliation report (exit 1 on failure)
npm run freeze -- on "reason" # economy freeze from the host (status | on "<reason>" | off)
node scripts/import-live.js --live-db <snapshot> [--dry-run] [--json]
++node scripts/import-live.js --live-db <snapshot> --accounts-only # only PowerChat account → creator mappings
```
Production: `/opt/openvibe.billing`, env `/etc/openvibe/billing.env`, unit
@@@ -66,9 +67,22 @@ Production: `/opt/openvibe.billing`, env `/etc/openvibe/billing.env`, unit
`503 billing.frozen`, reads work, webhooks are stored (202) and processed in arrival order after
unfreeze. Jobs pause.
- **Outbox**: `billing.transaction.settled|reversed`, `billing.entitlement.changed`,
-- `billing.subscription.canceled`, `billing.cashout.requested|paid|denied`, written in the same
-- transaction as the effect (events.event-envelope@1, source `billing`, actor `service:billing`).
-- Relayed to OpenVibe.Events only when `EVENTS_URL` is set.
++ `billing.subscription.canceled`, `billing.cashout.requested|paid|denied`, `billing.staff.action`,
++ `billing.receipt.external`, written in the same transaction as the effect (events.event-envelope@1,
++ source `billing`, actor `service:billing`). Relayed to OpenVibe.Events only when `EVENTS_URL` is set.
++- **Authority** (`BILLING_AUTHORITY`, `live` by default = shadow; `billing` after the cutover): decides
++ whether EXTERNAL tips are announced (see below). Anything else stops Billing at boot.
++- **EXTERNAL receipts** ([server/ops/external.js](server/ops/external.js)): a PowerChat tip paid to the
++ streamer's own account is stored once per payment in `external_receipts` (never in the journal) and,
++ under `billing`, announced once as `billing.receipt.external` — subject `provider_receipt
++ powerchat:<payment id>`, payload `streamer` (SubjectRef), `amount_cents`, `currency`, `value_bits`,
++ `donor_name` (null when `anonymous`), `anonymous`, `message`, `provider`, `provider_event_id` (the
++ payment's id), `delivery_id`, `receiving_account`, `app_ref`/`app_purpose` (goal pick), `occurred_at`,
++ `test`. OpenVibe.Tips turns it into the chat line, alert and goal progress Live's webhook used to
++ produce. The streamer is whoever the receiving account is mapped to in `provider_accounts` (the importer
++ copies Live's `powerchat_connections`; an operator can add one); an unmapped account's tip is held for
++ review and announced when reprocessed after mapping. Under `live` it is recorded, not announced —
++ Live's own webhook announces it, so a tip is never celebrated twice.
## API
@@@ -101,10 +115,13 @@ are not stored. Errors are RFC 9457 problem+json. People are SubjectRefs `{ "typ
| `GET /api/v1/transactions?subject=&cursor=&limit=`, `GET …/:id` | `billing.balance.read` | history (cursor paging), one transaction with `reversed_by` |
| `GET\|POST /api/v1/admin/freeze` | `billing.ledger.admin` | read / set `{on, reason?}`; unfreeze drains queued webhooks |
| `GET /api/v1/admin/reconcile` | `billing.ledger.admin` | run + store a reconciliation |
++| `GET /api/v1/admin/reconciliations[?limit&failed=1]`, `GET …/:id` (`latest`) | `billing.ledger.admin` | stored runs (scheduled and on demand) with their trigger; one full report |
++| `GET\|POST /api/v1/admin/provider-accounts` | `billing.ledger.admin` | list / map `{provider, username, account_id?, subject}` a provider account to its creator (allowed while frozen) |
| `POST /api/v1/admin/adjustments` | `billing.ledger.admin` | `{from: account, to: account, amount, reason, relates_to?}` (same currency) |
| `GET /api/v1/admin/provider-events[?pending=1]`, `POST …/:id/reprocess` | `billing.ledger.admin` | receipts; retry an unprocessed/rejected one |
| `POST /api/v1/admin/sweep`, `GET /api/v1/admin/import-holds` | `billing.ledger.admin` | renewal sweep now; unmapped import users |
| `POST /webhooks/<provider>` | provider signature | `powerchat`, `stripe`, `paypal`, `ccbill` (GET too), `nowpayments` |
++| `GET /metrics` | direct loopback caller | Prometheus text (openvibe-shared/metrics): HTTP golden signals, process, `release_info`, and `billing_*` gauges — freeze, authority, receipts by state, oldest pending receipt, EXTERNAL receipts, outbox backlog, cashouts, the latest reconciliation. 404 to anything a proxy relayed; nginx answers 404 on the public host |
The capabilities and the service manifest are released in openvibe-contracts (v0.8.0;
`billing.staff.action` since v0.17.0); the drafts they came from stay in
@@@ -168,7 +185,7 @@ transaction as their effect. Adjustments are not in the console; they stay on th
| Adapter | Enabled when | Settles | Reverses |
|---|---|---|---|
--| PowerChat | `POWERCHAT_WEBHOOK_SECRET` | `donation.completed` with `pcorder:` (purchase), `pcsub:` (site → share; direct → EXTERNAL entitlement), `pcdon:` (site-routed tip → payable) | `donation.refunded`, `donation.disputed`/`donation.chargeback` (event names to confirm with PowerChat) |
++| PowerChat | `POWERCHAT_WEBHOOK_SECRET` | `donation.completed` with `pcorder:` (purchase), `pcsub:` (site → share; direct → EXTERNAL entitlement), `pcdon:` (site-routed tip → payable); any other tip on a streamer's own account → EXTERNAL receipt (`billing.receipt.external` under `billing`) | `donation.refunded`, `donation.disputed`/`donation.chargeback` (event names to confirm with PowerChat) |
| Stripe | `STRIPE_SECRET_KEY` + `STRIPE_WEBHOOK_SECRET` | `checkout.session.completed` (payment), every `invoice.paid` | `charge.refunded` (cumulative), `charge.dispute.funds_withdrawn` |
| PayPal | client id + secret + `PAYPAL_WEBHOOK_ID` | `PAYMENT.CAPTURE.COMPLETED`, capture after return | `PAYMENT.CAPTURE.REFUNDED`, `…REVERSED` |
| CCBill | `CCBILL_WEBHOOK_SECRET` | `NewSaleSuccess` (reported price required) | `Refund`, `Void`, `Chargeback` |
@@@ -181,12 +198,22 @@ Reversing a subscription payment revokes the periods it granted.
## Operations
--- **Reconciliation** (`npm run reconcile`, `GET /api/v1/admin/reconcile`, or the console's
-- Reconciliation page) checks: journal zero-sum
-- per currency, per-transaction balance, cached balances = entry sums, fx mirror, every settled
++- **Reconciliation** runs **every hour on its own** (`BILLING_RECONCILE_INTERVAL_MS`, a first run a
++ minute after boot, also while frozen — it only reads) and on demand (`npm run reconcile`,
++ `GET /api/v1/admin/reconcile`, the console's Reconciliation page, after every import). It checks: journal
++ zero-sum per currency, per-transaction balance, cached balances = entry sums, fx mirror, every settled
provider event has exactly one transaction, paid cashouts have payout references, payouts_pending =
-- requested cashouts. It also lists negative balances, unprocessed/rejected events, items for review
-- and import holds; totals exclude test transactions. Runs are stored in `reconciliation_runs`.
++ requested cashouts — and the ledger against the provider receipts: every transaction settled from a
++ receipt names its provider and moved exactly the receipt's cents through that provider's clearing
++ account (`receipts.ledger`), EXTERNAL receipts are never in the journal and each announced one has one
++ outbox event (`receipts.external`), and no receipt waits unprocessed longer than
++ `BILLING_RECONCILE_STALE_RECEIPT_MIN` (60) while the economy is open (`receipts.stale`). It also lists
++ negative balances, unprocessed/rejected events, items for review and import holds; totals exclude test
++ transactions. Every run is stored in `reconciliation_runs` with its trigger (`scheduled`, `api`,
++ `console`, `script`, `import`); passing scheduled runs older than `BILLING_RECONCILE_KEEP_DAYS` (30) are
++ pruned, failed ones are kept. Results: the console, `GET /api/v1/admin/reconciliations`, and `/metrics`
++ (`billing_reconciliation_ok`, `…_failed_checks`, `…_last_run_timestamp_seconds`); a failed scheduled run
++ is also logged.
- **Freeze** before any risky change: the console's Freeze page, `POST /api/v1/admin/freeze {"on": true, "reason": "…"}`, or on the
host `node scripts/freeze.js on "<reason>"` (same switch; after `off` the running service processes the
held webhooks on its next retry tick).
@@@ -194,7 +221,10 @@ Reversing a subscription payment revokes the periods it granted.
`warnings` and on the console's Receipts page; resolve them with an adjustment (`relates_to` = the transaction).
- **Payouts** are decided in the staff console (or with `billing.cashout.manage` on the API): pay at the
provider, then approve with its payout reference once the escrow has ended.
++- **Metrics**: `curl -s http://127.0.0.1:4600/metrics` on the host.
- **Backups**: `sqlite3 /var/lib/openvibe-billing/billing.db ".backup billing-$(date +%F).db"`.
++- **Mapping of Live's tables** to Billing's (including payouts, refunds and plans, which have no table
++ of their own on Live): [docs/live-mapping.md](docs/live-mapping.md).
## Cutover runbook (Live → Billing)
@@@ -207,7 +237,8 @@ dashboard). In short:
1. **Shadow import** as often as needed (`node scripts/import-live.js --live-db <snapshot> [--dry-run]`);
map held users in the Network and explain every adjustment until reconciliation is clean.
--2. **Freeze Live money writes**, wait for in-flight PowerChat checkouts (an hour), back up and **freeze Billing**.
++2. **Freeze Live money writes**, wait for in-flight PowerChat checkouts (an hour), back up and **freeze Billing**,
++ set `BILLING_AUTHORITY=billing` in billing.env (EXTERNAL tips are then announced for Tips).
3. **Re-point the PowerChat webhook** to `https://billing.openvibe.network/webhooks/powerchat` (Billing holds
the deliveries while frozen), then take the **final snapshot and import**; reconcile.
4. **Switch Live** (`BILLING_AUTHORITY=billing`, deploy with `--wait-idle`), **unfreeze Billing** (held
@@@ -222,7 +253,7 @@ Billing did itself.
`billing.openvibe.network` left OpenVibe.Sites on 2026-09-23: it serves the staff console only. Billing
is not a public product and is not the money authority until the cutover above and everything in plan
--§12.12 holds: owning runtime with health/readiness ✔ (`/metrics` not yet); canonical identity and
++§12.12 holds: owning runtime with health/readiness and `/metrics` ✔; canonical identity and
scoped service principals ✔ (principal `billing`); server-rendered public routes useful without
JavaScript (not yet — the only pages are the staff-only console); real persistence and end-to-end
workflows ✔ in tests, shadow only in production; capability/event registration against
ddiff --git a/deploy/nginx/billing.openvibe.network.conf b/deploy/nginx/billing.openvibe.network.conf
iindex f27d638..f98f4f1 100644
---- a/deploy/nginx/billing.openvibe.network.conf
++++ b/deploy/nginx/billing.openvibe.network.conf
@@@ -58,6 +58,10 @@ server {
location = /api/health { proxy_pass http://127.0.0.1:4600; }
++ # ── Prometheus metrics: host-local scrapers only (http://127.0.0.1:4600/metrics). Node also
++ # answers 404 to any request a proxy relayed.
++ location = /metrics { return 404; }
++
# The service API: loopback only. Never proxied from the public host.
location ^~ /api/v1 { deny all; }
location ^~ /api/ { return 404; }
ddiff --git a/docs/service-manifest-proposal.json b/docs/service-manifest-proposal.json
iindex 9fab69c..4655950 100644
---- a/docs/service-manifest-proposal.json
++++ b/docs/service-manifest-proposal.json
@@@ -1,5 +1,5 @@
{
-- "version": "0.3.0",
++ "version": "0.4.0",
"contractRanges": {
"openvibe-contracts": ">=0.5.1 <1.0.0"
},
@@@ -31,11 +31,12 @@
"billing.cashout.requested",
"billing.cashout.paid",
"billing.cashout.denied",
-- "billing.staff.action"
++ "billing.staff.action",
++ "billing.receipt.external"
],
"eventsConsumed": [],
"namespacesOwned": [
"billing.*"
],
-- "notes": "The isolated money ledger (ADR-012): balanced append-only journal, provider receipts, subscriptions/entitlements, refunds and payouts, and a staff console (Network SSO, staff only) for payouts, review, reconciliation and the freeze. Proposed manifest: adds billing.staff.action (internal) to the released 0.2.0."
++ "notes": "The isolated money ledger (ADR-012): balanced append-only journal, provider receipts, subscriptions/entitlements, refunds and payouts, and a staff console (Network SSO, staff only) for payouts, review, reconciliation and the freeze. Proposed manifest: adds billing.staff.action (internal; released in contracts v0.17.0) and billing.receipt.external (internal: an EXTERNAL tip on a streamer's own PowerChat — subject provider_receipt <provider>:<payment id>; payload streamer SubjectRef, amount_cents, currency usd-cents, value_bits, donor_name (null when anonymous), anonymous, message, provider, provider_event_id, delivery_id, receipt_ref, receiving_account, app_ref, app_purpose, occurred_at, test, rates — sent only while BILLING_AUTHORITY=billing, consumed by OpenVibe.Tips) to the released 0.2.0."
}
ddiff --git a/package-lock.json b/package-lock.json
iindex 8aa4ffd..e69011e 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -12,7 +12,8 @@
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"express": "^4.21.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.19.0"
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.19.0",
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0"
},
"engines": {
"node": ">=22.0.0"
@@@ -169,6 +170,12 @@
"ieee754": "^1.1.13"
}
},
++ "node_modules/buffer-equal-constant-time": {
++ "version": "1.0.1",
++ "resolved": "https://registry.npmjs.org/buffer-equal-constant-time/-/buffer-equal-constant-time-1.0.1.tgz",
++ "integrity": "sha512-zRpUiDwd/xk6ADqPMATG8vc9VPrkck7T07OIx0gnjmJAnHnTVXNQG3vfvWNuiZIkwu9KrKdA1iJKfsfTVxE6NA==",
++ "license": "BSD-3-Clause"
++ },
"node_modules/bytes": {
"version": "3.1.2",
"resolved": "https://registry.npmjs.org/bytes/-/bytes-3.1.2.tgz",
@@@ -336,6 +343,15 @@
"node": ">= 0.4"
}
},
++ "node_modules/ecdsa-sig-formatter": {
++ "version": "1.0.11",
++ "resolved": "https://registry.npmjs.org/ecdsa-sig-formatter/-/ecdsa-sig-formatter-1.0.11.tgz",
++ "integrity": "sha512-nagl3RYrbNv6kQkeJIpt6NJZy8twLB/2vtz6yN9Z4vRKHN4/QZJIEbqohALSgwKdnksuY3k5Addp5lg8sVoVcQ==",
++ "license": "Apache-2.0",
++ "dependencies": {
++ "safe-buffer": "^5.0.1"
++ }
++ },
"node_modules/ee-first": {
"version": "1.1.1",
"resolved": "https://registry.npmjs.org/ee-first/-/ee-first-1.1.1.tgz",
@@@ -697,6 +713,97 @@
"integrity": "sha512-NM8/P9n3XjXhIZn1lLhkFaACTOURQXjWhV4BA/RnOv8xvgqtqpAX9IO4mRQxSx1Rlo4tqzeqb0sOlruaOy3dug==",
"license": "MIT"
},
++ "node_modules/jsonwebtoken": {
++ "version": "9.0.3",
++ "resolved": "https://registry.npmjs.org/jsonwebtoken/-/jsonwebtoken-9.0.3.tgz",
++ "integrity": "sha512-MT/xP0CrubFRNLNKvxJ2BYfy53Zkm++5bX9dtuPbqAeQpTVe0MQTFhao8+Cp//EmJp244xt6Drw/GVEGCUj40g==",
++ "license": "MIT",
++ "dependencies": {
++ "jws": "^4.0.1",
++ "lodash.includes": "^4.3.0",
++ "lodash.isboolean": "^3.0.3",
++ "lodash.isinteger": "^4.0.4",
++ "lodash.isnumber": "^3.0.3",
++ "lodash.isplainobject": "^4.0.6",
++ "lodash.isstring": "^4.0.1",
++ "lodash.once": "^4.0.0",
++ "ms": "^2.1.1",
++ "semver": "^7.5.4"
++ },
++ "engines": {
++ "node": ">=12",
++ "npm": ">=6"
++ }
++ },
++ "node_modules/jsonwebtoken/node_modules/ms": {
++ "version": "2.1.3",
++ "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz",
++ "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==",
++ "license": "MIT"
++ },
++ "node_modules/jwa": {
++ "version": "2.0.1",
++ "resolved": "https://registry.npmjs.org/jwa/-/jwa-2.0.1.tgz",
++ "integrity": "sha512-hRF04fqJIP8Abbkq5NKGN0Bbr3JxlQ+qhZufXVr0DvujKy93ZCbXZMHDL4EOtodSbCWxOqR8MS1tXA5hwqCXDg==",
++ "license": "MIT",
++ "dependencies": {
++ "buffer-equal-constant-time": "^1.0.1",
++ "ecdsa-sig-formatter": "1.0.11",
++ "safe-buffer": "^5.0.1"
++ }
++ },
++ "node_modules/jws": {
++ "version": "4.0.1",
++ "resolved": "https://registry.npmjs.org/jws/-/jws-4.0.1.tgz",
++ "integrity": "sha512-EKI/M/yqPncGUUh44xz0PxSidXFr/+r0pA70+gIYhjv+et7yxM+s29Y+VGDkovRofQem0fs7Uvf4+YmAdyRduA==",
++ "license": "MIT",
++ "dependencies": {
++ "jwa": "^2.0.1",
++ "safe-buffer": "^5.0.1"
++ }
++ },
++ "node_modules/lodash.includes": {
++ "version": "4.3.0",
++ "resolved": "https://registry.npmjs.org/lodash.includes/-/lodash.includes-4.3.0.tgz",
++ "integrity": "sha512-W3Bx6mdkRTGtlJISOvVD/lbqjTlPPUDTMnlXZFnVwi9NKJ6tiAk6LVdlhZMm17VZisqhKcgzpO5Wz91PCt5b0w==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.isboolean": {
++ "version": "3.0.3",
++ "resolved": "https://registry.npmjs.org/lodash.isboolean/-/lodash.isboolean-3.0.3.tgz",
++ "integrity": "sha512-Bz5mupy2SVbPHURB98VAcw+aHh4vRV5IPNhILUCsOzRmsTmSQ17jIuqopAentWoehktxGd9e/hbIXq980/1QJg==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.isinteger": {
++ "version": "4.0.4",
++ "resolved": "https://registry.npmjs.org/lodash.isinteger/-/lodash.isinteger-4.0.4.tgz",
++ "integrity": "sha512-DBwtEWN2caHQ9/imiNeEA5ys1JoRtRfY3d7V9wkqtbycnAmTvRRmbHKDV4a0EYc678/dia0jrte4tjYwVBaZUA==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.isnumber": {
++ "version": "3.0.3",
++ "resolved": "https://registry.npmjs.org/lodash.isnumber/-/lodash.isnumber-3.0.3.tgz",
++ "integrity": "sha512-QYqzpfwO3/CWf3XP+Z+tkQsfaLL/EnUlXWVkIk5FUPc4sBdTehEqZONuyRt2P67PXAk+NXmTBcc97zw9t1FQrw==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.isplainobject": {
++ "version": "4.0.6",
++ "resolved": "https://registry.npmjs.org/lodash.isplainobject/-/lodash.isplainobject-4.0.6.tgz",
++ "integrity": "sha512-oSXzaWypCMHkPC3NvBEaPHf0KsA5mvPrOPgQWDsbg8n7orZ290M0BmC/jgRZ4vcJ6DTAhjrsSYgdsW/F+MFOBA==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.isstring": {
++ "version": "4.0.1",
++ "resolved": "https://registry.npmjs.org/lodash.isstring/-/lodash.isstring-4.0.1.tgz",
++ "integrity": "sha512-0wJxfxH1wgO3GrbuP+dTTk7op+6L41QCXbGINEmD+ny/G/eCqGzxyCsh7159S+mgDDcoarnBw6PC1PS5+wUGgw==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.once": {
++ "version": "4.1.1",
++ "resolved": "https://registry.npmjs.org/lodash.once/-/lodash.once-4.1.1.tgz",
++ "integrity": "sha512-Sb487aTOCr9drQVL8pIxOzVhafOjZN9UU54hiN8PU3uAiSV7lx1yYNpbNmex2PK6dSJoNTSJUUswT651yww3Mg==",
++ "license": "MIT"
++ },
"node_modules/math-intrinsics": {
"version": "1.1.0",
"resolved": "https://registry.npmjs.org/math-intrinsics/-/math-intrinsics-1.1.0.tgz",
@@@ -875,6 +982,15 @@
"node": ">=20"
}
},
++ "node_modules/openvibe-shared": {
++ "version": "1.3.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
++ "integrity": "sha512-w8yl/5+9eend8I2O/DzAJnEAL2N3GTUY7c7pzK+MLJuLAGD8FmJJQ9gwbgyrq70Snjt3ebrg3TlpORhRAp2kgA==",
++ "license": "MIT",
++ "dependencies": {
++ "jsonwebtoken": "^9.0.0"
++ }
++ },
"node_modules/parseurl": {
"version": "1.3.3",
"resolved": "https://registry.npmjs.org/parseurl/-/parseurl-1.3.3.tgz",
ddiff --git a/package.json b/package.json
iindex 6758711..66562e9 100644
---- a/package.json
++++ b/package.json
@@@ -19,7 +19,8 @@
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"express": "^4.21.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.19.0"
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.19.0",
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0"
},
"engines": {
"node": ">=22.0.0"
ddiff --git a/scripts/import-live.js b/scripts/import-live.js
iindex f4ffe26..92adab7 100644
---- a/scripts/import-live.js
++++ b/scripts/import-live.js
@@@ -4,6 +4,10 @@
* Import OpenVibe.Live's money state into Billing (ADR-012 migration).
*
* node scripts/import-live.js --live-db /path/to/live-snapshot.db [--dry-run] [--json]
++ * node scripts/import-live.js --live-db /path/to/live-snapshot.db --accounts-only [--dry-run] [--json]
++ *
++ * --accounts-only refreshes only provider_accounts from Live's powerchat_connections (who an
++ * EXTERNAL PowerChat tip belongs to); no money, no reconciliation. Safe to run daily after cutover.
*
* --live-db must be a COPY of live.db (e.g. `sqlite3 live.db ".backup live-snapshot.db"`); it is
* opened read-only. Live user ids are resolved to Network subjects through resolve-batch with this
@@@ -17,7 +21,7 @@ const { loadConfig } = require('../server/config');
const { openDb } = require('../server/db');
const { createRates } = require('../server/rates');
const { createIdentity } = require('../server/network');
--const { importLive } = require('../server/importer/live');
++const { importLive, importProviderAccounts } = require('../server/importer/live');
const { reconcile } = require('../server/reconcile');
const args = process.argv.slice(2);
@@@ -33,8 +37,20 @@ async function main() {
const db = openDb(config.dbPath);
const ctx = { db, config, rates: createRates(config.rates), now: () => Date.now(), log: console };
const identity = createIdentity(config);
++ if (flag('accounts-only')) {
++ const r = await importProviderAccounts(ctx, { live, resolveLiveUsers: identity.resolveLiveUsers, dryRun: flag('dry-run') });
++ const a = r.provider_accounts;
++ if (flag('json')) console.log(JSON.stringify(r, null, 2));
++ else {
++ console.log(`provider accounts ${r.run_id}${r.dry_run ? ' (DRY RUN — nothing kept)' : ''}: ${a.mapped} mapped, ${a.unchanged} unchanged, ${a.kept_admin.length} kept (set by an operator)`);
++ for (const u of a.unmapped) console.log(` unmapped: live user ${u.live_user_id} (${u.username}): ${u.reason}`);
++ }
++ live.close();
++ db.close();
++ return;
++ }
const report = await importLive(ctx, { live, resolveLiveUsers: identity.resolveLiveUsers, dryRun: flag('dry-run') });
-- const rec = flag('dry-run') ? null : reconcile(ctx);
++ const rec = flag('dry-run') ? null : reconcile(ctx, { trigger: 'import' });
if (flag('json')) console.log(JSON.stringify({ import: report, reconciliation: rec }, null, 2));
else {
console.log(`import ${report.run_id}${report.dry_run ? ' (DRY RUN — nothing kept)' : ''}`);
@@@ -46,6 +62,8 @@ async function main() {
for (const h of report.holds) console.log(` live user ${h.live_user_id} (${h.username || '?'}): credit ${h.credit_bits}, payable ${h.payable_bits}`);
console.log(` duplicate provider refs: ${report.duplicate_provider_refs.length}; route markers parsed: ${report.route_markers}`);
console.log(` unreplayable history rows: ${report.unreplayable.length}; anomalies: ${report.anomalies.length}`);
++ const pa = report.provider_accounts;
++ if (pa) console.log(` PowerChat accounts: ${pa.mapped} mapped, ${pa.unchanged} unchanged, ${pa.unmapped.length} unmapped, ${pa.kept_admin.length} kept (operator)`);
if (rec) console.log(` reconciliation ${rec.id}: ${rec.ok ? 'OK' : 'FAILED'} (${rec.checks.filter((c) => !c.ok).map((c) => c.id).join(', ') || 'all checks pass'})`);
}
live.close();
ddiff --git a/scripts/reconcile.js b/scripts/reconcile.js
iindex 438dff2..822671c 100644
---- a/scripts/reconcile.js
++++ b/scripts/reconcile.js
@@@ -11,7 +11,7 @@ const { reconcile } = require('../server/reconcile');
const config = loadConfig();
const db = openDb(config.dbPath);
--const report = reconcile({ db, config, rates: createRates(config.rates), now: () => Date.now() });
++const report = reconcile({ db, config, rates: createRates(config.rates), now: () => Date.now() }, { trigger: 'script' });
if (process.argv.includes('--json')) console.log(JSON.stringify(report, null, 2));
else {
console.log(`reconciliation ${report.id}: ${report.ok ? 'OK' : 'FAILED'}`);
@@@ -19,6 +19,7 @@ else {
const w = report.warnings;
console.log(` warnings: ${w.negative_balances.length} negative balances, ${w.unprocessed_events.length} unprocessed events, ${w.rejected_events.length} rejected events, ` +
`${w.events_for_review.length + w.transactions_for_review.length} for review, ${w.import_holds.length} import holds`);
++ console.log(` EXTERNAL receipts (test excluded): ${report.totals.external_receipts}, announced ${report.totals.external_announced}`);
console.log(` totals (test excluded): ${JSON.stringify(report.totals)}`);
}
db.close();
ddiff --git a/server/api/v1.js b/server/api/v1.js
iindex f74acd2..30eb5cc 100644
---- a/server/api/v1.js
++++ b/server/api/v1.js
@@@ -22,6 +22,7 @@ const intents = require('../ops/intents');
const admin = require('../ops/admin');
const providers = require('../providers');
const { reconcile } = require('../reconcile');
++const external = require('../ops/external');
const CAP = {
intent: 'billing.intent.create',
@@@ -248,9 +249,40 @@ function v1Router({ ctx, auth, adapters }) {
res.json({ ...state, processed_after_unfreeze: drained });
}));
r.get('/admin/reconcile', ...read(CAP.admin, (req, res) => {
-- const report = reconcile(ctx);
++ const report = reconcile(ctx, { trigger: 'api' });
res.status(200).json(report);
}));
++ // Stored runs (scheduled and on demand), newest first; ?failed=1 for failed runs only.
++ r.get('/admin/reconciliations', ...read(CAP.admin, (req, res) => {
++ const limit = Math.min(200, positiveInt(req.query.limit || 20, 'limit'));
++ const failed = req.query.failed === '1' || req.query.failed === 'true';
++ const rows = db.prepare(`SELECT id, started_at, finished_at, ok, report FROM reconciliation_runs ${failed ? 'WHERE ok = 0' : ''} ORDER BY finished_at DESC, id DESC LIMIT ?`).all(limit);
++ res.json({
++ runs: rows.map((r) => {
++ const rep = JSON.parse(r.report);
++ return { id: r.id, ok: !!r.ok, trigger: rep.trigger || null, started_at: r.started_at, finished_at: r.finished_at, failed_checks: rep.checks.filter((c) => !c.ok).map((c) => c.id) };
++ }),
++ });
++ }));
++ r.get('/admin/reconciliations/:id', ...read(CAP.admin, (req, res) => {
++ const row = req.params.id === 'latest'
++ ? db.prepare('SELECT report FROM reconciliation_runs ORDER BY finished_at DESC, id DESC LIMIT 1').get()
++ : db.prepare('SELECT report FROM reconciliation_runs WHERE id = ?').get(req.params.id);
++ if (!row) fail(404, 'billing.reconciliation_not_found', `no reconciliation run ${req.params.id}`);
++ res.json(JSON.parse(row.report));
++ }));
++ // Provider account → creator (who an EXTERNAL tip on that account belongs to).
++ r.get('/admin/provider-accounts', ...read(CAP.admin, (req, res) => {
++ res.json({ accounts: db.prepare('SELECT * FROM provider_accounts ORDER BY provider, username').all() });
++ }));
++ // Not a money movement: allowed while frozen (the cutover maps accounts with Billing frozen).
++ r.post('/admin/provider-accounts', auth.needs(CAP.admin), idem, wrap((req, res) => {
++ const b = req.body || {};
++ const provider = String(b.provider || '').toLowerCase();
++ if (!adapters[provider]) fail(422, 'billing.invalid_input', `provider must be one of ${Object.keys(adapters).join(', ')}`);
++ const row = external.mapAccount(ctx, { provider, username: b.username, accountId: b.account_id, subject: b.subject, source: 'admin' });
++ res.status(201).json({ account: row });
++ }));
r.post('/admin/adjustments', ...write(CAP.admin, (req, res) => {
const b = req.body || {};
const out = admin.adjust(ctx, { from: b.from, to: b.to, amount: b.amount, reason: b.reason, relatesTo: b.relates_to, idempotencyKey: req.idempotencyKey, actor: actorOf(req) });
ddiff --git a/server/app.js b/server/app.js
iindex f0ec385..230a870 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -5,14 +5,18 @@
* listens, tests build their own instance.
*
* GET /api/health, /api/ready liveness / readiness
++ * GET /metrics Prometheus text (direct loopback callers only; server/metrics.js)
* /api/v1/* operations API (service tokens, see api/v1.js)
* /webhooks/<provider> provider receipts (see api/webhooks.js)
* /, /auth/*, /cashouts, … staff console (Network SSO, server-rendered; see console/index.js)
*
* createApp({ config, db, keys, identity, adapters, now, fetchImpl, log }) — everything injectable.
*/
++const path = require('path');
const express = require('express');
const { http } = require('openvibe-contracts');
++const { instrument } = require('openvibe-shared/metrics');
++const { createRelease } = require('openvibe-shared/release');
const { loadConfig } = require('./config');
const { openDb } = require('./db');
const { createRates } = require('./rates');
@@@ -23,6 +27,7 @@ const { webhooksRouter } = require('./api/webhooks');
const { consoleRouter } = require('./console');
const providers = require('./providers');
const { isFrozen } = require('./ops/common');
++const { createMetrics } = require('./metrics');
const VERSION = require('../package.json').version;
@@@ -43,11 +48,17 @@ function createApp(opts = {}) {
// route /API/v1/… to the same API, walking past the public-host deny.
app.set('case sensitive routing', true);
app.set('trust proxy', config.trustProxy);
++ // HTTP golden signals by route template, process metrics, release_info and Billing's gauges;
++ // GET /metrics before any other route (loopback only — relayed requests get 404). After the
++ // routing settings above: the first app.use() fixes the router's case sensitivity.
++ const metrics = createMetrics({ db, config, now: ctx.now });
++ const release = createRelease({ service: 'billing', root: path.join(__dirname, '..') });
++ instrument(app, { service: 'billing', release: release.release, registry: metrics.registry });
app.use(http.middleware());
app.use((req, res, next) => { res.setHeader('Cache-Control', 'no-store'); next(); });
app.get('/api/health', (req, res) => res.json({
-- ok: true, service: 'billing', version: VERSION, frozen: isFrozen(db),
++ ok: true, service: 'billing', version: VERSION, frozen: isFrozen(db), authority: config.authority,
providers: Object.fromEntries(Object.values(adapters).map((a) => [a.name, a.enabled])),
events_relay: !!config.events.url,
}));
@@@ -77,6 +88,7 @@ function createApp(opts = {}) {
app.locals.ctx = ctx;
app.locals.adapters = adapters;
app.locals.keys = keys;
++ app.locals.metrics = metrics.registry;
return app;
}
ddiff --git a/server/config.js b/server/config.js
iindex 3a29754..5d6c2ab 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -45,6 +45,16 @@ function parseStaff(raw) {
return { subjects: [...new Set(items.filter((s) => USER_SUBJECT_RE.test(s)))], invalid: items.filter((s) => !USER_SUBJECT_RE.test(s)) };
}
++// Who keeps the money books (docs/live-cutover.md): 'live' while OpenVibe.Live is the ledger and
++// receives the PowerChat webhook (Billing runs in shadow), 'billing' once the webhook points here.
++// Anything else is refused at boot — a typo must not silently change what Billing announces.
++const AUTHORITIES = ['live', 'billing'];
++function parseAuthority(raw) {
++ const v = String(raw == null || raw === '' ? 'live' : raw).trim().toLowerCase();
++ if (!AUTHORITIES.includes(v)) throw new Error(`BILLING_AUTHORITY must be one of ${AUTHORITIES.join(', ')} (got "${raw}")`);
++ return v;
++}
++
function loadConfig(env = process.env) {
const nodeEnv = env.NODE_ENV || 'development';
const isProduction = nodeEnv === 'production';
@@@ -60,6 +70,10 @@ function loadConfig(env = process.env) {
baseUrl,
trustProxy: env.TRUST_PROXY != null ? Number(env.TRUST_PROXY) : 1,
dbPath: env.BILLING_DB_PATH || './data/billing.db',
++ // 'live' (default, shadow): EXTERNAL PowerChat tips are recorded but not announced, because
++ // Live announces them from its own webhook. 'billing': they are announced as
++ // billing.receipt.external (OpenVibe.Tips turns that into the chat line, alert and goal).
++ authority: parseAuthority(env.BILLING_AUTHORITY),
// Identity: service tokens are RS256 JWTs signed by OpenVibe.Network.
network: {
@@@ -132,6 +146,15 @@ function loadConfig(env = process.env) {
enabled: env.BILLING_JOBS !== 'off',
sweepIntervalMs: int(env.BILLING_SWEEP_INTERVAL_MS, 60 * 60 * 1000),
webhookRetryMs: int(env.BILLING_WEBHOOK_RETRY_MS, 60 * 1000),
++ // Scheduled reconciliation (0 turns it off): stored like any other run; passing scheduled
++ // runs older than reconcileKeepDays are pruned, failed ones are kept.
++ reconcileIntervalMs: int(env.BILLING_RECONCILE_INTERVAL_MS, 60 * 60 * 1000),
++ reconcileKeepDays: int(env.BILLING_RECONCILE_KEEP_DAYS, 30),
++ },
++ reconcile: {
++ // A provider receipt still unprocessed this long after it arrived (economy not frozen)
++ // fails reconciliation: the provider was paid and the ledger does not show it yet.
++ staleReceiptMin: int(env.BILLING_RECONCILE_STALE_RECEIPT_MIN, 60),
},
// Staff console (server/console): Network SSO (authorization code + PKCE S256, OAuth
@@@ -154,4 +177,4 @@ function loadConfig(env = process.env) {
};
}
--module.exports = { loadConfig, DEFAULT_PRICE_TIERS, USER_SUBJECT_RE };
++module.exports = { loadConfig, DEFAULT_PRICE_TIERS, USER_SUBJECT_RE, AUTHORITIES };
ddiff --git a/server/console/index.js b/server/console/index.js
iindex ff2f033..c7a51ff 100644
---- a/server/console/index.js
++++ b/server/console/index.js
@@@ -311,7 +311,7 @@ function consoleRouter({ ctx, adapters, keys, fetchImpl = globalThis.fetch }) {
});
r.post('/reconciliation', needs(CAP.admin), (req, res) => {
const report = db.transaction(() => {
-- const rep = reconcile(ctx);
++ const rep = reconcile(ctx, { trigger: 'console' });
record(req, { action: 'reconciliation.run', target: { type: 'reconciliation_run', id: rep.id }, detail: { ok: rep.ok, failed: rep.checks.filter((c) => !c.ok).map((c) => c.id) } });
return rep;
})();
ddiff --git a/server/console/queries.js b/server/console/queries.js
iindex efdfb2d..fdf203f 100644
---- a/server/console/queries.js
++++ b/server/console/queries.js
@@@ -30,7 +30,7 @@ function summarizeReport(report) {
return count(d.offenders) + count(d.mismatches);
};
return {
-- id: report.id, ok: !!report.ok, started_at: report.started_at, finished_at: report.finished_at,
++ id: report.id, ok: !!report.ok, started_at: report.started_at, finished_at: report.finished_at, trigger: report.trigger || null,
checks: (report.checks || []).map((c) => ({ id: c.id, ok: !!c.ok, offenders: offenders(c) })),
warnings: Object.fromEntries(Object.entries(report.warnings || {}).map(([k, v]) => [k, count(v)])),
totals: report.totals || {},
@@@ -49,7 +49,7 @@ function reconciliationRuns(db, limit = 50) {
return db.prepare('SELECT id, finished_at, ok, report FROM reconciliation_runs ORDER BY finished_at DESC, id DESC LIMIT ?').all(limit)
.map((r) => {
const s = summarizeReport(JSON.parse(r.report));
-- return { id: r.id, finished_at: r.finished_at, ok: !!r.ok, failed: s.checks.filter((c) => !c.ok).map((c) => c.id) };
++ return { id: r.id, finished_at: r.finished_at, ok: !!r.ok, trigger: s.trigger, failed: s.checks.filter((c) => !c.ok).map((c) => c.id) };
});
}
ddiff --git a/server/console/views.js b/server/console/views.js
iindex 4456e37..3f33548 100644
---- a/server/console/views.js
++++ b/server/console/views.js
@@@ -192,7 +192,7 @@ function eventsTable(rows, { csrf, reprocess, frozen }) {
return html`<div class="tablewrap"><table><thead><tr><th>#</th><th>Provider</th><th>Event</th><th>Type</th><th>Received</th><th>Result</th><th>Attempts</th>${reprocess ? html`<th></th>` : ''}</tr></thead><tbody>
${rows.map((e) => html`<tr><td>${e.id}</td><td>${e.provider}</td><td class="mono">${e.provider_event_id}</td><td>${e.type || '—'}</td><td>${when(e.received_at)}</td>
<td>${e.result ? html`${e.result.effect}${e.result.code ? html` · <code>${e.result.code}</code>` : ''}${e.result.reason ? html`<br><span class="muted">${e.result.reason}</span>` : ''}` : html`<span class="muted">not processed</span>${e.last_error ? html`<br><span class="muted">${e.last_error}</span>` : ''}`}</td>
--<td>${e.attempts}</td>${reprocess ? html`<td>${!e.processed_at || (e.result && e.result.effect === 'rejected') ? html`<form class="inline" method="post" action="/receipts/${e.id}/reprocess"><input type="hidden" name="_csrf" value="${csrf}"><button type="submit" ${frozen ? html`disabled` : ''}>Reprocess</button></form>` : ''}</td>` : ''}</tr>`)}
++<td>${e.attempts}</td>${reprocess ? html`<td>${!e.processed_at || (e.result && (e.result.effect === 'rejected' || (e.result.effect === 'external' && e.result.review))) ? html`<form class="inline" method="post" action="/receipts/${e.id}/reprocess"><input type="hidden" name="_csrf" value="${csrf}"><button type="submit" ${frozen ? html`disabled` : ''}>Reprocess</button></form>` : ''}</td>` : ''}</tr>`)}
</tbody></table></div>`;
}
@@@ -227,10 +227,11 @@ ${rows.map((h) => html`<tr><td>${h.live_user_id}</td><td><code>${h.owner}</code>
function reconciliation({ staff, csrf, frozen, runs, notice }) {
return layout({
title: 'Reconciliation', section: 'reconciliation', staff, csrf, frozen, notice, body: html`<h1>Reconciliation</h1>
++<p class="muted">Also runs on its own (hourly by default, <code>BILLING_RECONCILE_INTERVAL_MS</code>). Passing scheduled runs are kept for <code>BILLING_RECONCILE_KEEP_DAYS</code> (30); failed and manual runs are kept.</p>
<form method="post" action="/reconciliation"><input type="hidden" name="_csrf" value="${csrf}"><button class="primary" type="submit">Run reconciliation now</button></form>
<h2>History</h2>
--${runs.length ? html`<div class="tablewrap"><table><thead><tr><th>Run</th><th>Finished</th><th>Result</th><th>Failed checks</th></tr></thead><tbody>
--${runs.map((r) => html`<tr><td><a class="mono" href="/reconciliation/${r.id}">${r.id}</a></td><td>${when(r.finished_at)}</td><td>${okBad(r.ok)}</td><td>${r.failed.join(', ') || '—'}</td></tr>`)}
++${runs.length ? html`<div class="tablewrap"><table><thead><tr><th>Run</th><th>Finished</th><th>Trigger</th><th>Result</th><th>Failed checks</th></tr></thead><tbody>
++${runs.map((r) => html`<tr><td><a class="mono" href="/reconciliation/${r.id}">${r.id}</a></td><td>${when(r.finished_at)}</td><td>${r.trigger || '—'}</td><td>${okBad(r.ok)}</td><td>${r.failed.join(', ') || '—'}</td></tr>`)}
</tbody></table></div>` : html`<div class="card muted">No runs yet.</div>`}`,
});
}
ddiff --git a/server/db.js b/server/db.js
iindex 9df9aa9..5015ebb 100644
---- a/server/db.js
++++ b/server/db.js
@@@ -103,6 +103,41 @@ BEGIN SELECT RAISE(ABORT, 'provider receipts are immutable'); END;
CREATE TRIGGER IF NOT EXISTS provider_events_no_delete BEFORE DELETE ON provider_events
BEGIN SELECT RAISE(ABORT, 'provider receipts are immutable'); END;
++-- EXTERNAL receipts (ADR-012): tips paid on a streamer's OWN provider account. No journal entry,
++-- no liability — one row per provider payment (receipt_ref), whatever delivery carried it, so the
++-- announcement (billing.receipt.external) happens at most once. status says whether it was
++-- announced and, if not, why (Live still the authority, an account no creator has connected).
++CREATE TABLE IF NOT EXISTS external_receipts (
++ receipt_ref TEXT PRIMARY KEY,
++ provider TEXT NOT NULL,
++ provider_event INTEGER NOT NULL REFERENCES provider_events (id),
++ receiving_account TEXT,
++ streamer_subject TEXT,
++ amount_cents INTEGER NOT NULL,
++ test INTEGER NOT NULL DEFAULT 0 CHECK (test IN (0, 1)),
++ status TEXT NOT NULL CHECK (status IN ('announced', 'not_announced')),
++ reason TEXT,
++ event_id TEXT,
++ created_at TEXT NOT NULL,
++ updated_at TEXT NOT NULL
++);
++CREATE INDEX IF NOT EXISTS external_receipts_streamer ON external_receipts (streamer_subject);
++
++-- Which creator a provider account belongs to (a PowerChat username / account id → Network subject):
++-- the importer copies Live's powerchat_connections; an operator can add or correct one.
++CREATE TABLE IF NOT EXISTS provider_accounts (
++ provider TEXT NOT NULL,
++ username TEXT NOT NULL,
++ account_id TEXT,
++ subject TEXT NOT NULL,
++ source TEXT NOT NULL,
++ live_user_id INTEGER,
++ created_at TEXT NOT NULL,
++ updated_at TEXT NOT NULL,
++ PRIMARY KEY (provider, username)
++);
++CREATE INDEX IF NOT EXISTS provider_accounts_account ON provider_accounts (provider, account_id);
++
CREATE TABLE IF NOT EXISTS payment_intents (
id TEXT PRIMARY KEY,
provider TEXT NOT NULL,
ddiff --git a/server/importer/live.js b/server/importer/live.js
iindex cea0d9d..4fc4637 100644
---- a/server/importer/live.js
++++ b/server/importer/live.js
@@@ -22,6 +22,12 @@
* 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.
++ * 6. accounts powerchat_connections → provider_accounts (PowerChat username / account id →
++ * the creator's subject), so an EXTERNAL tip on a streamer's own PowerChat is
++ * attributed to them (ops/external.js). A mapping an operator set (source admin) is
++ * never overwritten; a connection whose Live user has no subject is reported, not held
++ * (no money sits on it). importProviderAccounts() runs this step alone
++ * (scripts/import-live.js --accounts-only), e.g. daily after the cutover.
*
* Idempotent: every write is keyed (import:live:txn:<id>, legacy ids, target-state keys), so a
* re-run over the same snapshot changes nothing, and a later snapshot (the final import at
@@@ -30,6 +36,7 @@
*/
const { post, balance, iso, prefixedId } = require('../ledger');
const { A, entry } = require('../ops/common');
++const external = require('../ops/external');
const ROUTE_MARKER = /^(direct|site)(:|$)/;
@@@ -45,6 +52,58 @@ function columns(db, table) { return tableExists(db, table) ? db.prepare(`PRAGMA
class DryRun extends Error {}
++/** Live's PowerChat connections with the Network subject of their owner (resolved here when `map` lacks it). */
++async function readProviderAccounts(live, resolveLiveUsers, map = new Map()) {
++ if (!tableExists(live, 'powerchat_connections')) return [];
++ const hasId = columns(live, 'powerchat_connections').includes('powerchat_user_id');
++ const rows = live.prepare(`SELECT user_id, powerchat_username${hasId ? ', powerchat_user_id' : ''} FROM powerchat_connections
++ WHERE powerchat_username IS NOT NULL AND TRIM(powerchat_username) <> '' ORDER BY user_id`).all();
++ const missing = [...new Set(rows.map((r) => String(r.user_id)).filter((id) => !map.has(id)))];
++ const more = missing.length ? await resolveLiveUsers(missing) : new Map();
++ return rows.map((r) => ({
++ live_user_id: r.user_id, username: String(r.powerchat_username).trim().toLowerCase(),
++ account_id: hasId && r.powerchat_user_id != null ? String(r.powerchat_user_id) : null,
++ subject: map.get(String(r.user_id)) || more.get(String(r.user_id)) || null,
++ }));
++}
++
++/** Inside the import transaction: upsert the mappings; fills report.provider_accounts. */
++function applyProviderAccounts(ctx, rows, report) {
++ const out = { mapped: 0, unchanged: 0, unmapped: [], kept_admin: [] };
++ for (const r of rows) {
++ if (!r.subject) { out.unmapped.push({ live_user_id: r.live_user_id, username: r.username, reason: 'no Network subject' }); continue; }
++ if (!external.USERNAME_RE.test(r.username)) { out.unmapped.push({ live_user_id: r.live_user_id, username: r.username.slice(0, 80), reason: 'not a PowerChat username' }); continue; }
++ const cur = ctx.db.prepare("SELECT * FROM provider_accounts WHERE provider = 'powerchat' AND username = ?").get(r.username);
++ if (cur && cur.source === 'admin') { if (cur.subject !== r.subject) out.kept_admin.push({ username: r.username, admin_subject: cur.subject, live_subject: r.subject }); continue; }
++ if (cur && cur.subject === r.subject && (cur.account_id || null) === (r.account_id || cur.account_id || null)) { out.unchanged++; continue; }
++ external.mapAccount(ctx, { provider: 'powerchat', username: r.username, accountId: r.account_id, subject: r.subject, source: 'live-import', liveUserId: r.live_user_id });
++ out.mapped++;
++ }
++ report.provider_accounts = out;
++ return out;
++}
++
++/** Step 6 alone: refresh provider_accounts from a Live snapshot (no money is touched). */
++async function importProviderAccounts(ctx, { live, resolveLiveUsers, dryRun = false, log = console }) {
++ const { db } = ctx;
++ const started = ctx.now();
++ const runId = prefixedId('imp', started);
++ const report = { run_id: runId, dry_run: !!dryRun, started_at: iso(started), source: 'openvibe-live snapshot (provider accounts only)' };
++ const rows = await readProviderAccounts(live, resolveLiveUsers);
++ try {
++ db.transaction(() => {
++ applyProviderAccounts(ctx, rows, report);
++ if (dryRun) throw new DryRun();
++ })();
++ } catch (e) { if (!(e instanceof DryRun)) throw e; }
++ report.finished_at = iso(ctx.now());
++ db.prepare('INSERT INTO import_runs (id, source, dry_run, started_at, finished_at, report) VALUES (?, ?, ?, ?, ?, ?)')
++ .run(runId, 'live-accounts', dryRun ? 1 : 0, report.started_at, report.finished_at, JSON.stringify(report));
++ const a = report.provider_accounts;
++ log.log && log.log(`[Billing] provider accounts ${runId}${dryRun ? ' (dry run, rolled back)' : ''}: ${a.mapped} mapped, ${a.unchanged} unchanged, ${a.unmapped.length} unmapped, ${a.kept_admin.length} kept (operator)`);
++ return report;
++}
++
async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = console }) {
const { db, rates } = ctx;
const started = ctx.now();
@@@ -88,6 +147,7 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c
const owner = (liveId) => (liveId == null ? null : subjectOf(liveId) || `hold:live:${liveId}`);
const username = new Map(users.map((u) => [String(u.id), u.username]));
report.counts.identities = { resolved: [...ids].filter((i) => subjectOf(i)).length, unmapped: [...ids].filter((i) => !subjectOf(i)).length };
++ const accountRows = await readProviderAccounts(live, resolveLiveUsers, map);
const before = db.prepare('SELECT COUNT(*) AS n FROM transactions').get().n;
const ms = ctx.now();
@@@ -307,6 +367,9 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c
}
}
++ // ── 6. provider accounts ─────────────────────────────
++ applyProviderAccounts(ctx, accountRows, report);
++
report.new_transactions = db.prepare('SELECT COUNT(*) AS n FROM transactions').get().n - before;
if (dryRun) throw new DryRun();
};
@@@ -322,4 +385,4 @@ async function importLive(ctx, { live, resolveLiveUsers, dryRun = false, log = c
return report;
}
--module.exports = { importLive, liveTs };
++module.exports = { importLive, importProviderAccounts, liveTs };
ddiff --git a/server/index.js b/server/index.js
iindex 95762fd..6c0644a 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -5,8 +5,11 @@
* Listens on PORT (4600) behind nginx (billing.openvibe.network); see deploy/.
*
* Background jobs (BILLING_JOBS=off disables them): stored-but-unprocessed provider events are
-- * retried every minute; the subscription renewal sweep runs hourly; the outbox relay publishes to
-- * OpenVibe.Events when EVENTS_URL is set. Nothing runs while the economy is frozen except the relay.
++ * retried every minute; the subscription renewal sweep runs hourly; reconciliation runs hourly
++ * (BILLING_RECONCILE_INTERVAL_MS, 0 = off) and is stored in reconciliation_runs (API
++ * /api/v1/admin/reconciliations, the console, /metrics); the outbox relay publishes to
++ * OpenVibe.Events when EVENTS_URL is set. While the economy is frozen only the relay and the
++ * reconciliation (it only reads the books) run.
*/
const { loadConfig } = require('./config');
const { createApp } = require('./app');
@@@ -14,6 +17,7 @@ const providers = require('./providers');
const subscriptions = require('./ops/subscriptions');
const { createRelay } = require('./outbox');
const { isFrozen } = require('./ops/common');
++const { runScheduled } = require('./reconcile');
const config = loadConfig();
const app = createApp({ config });
@@@ -30,6 +34,14 @@ if (config.jobs.enabled) {
const out = subscriptions.sweep(ctx);
if (out.renewed.length || out.expired.length || out.canceled.length) console.log(`[Billing] subscription sweep: ${out.renewed.length} renewed, ${out.expired.length} expired, ${out.canceled.length} canceled`);
});
++ if (config.jobs.reconcileIntervalMs > 0) {
++ const reconcileNow = () => runScheduled(ctx, { keepDays: config.jobs.reconcileKeepDays });
++ every(config.jobs.reconcileIntervalMs, reconcileNow);
++ // A first run shortly after boot, so /metrics and the console have a fresh result.
++ const first = setTimeout(() => { try { reconcileNow(); } catch (e) { console.warn('[Billing] job:', e.message); } }, 60_000);
++ first.unref();
++ timers.push(first);
++ }
relay.start();
}
@@@ -37,12 +49,13 @@ const server = app.listen(config.port, config.host, () => {
const enabled = Object.values(adapters).filter((a) => a.enabled).map((a) => a.name);
console.log(`[Billing] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (db ${config.dbPath})`);
console.log(`[Billing] providers enabled: ${enabled.length ? enabled.join(', ') : 'none'}; economy ${isFrozen(ctx.db) ? 'FROZEN' : 'open'}; events relay ${config.events.url ? `→ ${config.events.url}` : 'off (outbox accumulates)'}`);
++ console.log(`[Billing] money authority: ${config.authority}${config.authority === 'live' ? ' (shadow: EXTERNAL tips are recorded, not announced)' : ' (EXTERNAL tips are announced as billing.receipt.external)'}`);
});
server.keepAliveTimeout = 65_000;
function shutdown(signal) {
console.log(`[Billing] ${signal} — closing`);
-- timers.forEach(clearInterval);
++ timers.forEach((t) => { clearInterval(t); clearTimeout(t); });
relay.stop();
keys.stop();
server.close(() => { try { ctx.db.close(); } catch { /* */ } process.exit(0); });
ddiff --git a/server/ledger.js b/server/ledger.js
iindex d1a0171..4a612d1 100644
---- a/server/ledger.js
++++ b/server/ledger.js
@@@ -15,7 +15,9 @@
* - entries are never updated or deleted (db triggers); corrections are reversing transactions;
* - account_balances changes only here, in the same SQLite transaction as the entries;
* - a transaction's idempotency_key is unique: posting the same key again returns the
-- * original transaction and moves nothing.
++ * original transaction and moves nothing;
++ * - every amount and every resulting balance is a safe integer (|n| <= 2^53 - 1): past that JS
++ * arithmetic silently rounds and SQLite would store a REAL, so such a posting is refused whole.
*
* Callers run post() inside db.transaction() together with their funds checks, so a check and
* the movement it guards are one atomic unit.
@@@ -90,6 +92,7 @@ function post(ctx, spec) {
const sums = new Map();
for (const e of spec.entries || []) {
if (!Number.isInteger(e.amount)) throw new BillingError(500, 'ledger.bad_amount', `non-integer amount ${e.amount}`);
++ if (!Number.isSafeInteger(e.amount)) throw new BillingError(422, 'billing.amount_overflow', `amount ${e.amount} is beyond the exact integer range`);
if (e.amount === 0) continue;
const key = `${e.kind}|${e.owner == null ? '' : e.owner}|${e.currency}`;
const m = merged.get(key) || { kind: e.kind, owner: e.owner == null ? null : e.owner, currency: e.currency, amount: 0 };
@@@ -98,6 +101,7 @@ function post(ctx, spec) {
sums.set(e.currency, (sums.get(e.currency) || 0) + e.amount);
}
for (const [currency, sum] of sums) {
++ if (!Number.isSafeInteger(sum)) throw new BillingError(422, 'billing.amount_overflow', `entries in ${currency} overflow the exact integer range`);
if (sum !== 0) throw new BillingError(500, 'ledger.unbalanced', `entries sum to ${sum} ${currency}, not zero`, { entries: spec.entries });
}
@@@ -112,6 +116,13 @@ function post(ctx, spec) {
);
const insEntry = db.prepare('INSERT INTO ledger_entries (txn_id, account_id, amount) VALUES (?, ?, ?)');
const bump = db.prepare('UPDATE account_balances SET balance = balance + ?, updated_at = ? WHERE account_id = ?');
++ for (const m of merged.values()) {
++ if (!Number.isSafeInteger(m.amount)) throw new BillingError(422, 'billing.amount_overflow', `the ${m.kind} entry overflows the exact integer range`);
++ if (m.amount === 0) continue;
++ if (!Number.isSafeInteger(balance(db, m) + m.amount)) {
++ throw new BillingError(422, 'billing.amount_overflow', `${m.kind}${m.owner ? `:${m.owner}` : ''} would leave the exact integer range`);
++ }
++ }
for (const m of merged.values()) {
if (m.amount === 0) continue;
const acct = accountId(db, m, nowIso);
ddiff --git a/server/metrics.js b/server/metrics.js
nnew file mode 100644
iindex 0000000..f43678b
---- /dev/null
++++ b/server/metrics.js
@@@ -0,0 +1,71 @@
++'use strict';
++
++/**
++ * Prometheus metrics (roadmap Track O): the openvibe-shared registry that app.js instruments with the
++ * HTTP golden signals, process metrics and release_info, plus Billing's own gauges, read from the
++ * database at scrape time. GET /metrics answers direct loopback callers only (a request nginx relayed
++ * carries X-Real-IP and gets 404). No gauge carries a subject, an amount per person or a provider
++ * payload — only counts and platform totals.
++ *
++ * billing_frozen 1 while the economy is frozen
++ * billing_authority{authority} 1 for the configured BILLING_AUTHORITY (live | billing)
++ * billing_provider_events{provider,state} stored receipts: pending | settled | external |
++ * rejected | review | other
++ * billing_provider_event_oldest_pending_seconds age of the oldest unprocessed receipt (0 when none)
++ * billing_external_receipts{status} EXTERNAL receipts: announced | not_announced
++ * billing_outbox_pending events not yet relayed to OpenVibe.Events
++ * billing_outbox_failing unsent events whose last relay attempt failed
++ * billing_cashouts{status} requested | paid | denied
++ * billing_reconciliation_ok 1 when the latest stored run passed, 0 when it failed
++ * billing_reconciliation_last_run_timestamp_seconds when the latest stored run finished
++ * billing_reconciliation_failed_checks failed checks in the latest stored run
++ */
++const { createRegistry } = require('openvibe-shared/metrics');
++const { isFrozen } = require('./ops/common');
++const { latest } = require('./reconcile');
++
++function createMetrics({ db, config, now = () => Date.now() }) {
++ const registry = createRegistry();
++ const one = (sql, ...a) => db.prepare(sql).get(...a).n;
++
++ registry.gauge({ name: 'billing_frozen', help: 'Whether the economy is frozen (1) or open (0)', collect: () => (isFrozen(db) ? 1 : 0) });
++ registry.gauge({
++ name: 'billing_authority', help: 'The configured money authority (BILLING_AUTHORITY); the value is always 1', labelNames: ['authority'],
++ collect: () => [{ labels: { authority: config.authority }, value: 1 }],
++ });
++ registry.gauge({
++ name: 'billing_provider_events', help: 'Stored provider receipts by provider and processing state', labelNames: ['provider', 'state'],
++ collect: () => db.prepare(`SELECT provider, CASE
++ WHEN processed_at IS NULL THEN 'pending'
++ WHEN json_extract(result, '$.effect') = 'rejected' THEN 'rejected'
++ WHEN json_extract(result, '$.review') = 1 THEN 'review'
++ WHEN json_extract(result, '$.effect') IN ('settled', 'external') THEN json_extract(result, '$.effect')
++ ELSE 'other' END AS state, COUNT(*) AS n
++ FROM provider_events GROUP BY 1, 2`).all().map((r) => ({ labels: { provider: r.provider, state: r.state }, value: r.n })),
++ });
++ registry.gauge({
++ name: 'billing_provider_event_oldest_pending_seconds', help: 'Age of the oldest stored-but-unprocessed provider receipt (0 when none)',
++ collect: () => {
++ const r = db.prepare('SELECT MIN(received_at) AS at FROM provider_events WHERE processed_at IS NULL').get();
++ return r && r.at ? Math.max(0, Math.round((now() - Date.parse(r.at)) / 1000)) : 0;
++ },
++ });
++ registry.gauge({
++ name: 'billing_external_receipts', help: 'EXTERNAL receipts (tips on a streamer\'s own provider account) by announcement status', labelNames: ['status'],
++ collect: () => ['announced', 'not_announced'].map((status) => ({ labels: { status }, value: one('SELECT COUNT(*) AS n FROM external_receipts WHERE status = ?', status) })),
++ });
++ registry.gauge({ name: 'billing_outbox_pending', help: 'Outbox events not yet relayed to OpenVibe.Events', collect: () => one('SELECT COUNT(*) AS n FROM outbox WHERE sent_at IS NULL') });
++ registry.gauge({ name: 'billing_outbox_failing', help: 'Unsent outbox events whose last relay attempt failed', collect: () => one('SELECT COUNT(*) AS n FROM outbox WHERE sent_at IS NULL AND last_error IS NOT NULL') });
++ registry.gauge({
++ name: 'billing_cashouts', help: 'Cashouts by status', labelNames: ['status'],
++ collect: () => ['requested', 'paid', 'denied'].map((status) => ({ labels: { status }, value: one('SELECT COUNT(*) AS n FROM cashouts WHERE status = ?', status) })),
++ });
++ // The latest stored reconciliation (scheduled or on demand). No run yet: the gauges are left out.
++ registry.gauge({ name: 'billing_reconciliation_ok', help: 'Whether the latest stored reconciliation passed (1) or failed (0)', collect: () => { const r = latest(db); return r ? (r.ok ? 1 : 0) : undefined; } });
++ registry.gauge({ name: 'billing_reconciliation_last_run_timestamp_seconds', help: 'When the latest stored reconciliation finished (Unix seconds)', collect: () => { const r = latest(db); return r ? Math.round(Date.parse(r.finished_at) / 1000) : undefined; } });
++ registry.gauge({ name: 'billing_reconciliation_failed_checks', help: 'Failed checks in the latest stored reconciliation', collect: () => { const r = latest(db); return r ? r.checks.filter((c) => !c.ok).length : undefined; } });
++
++ return { registry };
++}
++
++module.exports = { createMetrics };
ddiff --git a/server/ops/common.js b/server/ops/common.js
iindex 5df46e1..f8c26ac 100644
---- a/server/ops/common.js
++++ b/server/ops/common.js
@@@ -9,6 +9,9 @@ const { BillingError } = require('../ledger');
const BITS = 'vibes-bits';
const CENTS = 'usd-cents';
++// The largest single provider receipt Billing accepts ($1,000,000). Anything above is refused
++// (a provider event is then rejected and listed for review), never booked.
++const MAX_RECEIPT_CENTS = 100_000_000;
/** Account addresses. owner is a subject id, a provider slug, 'hold:live:<id>' or null (platform). */
const A = {
@@@ -55,9 +58,10 @@ function userSubject(v, field = 'subject') {
return ref.id;
}
++/** A positive safe integer (never a float, never past 2^53 where JS arithmetic stops being exact). */
function positiveInt(v, field, max) {
const n = Number(v);
-- if (!Number.isInteger(n) || n <= 0) fail(422, 'billing.invalid_amount', `${field} must be a positive integer`);
++ if (!Number.isSafeInteger(n) || n <= 0) fail(422, 'billing.invalid_amount', `${field} must be a positive integer`);
if (max && n > max) fail(422, 'billing.invalid_amount', `${field} exceeds the maximum of ${max}`);
return n;
}
@@@ -79,4 +83,4 @@ function assertNotFrozen(ctx) {
/** Movement summary for a subject from a transaction's entries (used in API responses). */
function subjectRef(id) { return id ? { type: 'user', id } : null; }
--module.exports = { A, BITS, CENTS, entry, receiptEntries, fail, userSubject, positiveInt, text, isFrozen, assertNotFrozen, subjectRef };
++module.exports = { A, BITS, CENTS, MAX_RECEIPT_CENTS, entry, receiptEntries, fail, userSubject, positiveInt, text, isFrozen, assertNotFrozen, subjectRef };
ddiff --git a/server/ops/external.js b/server/ops/external.js
nnew file mode 100644
iindex 0000000..2848879
---- /dev/null
++++ b/server/ops/external.js
@@@ -0,0 +1,137 @@
++'use strict';
++
++/**
++ * EXTERNAL receipts (ADR-012): a tip paid on a streamer's OWN provider account (PowerChat route
++ * `direct`). The money never touches OpenVibe, so there is no journal entry and no liability — but
++ * the tip still has to be celebrated where the streamer is live. Until the cutover OpenVibe.Live did
++ * that from its own PowerChat webhook; once the webhook points here, Billing announces it:
++ *
++ * billing.receipt.external subject { type: 'provider_receipt', id: '<provider>:<payment id>' }
++ * payload {
++ * classification: 'EXTERNAL', provider, receipt_ref, provider_event_id (the provider's id of the
++ * payment, stable across redeliveries), delivery_id (the webhook delivery that carried it),
++ * streamer: SubjectRef, receiving_account: { provider, id, username },
++ * amount_cents, currency: 'usd-cents', value_bits (at Billing's value rate — what Live counted
++ * toward goals), donor_name (null when anonymous), anonymous, message, app_ref, app_purpose,
++ * occurred_at, test, rates
++ * }
++ *
++ * OpenVibe.Tips consumes it (the chat line through Live's /internal/tips/deliveries, the overlay
++ * alert, goal progress). Written to the outbox in the same SQLite transaction that marks the
++ * provider event processed.
++ *
++ * Announced at most once per payment: external_receipts is keyed by receipt_ref, so a redelivery
++ * under a new delivery id is 'duplicate_receipt'. Announced only when
++ * - BILLING_AUTHORITY=billing. While Live is the authority it receives the webhook and announces
++ * the tip itself; a delivery that reaches Billing then (a test, a manual resend) is recorded
++ * 'not_announced' so nothing is celebrated twice;
++ * - the receiving account belongs to a known creator (provider_accounts). Otherwise the receipt is
++ * recorded for review; map the account and reprocess the event to announce it.
++ */
++const { iso } = require('../ledger');
++const { enqueue } = require('../outbox');
++const { MAX_RECEIPT_CENTS, fail, userSubject } = require('./common');
++
++const EVENT_TYPE = 'billing.receipt.external';
++
++/** The creator a provider account belongs to, by account id, then by (lower-case) username. */
++function accountSubject(db, provider, { id, username } = {}) {
++ if (id != null && id !== '') {
++ const r = db.prepare('SELECT subject FROM provider_accounts WHERE provider = ? AND account_id = ?').get(provider, String(id));
++ if (r) return r.subject;
++ }
++ const name = String(username || '').trim().toLowerCase();
++ if (!name) return null;
++ const r = db.prepare('SELECT subject FROM provider_accounts WHERE provider = ? AND username = ?').get(provider, name);
++ return r ? r.subject : null;
++}
++
++const USERNAME_RE = /^[a-z0-9_.-]{1,64}$/;
++
++/** Add or correct a mapping (operator or importer). Returns the stored row. */
++function mapAccount(ctx, { provider, username, accountId, subject, source, liveUserId }) {
++ const { db } = ctx;
++ const name = String(username || '').trim().toLowerCase();
++ if (!USERNAME_RE.test(name)) fail(422, 'billing.invalid_input', 'username must be the provider account name (letters, digits, _ . -)');
++ const owner = userSubject(subject, 'subject');
++ const at = iso(ctx.now());
++ db.prepare(`INSERT INTO provider_accounts (provider, username, account_id, subject, source, live_user_id, created_at, updated_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?)
++ ON CONFLICT (provider, username) DO UPDATE SET account_id = COALESCE(excluded.account_id, account_id), subject = excluded.subject,
++ source = excluded.source, live_user_id = COALESCE(excluded.live_user_id, live_user_id), updated_at = excluded.updated_at`)
++ .run(provider, name, accountId != null && accountId !== '' ? String(accountId).slice(0, 100) : null, owner, source, liveUserId || null, at, at);
++ return db.prepare('SELECT * FROM provider_accounts WHERE provider = ? AND username = ?').get(provider, name);
++}
++
++/**
++ * Inside the provider-event transaction. input: { provider, receiptRef, providerEventId, deliveryId,
++ * sourceEventId, account: { id, username }, streamer?, amountCents, donorName, anonymous, message,
++ * appRef, appPurpose, occurredAt, test }. Returns the provider event result.
++ */
++function record(ctx, input) {
++ const { db, config, rates } = ctx;
++ const cents = Number(input.amountCents);
++ if (cents === 0) return { effect: 'none', reason: 'EXTERNAL: a zero-amount tip on the streamer\'s own PowerChat — nothing to record' };
++ if (!Number.isSafeInteger(cents) || cents < 1 || cents > MAX_RECEIPT_CENTS) {
++ return { effect: 'none', review: true, reason: `EXTERNAL tip with an amount outside 1..${MAX_RECEIPT_CENTS} cents (${String(input.amountCents).slice(0, 40)}) — not recorded, held for review` };
++ }
++ const prev = db.prepare('SELECT * FROM external_receipts WHERE receipt_ref = ?').get(input.receiptRef);
++ if (prev && prev.status === 'announced') {
++ return { effect: 'duplicate_receipt', external: true, announced: false, reason: `payment ${input.receiptRef} was already announced (provider event ${prev.provider_event})` };
++ }
++ const username = String((input.account && input.account.username) || '').trim().toLowerCase() || null;
++ // The creator: named by the caller (a direct-subscription intent already verified the account),
++ // else whoever the receiving account is mapped to.
++ const streamer = input.streamer || accountSubject(db, input.provider, input.account || {});
++ let reason = null;
++ let review = false;
++ if (config.authority !== 'billing') {
++ reason = 'EXTERNAL: a tip on the streamer\'s own PowerChat — recorded, not announced: Live is the money authority (BILLING_AUTHORITY=live) and announces it from its own webhook';
++ } else if (!streamer) {
++ review = true;
++ reason = `EXTERNAL: a tip on PowerChat account "${username || 'unknown'}", which no OpenVibe creator has connected — recorded, not announced; map the account (POST /api/v1/admin/provider-accounts) and reprocess`;
++ }
++
++ const at = iso(ctx.now());
++ let env = null;
++ if (!reason) {
++ const anonymous = !!input.anonymous;
++ env = enqueue(ctx, {
++ event_type: EVENT_TYPE,
++ subject: { type: 'provider_receipt', id: input.receiptRef },
++ payload: {
++ classification: 'EXTERNAL',
++ provider: input.provider,
++ receipt_ref: input.receiptRef,
++ provider_event_id: input.providerEventId,
++ delivery_id: input.deliveryId,
++ streamer: { type: 'user', id: streamer },
++ receiving_account: { provider: input.provider, id: input.account && input.account.id != null ? String(input.account.id) : null, username },
++ amount_cents: cents,
++ currency: 'usd-cents',
++ value_bits: rates.bitsForValueCents(cents),
++ donor_name: anonymous ? null : (input.donorName ? String(input.donorName).slice(0, 80) : null),
++ anonymous,
++ message: input.message ? String(input.message).slice(0, 500) : null,
++ app_ref: input.appRef || null,
++ app_purpose: input.appPurpose || null,
++ occurred_at: input.occurredAt || null,
++ test: !!input.test,
++ rates: rates.snapshot(),
++ },
++ });
++ }
++ const status = env ? 'announced' : 'not_announced';
++ if (prev) {
++ db.prepare('UPDATE external_receipts SET status = ?, reason = ?, event_id = ?, streamer_subject = ?, updated_at = ? WHERE receipt_ref = ?')
++ .run(status, reason, env ? env.event_id : null, streamer, at, input.receiptRef);
++ } else {
++ db.prepare(`INSERT INTO external_receipts (receipt_ref, provider, provider_event, receiving_account, streamer_subject, amount_cents, test, status, reason, event_id, created_at, updated_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
++ .run(input.receiptRef, input.provider, input.sourceEventId, username, streamer, cents, input.test ? 1 : 0, status, reason, env ? env.event_id : null, at, at);
++ }
++ if (env) return { effect: 'external', announced: true, event_id: env.event_id, streamer };
++ return { effect: 'external', announced: false, reason, review: review || undefined };
++}
++
++module.exports = { record, mapAccount, accountSubject, EVENT_TYPE, USERNAME_RE };
ddiff --git a/server/ops/purchases.js b/server/ops/purchases.js
iindex 8ca6759..4da513a 100644
---- a/server/ops/purchases.js
++++ b/server/ops/purchases.js
@@@ -13,7 +13,7 @@
*/
const { post, getTxn, BillingError, present } = require('../ledger');
const { enqueue } = require('../outbox');
--const { A, receiptEntries, fail, positiveInt } = require('./common');
++const { A, MAX_RECEIPT_CENTS, receiptEntries, fail, positiveInt } = require('./common');
const intents = require('./intents');
function settle(ctx, input) {
@@@ -24,7 +24,7 @@ function settle(ctx, input) {
const dup = db.prepare('SELECT id FROM transactions WHERE receipt_ref = ?').get(receiptRef);
if (dup) return { txn: getTxn(db, dup.id), replay: true, duplicateReceipt: true };
-- const paidCents = positiveInt(input.paidCents, 'amount_cents', 100_000_000);
++ const paidCents = positiveInt(input.paidCents, 'amount_cents', MAX_RECEIPT_CENTS);
let intent = null;
if (input.intentId) {
intent = intents.find(db, input.intentId);
ddiff --git a/server/ops/subscriptions.js b/server/ops/subscriptions.js
iindex 70c1056..db7420a 100644
---- a/server/ops/subscriptions.js
++++ b/server/ops/subscriptions.js
@@@ -18,7 +18,7 @@
*/
const { post, getTxn, requireFunds, iso, prefixedId } = require('../ledger');
const { enqueue } = require('../outbox');
--const { A, entry, receiptEntries, fail, positiveInt } = require('./common');
++const { A, MAX_RECEIPT_CENTS, entry, receiptEntries, fail, positiveInt } = require('./common');
const { summary } = require('./purchases');
const intents = require('./intents');
@@@ -103,7 +103,7 @@ function pay(ctx, input) {
} else if (input.source === 'receipt' && r) {
provider = r.provider;
route = r.route || null;
-- const paid = positiveInt(r.paidCents, 'amount_cents', 100_000_000);
++ const paid = positiveInt(r.paidCents, 'amount_cents', MAX_RECEIPT_CENTS);
const fee = Math.max(0, Math.round(Number(r.feeCents) || 0));
if (route === 'direct') {
// EXTERNAL: the streamer holds the money already; nothing touches the journal.
ddiff --git a/server/ops/transfers.js b/server/ops/transfers.js
iindex b231c3d..6447b37 100644
---- a/server/ops/transfers.js
++++ b/server/ops/transfers.js
@@@ -13,7 +13,7 @@
const { validate } = require('openvibe-contracts');
const { post, getTxn, requireFunds } = require('../ledger');
const { enqueue } = require('../outbox');
--const { A, entry, receiptEntries, fail, positiveInt, text } = require('./common');
++const { A, MAX_RECEIPT_CENTS, entry, receiptEntries, fail, positiveInt, text } = require('./common');
const { summary } = require('./purchases');
const KINDS = ['tip', 'donation', 'paid_interaction'];
@@@ -85,7 +85,7 @@ function fromReceipt(ctx, input) {
return db.transaction(() => {
const dup = db.prepare('SELECT id FROM transactions WHERE receipt_ref = ?').get(input.receiptRef);
if (dup) return { txn: getTxn(db, dup.id), replay: true, duplicateReceipt: true };
-- const paidCents = positiveInt(input.paidCents, 'amount_cents');
++ const paidCents = positiveInt(input.paidCents, 'amount_cents', MAX_RECEIPT_CENTS);
if (input.from && input.from === input.to) fail(422, 'billing.self_dealing', 'a creator cannot route a tip to themselves');
const bits = rates.bitsForValueCents(paidCents);
const { txn } = post(ctx, {
ddiff --git a/server/providers/index.js b/server/providers/index.js
iindex b049b8a..ac4617c 100644
---- a/server/providers/index.js
++++ b/server/providers/index.js
@@@ -13,9 +13,16 @@
* arrival order after the freeze lifts (and on a timer for events that failed transiently).
*
* Result effects: settled (exactly one transaction points at the event) | duplicate_receipt (the
-- * payment was already settled by another event) | updated (state change, no money) | none (no
-- * money effect: test, echo, EXTERNAL, not handled; `review` marks ones an operator should see) |
++ * payment was already settled — or, EXTERNAL, already announced — by another event) | updated
++ * (state change, no money) | external (a tip on the streamer's own account: recorded in
++ * external_receipts, no journal entry; `announced` says whether billing.receipt.external went out) |
++ * none (no money effect: test, echo, not handled; `review` marks ones an operator should see) |
* rejected (a 4xx business refusal, e.g. a refund for an unknown payment).
++ *
++ * Races: the stored row is the lock. Applying a plan re-reads processed_at inside an IMMEDIATE
++ * transaction (the write lock is taken first, so a second process or connection waits rather than
++ * failing to upgrade), and a payment's receipt_ref is UNIQUE in the journal, so concurrent deliveries
++ * of one payment — same delivery id or not, same process or not — settle it once.
*/
const crypto = require('crypto');
const { iso, BillingError } = require('../ledger');
@@@ -24,6 +31,7 @@ const purchases = require('../ops/purchases');
const transfers = require('../ops/transfers');
const subscriptions = require('../ops/subscriptions');
const reversals = require('../ops/reversals');
++const external = require('../ops/external');
const { createPowerchat } = require('./powerchat');
const { createStripe } = require('./stripe');
const { createPaypal } = require('./paypal');
@@@ -62,6 +70,7 @@ function applyPlan(ctx, plan, row) {
case 'purchase': return settled(purchases.settle(ctx, withEvent(plan.args)));
case 'subscription': return settled(subscriptions.pay(ctx, withEvent(plan.args)));
case 'tip': return settled(transfers.fromReceipt(ctx, withEvent(plan.args)));
++ case 'external': return external.record(ctx, withEvent(plan.args));
case 'reverse': {
const r = reversals.reverseReceipt(ctx, withEvent(plan.args));
if (r.noop) return { effect: 'none', reason: r.noop };
@@@ -93,7 +102,7 @@ async function process(ctx, adapters, rowOrId) {
}
db.prepare('UPDATE provider_events SET processed_at = ?, result = ?, attempts = attempts + 1, last_error = NULL WHERE id = ?')
.run(iso(ctx.now()), JSON.stringify(result), row.id);
-- })();
++ }).immediate();
} catch (e) {
db.prepare('UPDATE provider_events SET attempts = attempts + 1, last_error = ? WHERE id = ?').run(String(e.message).slice(0, 500), row.id);
(ctx.log || console).warn(`[Billing] ${row.provider} event ${row.provider_event_id} not processed: ${e.message}`);
@@@ -119,7 +128,10 @@ async function processPending(ctx, adapters, { limit = 500 } = {}) {
async function reprocess(ctx, adapters, id) {
const row = parseRow(ctx.db.prepare('SELECT * FROM provider_events WHERE id = ?').get(id));
if (!row) throw new BillingError(404, 'billing.event_not_found', `no provider event ${id}`);
-- if (row.processed_at && row.result && row.result.effect !== 'rejected' && row.result.effect !== 'none') {
++ // An EXTERNAL receipt that was not announced (e.g. its account was mapped since) may be retried;
++ // one that was announced never is (external_receipts would answer duplicate_receipt anyway).
++ const retryable = (r) => r.effect === 'rejected' || r.effect === 'none' || (r.effect === 'external' && !r.announced);
++ if (row.processed_at && row.result && !retryable(row.result)) {
throw new BillingError(409, 'billing.event_settled', `event ${id} already had effect ${row.result.effect}`);
}
if (row.processed_at) ctx.db.prepare('UPDATE provider_events SET processed_at = NULL WHERE id = ?').run(id);
ddiff --git a/server/providers/powerchat.js b/server/providers/powerchat.js
iindex ed83e85..1f6b77b 100644
---- a/server/providers/powerchat.js
++++ b/server/providers/powerchat.js
@@@ -20,8 +20,10 @@
* (legacy links carry Live user ids; they are resolved to subjects)
* (pcorder, site pcsub and pcdon count only when the tip was paid to POWERCHAT_SITE_USERNAME:
* the ref is buyer-editable in the fallback tip link; from any other account they are held)
-- * anything else a tip on a streamer's own PowerChat → EXTERNAL, no Billing effect
-- * (recorded as an interaction by OpenVibe.Tips).
++ * anything else a tip on a streamer's own PowerChat → EXTERNAL: no journal entry;
++ * stored in external_receipts and, once Billing is the authority,
++ * announced as billing.receipt.external (ops/external.js) for
++ * OpenVibe.Tips to celebrate (chat line, alert, goal).
* Test deliveries (isTest, or source manual_test) move no money and are ignored unless
* POWERCHAT_ALLOW_TEST_FULFILLMENT is on, in which case they settle flagged test. App-sourced
* echoes of our own forwarded tips (source developer_app) are ignored.
@@@ -112,6 +114,15 @@ function createPowerchat(cfg, { network } = {}) {
? `${what} ${ref} was paid to PowerChat account "${host || 'unknown'}", not the site account — nothing credited, held for review`
: `${what} ${ref}: POWERCHAT_SITE_USERNAME is not set, so the receiving account cannot be verified — nothing credited, held for review`,
});
++ // EXTERNAL: a tip on the streamer's own account. Recorded (never booked) and announced once
++ // Billing is the authority — see ops/external.js.
++ const external = () => ({
++ provider: 'powerchat', receiptRef, providerEventId: String(paymentId), deliveryId: row.provider_event_id,
++ account: { id: envelope.streamer && envelope.streamer.id != null ? envelope.streamer.id : null, username: host || null },
++ amountCents: cents, donorName: data.donorName || null, anonymous: !!(data.isAnonymous || data.anonymous),
++ message: data.message || null, appRef: ref || null, appPurpose: data.appPurpose ? String(data.appPurpose).slice(0, 120) : null,
++ occurredAt: data.occurredAt || data.createdAt || null, test: isTest,
++ });
let m;
if ((m = ref.match(/^pcorder:(.+)$/))) {
if (!onSite) return offSite('purchase checkout');
@@@ -139,7 +150,7 @@ function createPowerchat(cfg, { network } = {}) {
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' };
++ if (route === 'direct') return { effect: 'external', args: { ...external(), streamer: intent.streamer_subject } };
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 {
@@@ -164,7 +175,7 @@ function createPowerchat(cfg, { network } = {}) {
// recorded here and listed by reconciliation instead of vanishing.
return { effect: 'none', reason: 'unattributed tip to the site PowerChat account — review', review: true };
}
-- return { effect: 'none', reason: 'EXTERNAL: a tip on the streamer\'s own PowerChat — no Billing liability' };
++ return { effect: 'external', args: external() };
}
return { name: 'powerchat', enabled, verify, parse, interpret };
ddiff --git a/server/reconcile.js b/server/reconcile.js
iindex 95559c3..bbff7d1 100644
---- a/server/reconcile.js
++++ b/server/reconcile.js
@@@ -13,13 +13,29 @@
* transaction pointing at it, and every other event has none
* payouts.references every paid cashout has a payout reference and a settling transaction
* payouts.pending payouts_pending per creator equals their requested cashouts
++ * The ledger against the provider receipts it rests on:
++ * receipts.ledger every transaction settled from a provider event names that provider and a
++ * receipt reference, and moved exactly the receipt's cents through that
++ * provider's clearing account (−paid for a settlement, +cents for a reversal)
++ * receipts.external EXTERNAL receipts are never in the journal, and each announced one has
++ * exactly one billing.receipt.external in the outbox
++ * receipts.stale no provider receipt waits unprocessed longer than
++ * BILLING_RECONCILE_STALE_RECEIPT_MIN while the economy is open (a provider,
++ * Network or code failure is holding money the ledger does not show)
* Warnings (reported, do not fail): negative user balances, unprocessed/rejected events, events and
-- * transactions flagged for review (chargebacks on donated credit, unattributed site tips), import
-- * holds. Totals exclude test transactions (ADR-012 rule 9).
++ * transactions flagged for review (chargebacks on donated credit, unattributed site tips, EXTERNAL
++ * tips on unmapped accounts), import holds. Totals exclude test transactions (ADR-012 rule 9).
++ *
++ * Runs on demand (API, console, scripts/reconcile.js, after an import) and on a schedule
++ * (runScheduled, server/index.js); every run is stored in reconciliation_runs with its trigger.
*/
const { iso, prefixedId } = require('./ledger');
++const { isFrozen } = require('./ops/common');
++
++const SETTLING = ['purchase', 'donation', 'subscription'];
++const REVERSING = ['refund', 'chargeback'];
--function reconcile(ctx, { store = true } = {}) {
++function reconcile(ctx, { store = true, trigger = 'manual' } = {}) {
const { db } = ctx;
const started = ctx.now();
const checks = [];
@@@ -64,6 +80,36 @@ function reconcile(ctx, { store = true } = {}) {
FROM accounts a JOIN account_balances b ON b.account_id = a.id WHERE a.kind = 'payouts_pending'`).all().filter((r) => r.balance !== r.requested);
check('payouts.pending', !pendingBad.length, pendingBad.length ? { offenders: pendingBad } : null);
++ const rcpt = db.prepare(`SELECT t.id, t.type, t.provider, t.receipt_ref, p.id AS event, p.provider AS event_provider,
++ json_extract(t.metadata, '$.paid_cents') AS paid_cents, json_extract(t.metadata, '$.cents') AS reversed_cents,
++ (SELECT COUNT(*) FROM ledger_entries e WHERE e.txn_id = t.id) AS entries,
++ COALESCE((SELECT SUM(e.amount) FROM ledger_entries e JOIN accounts a ON a.id = e.account_id
++ WHERE e.txn_id = t.id AND a.kind = 'provider_clearing' AND a.owner_subject = p.provider), 0) AS clearing
++ FROM transactions t JOIN provider_events p ON p.id = t.source_event_id`).all();
++ const rcptBad = rcpt.filter((r) => {
++ if (r.provider !== r.event_provider) return true;
++ if (SETTLING.includes(r.type)) return !r.receipt_ref || (r.entries > 0 && r.clearing !== -Number(r.paid_cents));
++ if (REVERSING.includes(r.type)) return r.entries > 0 && r.clearing !== Number(r.reversed_cents);
++ return false;
++ });
++ check('receipts.ledger', !rcptBad.length, rcptBad.length ? { offenders: rcptBad } : { transactions: rcpt.length });
++
++ const extBooked = db.prepare('SELECT x.receipt_ref, t.id AS txn_id FROM external_receipts x JOIN transactions t ON t.receipt_ref = x.receipt_ref').all();
++ const extUnsent = db.prepare(`SELECT x.receipt_ref, x.event_id FROM external_receipts x
++ WHERE x.status = 'announced' AND (x.event_id IS NULL OR NOT EXISTS (SELECT 1 FROM outbox o WHERE o.event_id = x.event_id))`).all();
++ const extTwice = db.prepare(`SELECT json_extract(event, '$.subject.id') AS receipt_ref, COUNT(*) AS events FROM outbox
++ WHERE json_extract(event, '$.event_type') = 'billing.receipt.external' GROUP BY 1 HAVING COUNT(*) > 1`).all();
++ const extOffenders = [...extBooked.map((r) => ({ ...r, problem: 'booked in the journal' })), ...extUnsent.map((r) => ({ ...r, problem: 'announced without an outbox event' })),
++ ...extTwice.map((r) => ({ ...r, problem: 'announced more than once' }))];
++ check('receipts.external', !extOffenders.length, extOffenders.length ? { offenders: extOffenders }
++ : { receipts: db.prepare('SELECT COUNT(*) AS n FROM external_receipts').get().n });
++
++ const frozen = isFrozen(db);
++ const staleMin = (ctx.config && ctx.config.reconcile && ctx.config.reconcile.staleReceiptMin) || 60;
++ const stale = frozen ? [] : db.prepare('SELECT id, provider, provider_event_id, type, received_at, attempts, last_error FROM provider_events WHERE processed_at IS NULL AND received_at < ? ORDER BY id LIMIT 100')
++ .all(iso(started - staleMin * 60_000));
++ check('receipts.stale', !stale.length, stale.length ? { offenders: stale, older_than_min: staleMin } : { older_than_min: staleMin, ...(frozen ? { frozen: true } : {}) });
++
const warnings = {
negative_balances: db.prepare(`SELECT a.kind, a.owner_subject AS owner, a.currency, b.balance FROM accounts a JOIN account_balances b ON b.account_id = a.id
WHERE a.kind IN ('user_credit', 'creator_payable', 'payouts_pending') AND b.balance < 0`).all(),
@@@ -90,12 +136,15 @@ function reconcile(ctx, { store = true } = {}) {
payable_outstanding_bits: sumKind('creator_payable', 'vibes-bits', false),
payouts_pending_bits: sumKind('payouts_pending', 'vibes-bits', false),
import_adjustment_bits: sumKind('import_adjustment', 'vibes-bits', false),
++ external_receipts: db.prepare('SELECT COUNT(*) AS n FROM external_receipts WHERE test = 0').get().n,
++ external_announced: db.prepare("SELECT COUNT(*) AS n FROM external_receipts WHERE test = 0 AND status = 'announced'").get().n,
++ external_cents: db.prepare('SELECT COALESCE(SUM(amount_cents), 0) AS n FROM external_receipts WHERE test = 0').get().n,
transactions: db.prepare('SELECT COUNT(*) AS n FROM transactions').get().n,
test_transactions: db.prepare('SELECT COUNT(*) AS n FROM transactions WHERE test = 1').get().n,
};
const finished = ctx.now();
-- const report = { id: prefixedId('rec', finished), ok: checks.every((c) => c.ok), started_at: iso(started), finished_at: iso(finished), checks, warnings, totals };
++ const report = { id: prefixedId('rec', finished), ok: checks.every((c) => c.ok), trigger, started_at: iso(started), finished_at: iso(finished), checks, warnings, totals };
if (store) {
db.prepare('INSERT INTO reconciliation_runs (id, started_at, finished_at, ok, report) VALUES (?, ?, ?, ?, ?)')
.run(report.id, report.started_at, report.finished_at, report.ok ? 1 : 0, JSON.stringify(report));
@@@ -103,4 +152,26 @@ function reconcile(ctx, { store = true } = {}) {
return report;
}
--module.exports = { reconcile };
++/**
++ * The scheduled run (server/index.js, every BILLING_RECONCILE_INTERVAL_MS; it only reads the books, so
++ * it also runs while the economy is frozen). Stored like any run; passing scheduled runs older than
++ * keepDays are pruned so the table does not grow without bound — failed runs and on-demand runs stay.
++ */
++function runScheduled(ctx, { keepDays = 30 } = {}) {
++ const report = reconcile(ctx, { trigger: 'scheduled' });
++ if (keepDays > 0) {
++ ctx.db.prepare("DELETE FROM reconciliation_runs WHERE ok = 1 AND json_extract(report, '$.trigger') = 'scheduled' AND finished_at < ?")
++ .run(iso(ctx.now() - keepDays * 86_400_000));
++ }
++ const log = ctx.log || console;
++ if (!report.ok) log.warn(`[Billing] scheduled reconciliation ${report.id} FAILED: ${report.checks.filter((c) => !c.ok).map((c) => c.id).join(', ')}`);
++ return report;
++}
++
++/** The most recent stored run (any trigger), parsed; null when none. */
++function latest(db) {
++ const row = db.prepare('SELECT report FROM reconciliation_runs ORDER BY finished_at DESC, id DESC LIMIT 1').get();
++ return row ? JSON.parse(row.report) : null;
++}
++
++module.exports = { reconcile, runScheduled, latest };
ddiff --git a/test/external.test.js b/test/external.test.js
nnew file mode 100644
iindex 0000000..dd3a5f6
---- /dev/null
++++ b/test/external.test.js
@@@ -0,0 +1,201 @@
++'use strict';
++/**
++ * EXTERNAL receipts (ADR-012; ops/external.js): a tip on a streamer's own PowerChat is never booked,
++ * is stored once per payment in external_receipts, and is announced as billing.receipt.external —
++ * exactly once, only when BILLING_AUTHORITY=billing and the receiving account belongs to a known
++ * creator. While Live is the authority nothing is announced (Live's own webhook does it), so a tip is
++ * never celebrated twice. Accounts come from Live's powerchat_connections (import) or an operator.
++ */
++const assert = require('assert');
++const path = require('path');
++const Database = require('better-sqlite3');
++const { validate } = require('openvibe-contracts');
++const { boot, check, done } = require('./helpers/app');
++const { startEvents } = require('./helpers/stubs');
++
++const donation = (data, streamer = { id: 'pc-77', username: 'StreamerPC' }) => ({ type: 'donation.completed', streamer, data: { eventId: `don-${Math.random().toString(36).slice(2)}`, ...data } });
++const externalEvents = (db) => db.prepare("SELECT event FROM outbox WHERE json_extract(event, '$.event_type') = 'billing.receipt.external' ORDER BY seq").all().map((r) => JSON.parse(r.event));
++
++(async () => {
++ console.log('external receipts');
++
++ await check('BILLING_AUTHORITY must be live or billing; live is the default', async () => {
++ const { loadConfig } = require('../server/config');
++ assert.strictEqual(loadConfig({}).authority, 'live');
++ assert.strictEqual(loadConfig({ BILLING_AUTHORITY: 'Billing' }).authority, 'billing');
++ assert.throws(() => loadConfig({ BILLING_AUTHORITY: 'biling' }), /BILLING_AUTHORITY/);
++ });
++
++ // ── Live is the authority (shadow): recorded, never announced ──
++ {
++ const t = await boot();
++ const streamer = t.user(40);
++ await t.call('POST', '/api/v1/admin/provider-accounts', { body: { provider: 'powerchat', username: 'StreamerPC', account_id: 'pc-77', subject: streamer.id } });
++
++ await check('while Live is the authority an EXTERNAL tip is recorded, not announced, and nothing is booked', async () => {
++ const before = t.db.prepare('SELECT COUNT(*) AS n FROM transactions').get().n;
++ const r = await t.powerchat(donation({ eventId: 'ext-L1', amountUsdCents: 500, donorName: 'Fan', message: 'hi' }));
++ assert.strictEqual(r.status, 200);
++ assert.strictEqual(r.json.result.effect, 'external');
++ assert.strictEqual(r.json.result.announced, false);
++ assert.match(r.json.result.reason, /BILLING_AUTHORITY=live/);
++ assert.strictEqual(externalEvents(t.db).length, 0, 'no billing.receipt.external while Live announces');
++ const row = t.db.prepare("SELECT * FROM external_receipts WHERE receipt_ref = 'powerchat:ext-L1'").get();
++ assert.deepStrictEqual([row.status, row.amount_cents, row.streamer_subject], ['not_announced', 500, streamer.id]);
++ assert.strictEqual(t.db.prepare('SELECT COUNT(*) AS n FROM transactions').get().n, before, 'EXTERNAL money is never journaled');
++ const rec = t.assertReconciled('external under live');
++ assert.strictEqual(rec.totals.external_receipts, 1);
++ assert.strictEqual(rec.totals.external_announced, 0);
++ });
++ await t.close();
++ }
++
++ // ── Billing is the authority: announced once ──
++ const t = await boot({ env: { BILLING_AUTHORITY: 'billing' } });
++ const streamer = t.user(41);
++ const other = t.user(42);
++
++ await check('an operator maps a PowerChat account to a creator (validated, idempotent, allowed while frozen)', async () => {
++ const bad = await t.call('POST', '/api/v1/admin/provider-accounts', { body: { provider: 'powerchat', username: 'has space', subject: streamer.id } });
++ assert.strictEqual(bad.status, 422);
++ const badSub = await t.call('POST', '/api/v1/admin/provider-accounts', { body: { provider: 'powerchat', username: 'streamerpc', subject: { type: 'user', id: '42' } } });
++ assert.strictEqual(badSub.status, 422);
++ const noCap = await t.call('POST', '/api/v1/admin/provider-accounts', { cap: ['billing.intent.create'], body: { provider: 'powerchat', username: 'streamerpc', subject: streamer.id } });
++ assert.strictEqual(noCap.status, 403);
++ await t.call('POST', '/api/v1/admin/freeze', { key: null, body: { on: true, reason: 'cutover' } });
++ const ok = await t.call('POST', '/api/v1/admin/provider-accounts', { body: { provider: 'powerchat', username: 'StreamerPC', account_id: 'pc-77', subject: streamer.id } });
++ assert.strictEqual(ok.status, 201, ok.text);
++ assert.deepStrictEqual([ok.json.account.username, ok.json.account.subject, ok.json.account.source], ['streamerpc', streamer.id, 'admin']);
++ await t.call('POST', '/api/v1/admin/freeze', { key: null, body: { on: false } });
++ const list = await t.call('GET', '/api/v1/admin/provider-accounts');
++ assert.strictEqual(list.json.accounts.length, 1);
++ });
++
++ let first;
++ await check('an EXTERNAL tip is announced as billing.receipt.external with everything Tips needs', async () => {
++ const r = await t.powerchat(donation({ eventId: 'ext-1', amountUsdCents: 1234, donorName: 'Generous Fan', message: 'love the stream', appPurpose: 'goal:12', occurredAt: '2026-09-23T20:00:00Z' }), { deliveryId: 'dlv-ext-1' });
++ assert.strictEqual(r.status, 200);
++ assert.deepStrictEqual([r.json.result.effect, r.json.result.announced, r.json.result.streamer], ['external', true, streamer.id]);
++ const evs = externalEvents(t.db);
++ assert.strictEqual(evs.length, 1);
++ first = evs[0];
++ assert.ok(validate('events.event-envelope@1', first).valid);
++ assert.deepStrictEqual([first.source, first.subject], ['billing', { type: 'provider_receipt', id: 'powerchat:ext-1' }]);
++ const p = first.payload;
++ assert.deepStrictEqual(p.streamer, { type: 'user', id: streamer.id });
++ assert.deepStrictEqual([p.amount_cents, p.currency, p.value_bits], [1234, 'usd-cents', 1234]);
++ assert.deepStrictEqual([p.donor_name, p.anonymous, p.message], ['Generous Fan', false, 'love the stream']);
++ assert.deepStrictEqual([p.provider, p.provider_event_id, p.delivery_id, p.receipt_ref], ['powerchat', 'ext-1', 'dlv-ext-1', 'powerchat:ext-1']);
++ assert.deepStrictEqual(p.receiving_account, { provider: 'powerchat', id: 'pc-77', username: 'streamerpc' });
++ assert.deepStrictEqual([p.classification, p.app_purpose, p.test], ['EXTERNAL', 'goal:12', false]);
++ assert.strictEqual(t.db.prepare("SELECT COUNT(*) AS n FROM transactions WHERE receipt_ref = 'powerchat:ext-1'").get().n, 0);
++ t.assertReconciled('after an announced external tip');
++ });
++
++ await check('a redelivery — same delivery id or a new one — announces nothing more', async () => {
++ const same = await t.powerchat(donation({ eventId: 'ext-1', amountUsdCents: 1234 }), { deliveryId: 'dlv-ext-1' });
++ assert.strictEqual(same.json.duplicate, true);
++ const again = await t.powerchat(donation({ eventId: 'ext-1', amountUsdCents: 1234 }), { deliveryId: 'dlv-ext-1-retry' });
++ assert.strictEqual(again.json.result.effect, 'duplicate_receipt');
++ assert.strictEqual(externalEvents(t.db).length, 1);
++ const firstEvent = t.db.prepare("SELECT id FROM provider_events WHERE provider_event_id = 'dlv-ext-1'").get().id;
++ for (const id of [firstEvent, again.json.event]) {
++ const refused = await t.call('POST', `/api/v1/admin/provider-events/${id}/reprocess`, { key: null });
++ assert.strictEqual(refused.status, 409, 'an announced receipt (or its duplicate) is never re-run');
++ }
++ assert.strictEqual(externalEvents(t.db).length, 1);
++ t.assertReconciled('after external redeliveries');
++ });
++
++ await check('an anonymous tip carries no donor name; the account is found by id when the username changed', async () => {
++ const r = await t.powerchat(donation({ eventId: 'ext-anon', amountUsdCents: 300, donorName: 'Real Name', isAnonymous: true }, { id: 'pc-77', username: 'renamedpc' }));
++ assert.strictEqual(r.json.result.announced, true);
++ const p = externalEvents(t.db).find((e) => e.payload.provider_event_id === 'ext-anon').payload;
++ assert.deepStrictEqual([p.donor_name, p.anonymous, p.streamer.id], [null, true, streamer.id]);
++ });
++
++ await check('a tip on an account no creator connected is held for review; mapped and reprocessed, it is announced once', async () => {
++ const r = await t.powerchat(donation({ eventId: 'ext-unknown', amountUsdCents: 700, donorName: 'Someone' }, { id: 'pc-99', username: 'OtherPC' }));
++ assert.deepStrictEqual([r.json.result.effect, r.json.result.announced, r.json.result.review], ['external', false, true]);
++ assert.match(r.json.result.reason, /otherpc/);
++ const rec = t.assertReconciled('with an unmapped external tip');
++ assert.ok(rec.warnings.events_for_review.some((e) => e.id === r.json.event));
++ const n = externalEvents(t.db).length;
++ await t.call('POST', '/api/v1/admin/provider-accounts', { body: { provider: 'powerchat', username: 'otherpc', subject: other.id } });
++ const re = await t.call('POST', `/api/v1/admin/provider-events/${r.json.event}/reprocess`, { key: null });
++ assert.strictEqual(re.status, 200, re.text);
++ assert.deepStrictEqual([re.json.event.result.effect, re.json.event.result.announced], ['external', true]);
++ assert.strictEqual(externalEvents(t.db).length, n + 1);
++ assert.strictEqual(externalEvents(t.db).at(-1).payload.streamer.id, other.id);
++ assert.strictEqual(t.db.prepare("SELECT status FROM external_receipts WHERE receipt_ref = 'powerchat:ext-unknown'").get().status, 'announced');
++ });
++
++ await check('an underpaid direct subscription is an EXTERNAL tip to the intent\'s streamer', async () => {
++ const i = await t.call('POST', '/api/v1/intents', { cap: ['billing.intent.create'], body: { provider: 'powerchat', kind: 'subscription', subject: other, streamer, route: 'direct', receiving_account: 'StreamerPC' } });
++ assert.strictEqual(i.status, 201, i.text);
++ const r = await t.powerchat(donation({ eventId: 'ext-under', amountUsdCents: 200, appExternalRef: i.json.intent.checkout_ref }));
++ assert.deepStrictEqual([r.json.result.effect, r.json.result.announced, r.json.result.streamer], ['external', true, streamer.id]);
++ const e = await t.call('GET', `/api/v1/entitlements/${other.id}?streamer=${streamer.id}`);
++ assert.strictEqual(e.json.active, false, 'underpaid: no subscription');
++ });
++
++ await check('zero, test and out-of-range amounts announce nothing', async () => {
++ const n = externalEvents(t.db).length;
++ const zero = await t.powerchat(donation({ amountUsdCents: 0 }));
++ assert.strictEqual(zero.json.result.effect, 'none');
++ const test = await t.powerchat(donation({ amountUsdCents: 500, isTest: true }));
++ assert.strictEqual(test.json.result.effect, 'none');
++ const huge = await t.powerchat(donation({ amountUsdCents: 1e20 }));
++ assert.deepStrictEqual([huge.json.result.effect, huge.json.result.review], ['none', true]);
++ const inf = await t.powerchat(donation({ amountUsdCents: '1e400' }));
++ assert.strictEqual(inf.json.result.effect, 'none');
++ assert.strictEqual(externalEvents(t.db).length, n);
++ t.assertReconciled('after refused external amounts');
++ });
++
++ await check('the outbox relay publishes billing.receipt.external to OpenVibe.Events', async () => {
++ const events = await startEvents();
++ const { createRelay } = require('../server/outbox');
++ const relay = createRelay({ db: t.db, config: { ...t.config, events: { url: events.url, intervalMs: 1000 } }, log: { warn() {} } });
++ await relay.flush();
++ const sent = events.batches.flatMap((b) => b.events).filter((e) => e.event_type === 'billing.receipt.external');
++ assert.strictEqual(sent.length, externalEvents(t.db).length);
++ assert.strictEqual(sent[0].event_id, first.event_id);
++ await events.close();
++ });
++
++ await check('the importer maps Live\'s powerchat_connections (never over an operator\'s mapping); --accounts-only alone', async () => {
++ const file = path.join(t.dir, 'live-accounts.db');
++ const live = new Database(file);
++ live.exec(`CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT, openvibe_bucks_balance REAL DEFAULT 0, openvibe_bucks_cashout_balance REAL DEFAULT 0);
++ CREATE TABLE powerchat_connections (user_id INTEGER PRIMARY KEY, powerchat_username TEXT, powerchat_user_id TEXT);
++ INSERT INTO users (id, username) VALUES (50, 'newbie'), (51, 'ghost'), (52, 'clash');
++ INSERT INTO powerchat_connections VALUES (50, 'NewbiePC', 'pc-50'), (51, 'GhostPC', 'pc-51'), (52, 'StreamerPC', 'pc-77'), (53, 'bad name!', NULL);`);
++ live.close();
++ const newbie = t.network.addUser(50);
++ t.network.addUser(52);
++ t.network.addUser(53);
++ const { importProviderAccounts, importLive } = require('../server/importer/live');
++ const ro = new Database(file, { readonly: true });
++ const dry = await importProviderAccounts(t.ctx, { live: ro, resolveLiveUsers: t.ctx.network.resolveLiveUsers, dryRun: true, log: {} });
++ assert.strictEqual(dry.provider_accounts.mapped, 1);
++ assert.strictEqual(t.db.prepare("SELECT COUNT(*) AS n FROM provider_accounts WHERE username = 'newbiepc'").get().n, 0, 'dry run keeps nothing');
++ const rep = await importProviderAccounts(t.ctx, { live: ro, resolveLiveUsers: t.ctx.network.resolveLiveUsers, log: {} });
++ const a = rep.provider_accounts;
++ assert.strictEqual(a.mapped, 1);
++ assert.deepStrictEqual(a.unmapped.map((u) => [u.live_user_id, u.reason]), [[51, 'no Network subject'], [53, 'not a PowerChat username']]);
++ assert.deepStrictEqual(a.kept_admin.map((k) => k.username), ['streamerpc']);
++ const row = t.db.prepare("SELECT * FROM provider_accounts WHERE username = 'newbiepc'").get();
++ assert.deepStrictEqual([row.subject, row.account_id, row.source, row.live_user_id], [newbie, 'pc-50', 'live-import', 50]);
++ assert.strictEqual(t.db.prepare("SELECT subject FROM provider_accounts WHERE username = 'streamerpc'").get().subject, streamer.id);
++ const full = await importLive(t.ctx, { live: ro, resolveLiveUsers: t.ctx.network.resolveLiveUsers, log: {} });
++ assert.strictEqual(full.provider_accounts.unchanged, 1, 'the full import runs the same step; a re-run changes nothing');
++ ro.close();
++ const r = await t.powerchat(donation({ eventId: 'ext-newbie', amountUsdCents: 100 }, { id: 'pc-50', username: 'NewbiePC' }));
++ assert.strictEqual(r.json.result.streamer, newbie);
++ t.assertReconciled('after the accounts import');
++ });
++
++ await t.close();
++ done();
++})();
ddiff --git a/test/hygiene.test.js b/test/hygiene.test.js
nnew file mode 100644
iindex 0000000..4014b8a
---- /dev/null
++++ b/test/hygiene.test.js
@@@ -0,0 +1,325 @@
++'use strict';
++/**
++ * Hygiene: amounts that would overflow exact integer arithmetic are refused whole; concurrent
++ * settlements of one provider payment (same delivery, redeliveries, several processes)
++ * settle it once; an outage during settlement (the Network down while a site-routed tip resolves its
++ * creator, a failure inside the settlement transaction, OpenVibe.Events down) leaves nothing half
++ * done and settles once on recovery; the scheduled reconciliation is stored, pruned and exposed
++ * (API, /metrics), and fails on a receipt stuck unprocessed.
++ */
++const assert = require('assert');
++const { boot, fund, check, done } = require('./helpers/app');
++const { startEvents } = require('./helpers/stubs');
++
++const donation = (data, streamer = 'openvibe') => ({ type: 'donation.completed', streamer: { id: 'pc1', username: streamer }, data: { eventId: `don-${Math.random().toString(36).slice(2)}`, ...data } });
++
++(async () => {
++ const t = await boot();
++ const buyer = t.user(61);
++ const streamer = t.user(62);
++ const streamerLiveId = 63;
++ const siteStreamer = t.network.addUser(streamerLiveId);
++ const intent = async (body) => {
++ const r = await t.call('POST', '/api/v1/intents', { cap: ['billing.intent.create'], body: { provider: 'powerchat', ...body } });
++ assert.strictEqual(r.status, 201, r.text);
++ return r.json.intent;
++ };
++ const txnCount = () => t.db.prepare('SELECT COUNT(*) AS n FROM transactions').get().n;
++ const providers = require('../server/providers');
++ console.log('hygiene');
++
++ // ── Overflow ─────────────────────────────────────────────
++ await check('huge provider amounts are rejected for review, never booked', async () => {
++ const before = txnCount();
++ const i = await intent({ kind: 'purchase', subject: buyer, bits: 100 });
++ for (const cents of [1e20, 2 ** 53, 9007199254740993, 100_000_001, -500]) {
++ const r = await t.powerchat(donation({ amountUsdCents: cents, appExternalRef: i.checkout_ref }));
++ assert.strictEqual(r.status, 200);
++ assert.ok(['rejected', 'none'].includes(r.json.result.effect), `${cents}: ${JSON.stringify(r.json.result)}`);
++ }
++ const tip = await t.powerchat(donation({ amountUsdCents: 1e17, appExternalRef: `pcdon:${streamerLiveId}:0` }));
++ assert.strictEqual(tip.json.result.effect, 'rejected');
++ assert.strictEqual(tip.json.result.code, 'billing.invalid_amount');
++ assert.strictEqual(txnCount(), before);
++ t.assertReconciled('after refused huge receipts');
++ });
++
++ await check('the API refuses amounts past the exact integer range or the per-call maximum', async () => {
++ const settle = await t.call('POST', '/api/v1/purchases/settle', { body: { provider: 'powerchat', provider_ref: 'huge-1', subject: buyer, amount_cents: 9007199254740993, bits: 100 } });
++ assert.strictEqual(settle.status, 422);
++ assert.strictEqual(settle.json.code, 'billing.invalid_amount');
++ const tr = await t.call('POST', '/api/v1/transfers', { body: { from: buyer, to: streamer, amount: 1e300 } });
++ assert.strictEqual(tr.status, 422);
++ const adj = await t.call('POST', '/api/v1/admin/adjustments', { body: { from: { kind: 'import_adjustment', currency: 'vibes-bits' }, to: { kind: 'user_credit', owner: buyer.id, currency: 'vibes-bits' }, amount: 2 ** 60, reason: 'x' } });
++ assert.strictEqual(adj.status, 422);
++ });
++
++ await check('the journal refuses a posting whose amount or resulting balance leaves the exact range, as a whole', async () => {
++ const { post, balance } = require('../server/ledger');
++ const { A, entry } = require('../server/ops/common');
++ const acct = A.credit(buyer.id);
++ const big = Number.MAX_SAFE_INTEGER - 10;
++ assert.throws(() => post(t.ctx, { type: 'adjustment', idempotencyKey: 'ovf-1', entries: [entry(acct, 2 ** 53), entry(A.importAdj('vibes-bits'), -(2 ** 53))] }), /exact integer range/);
++ post(t.ctx, { type: 'adjustment', idempotencyKey: 'ovf-2', entries: [entry(acct, big), entry(A.importAdj('vibes-bits'), -big)] });
++ const at = balance(t.db, acct);
++ const count = txnCount();
++ assert.throws(() => post(t.ctx, { type: 'adjustment', idempotencyKey: 'ovf-3', entries: [entry(acct, 100), entry(A.importAdj('vibes-bits'), -100)] }), (e) => e.code === 'billing.amount_overflow');
++ assert.strictEqual(balance(t.db, acct), at, 'nothing moved');
++ assert.strictEqual(txnCount(), count, 'no transaction row');
++ post(t.ctx, { type: 'adjustment', idempotencyKey: 'ovf-4', entries: [entry(acct, -big), entry(A.importAdj('vibes-bits'), big)] });
++ t.assertReconciled('after overflow refusals');
++ });
++
++ await check('a refund claiming more than was paid reverses at most what was paid', async () => {
++ const i = await intent({ kind: 'purchase', subject: buyer, bits: 500 });
++ const pay = await t.powerchat(donation({ eventId: 'don-cap', amountUsdCents: 700, appExternalRef: i.checkout_ref }));
++ assert.strictEqual(pay.json.result.effect, 'settled');
++ const rf = await t.powerchat({ type: 'donation.refunded', streamer: { username: 'openvibe' }, data: { eventId: 'rf-cap', originalEventId: 'don-cap', refundedUsdCents: 1e300 } });
++ assert.strictEqual(rf.json.result.effect, 'settled', JSON.stringify(rf.json));
++ const tx = (await t.call('GET', `/api/v1/transactions/${rf.json.result.txn_id}`)).json.transaction;
++ assert.strictEqual(tx.metadata.cents, 700);
++ t.assertReconciled('after an over-claimed refund');
++ });
++
++ // ── Concurrent settlement of one provider payment ────────
++ await check('20 concurrent copies of one delivery settle it once', async () => {
++ const i = await intent({ kind: 'purchase', subject: buyer, bits: 1000 });
++ const credit = (await t.balances(buyer.id)).credit;
++ const env = donation({ eventId: 'don-race-1', amountUsdCents: 1300, appExternalRef: i.checkout_ref });
++ const rs = await Promise.all(Array.from({ length: 20 }, () => t.powerchat(env, { deliveryId: 'dlv-race-1' })));
++ assert.ok(rs.every((r) => r.status === 200), rs.map((r) => r.status).join(','));
++ assert.strictEqual(t.db.prepare("SELECT COUNT(*) AS n FROM provider_events WHERE provider_event_id = 'dlv-race-1'").get().n, 1);
++ assert.strictEqual(t.db.prepare("SELECT COUNT(*) AS n FROM transactions WHERE receipt_ref = 'powerchat:don-race-1'").get().n, 1);
++ assert.strictEqual((await t.balances(buyer.id)).credit, credit + 1000);
++ t.assertReconciled('after a delivery race');
++ });
++
++ await check('20 concurrent redeliveries of one payment under different delivery ids settle it once', async () => {
++ const credit = (await t.balances(siteStreamer)).payable;
++ // pcdon resolves its creator through the Network (async), so the deliveries really interleave.
++ const env = donation({ eventId: 'don-race-2', amountUsdCents: 250, appExternalRef: `pcdon:${streamerLiveId}:0` });
++ const rs = await Promise.all(Array.from({ length: 20 }, (_, n) => t.powerchat(env, { deliveryId: `dlv-race-2-${n}` })));
++ const effects = rs.map((r) => r.json.result.effect).sort();
++ assert.strictEqual(effects.filter((e) => e === 'settled').length, 1, effects.join(','));
++ assert.strictEqual(effects.filter((e) => e === 'duplicate_receipt').length, 19);
++ assert.strictEqual((await t.balances(siteStreamer)).payable, credit + 250);
++ t.assertReconciled('after a redelivery race');
++ });
++
++ await check('four processes racing on one stored receipt (the service, a script, a retry) settle it once', async () => {
++ const i = await intent({ kind: 'purchase', subject: buyer, bits: 500 });
++ await t.call('POST', '/api/v1/admin/freeze', { key: null, body: { on: true, reason: 'race test' } });
++ const q = await t.powerchat(donation({ eventId: 'don-race-3', amountUsdCents: 700, appExternalRef: i.checkout_ref }), { deliveryId: 'dlv-race-3' });
++ assert.strictEqual(q.status, 202);
++ require('../server/ops/admin').setFreeze(t.ctx, { on: false });
++ const credit = (await t.balances(buyer.id)).credit;
++ // Each child opens its own connection to the same database file and, at the same instant,
++ // processes the same stored receipt — what the service's retry tick and an operator's
++ // unfreeze from scripts/ could do at once.
++ const child = `
++ const { loadConfig } = require('./server/config');
++ const { openDb } = require('./server/db');
++ const { createRates } = require('./server/rates');
++ const providers = require('./server/providers');
++ const config = loadConfig({ NODE_ENV: 'test', BILLING_DB_PATH: process.env.DB, POWERCHAT_WEBHOOK_SECRET: 'pc-secret', POWERCHAT_SITE_USERNAME: 'openvibe' });
++ const db = openDb(config.dbPath);
++ const ctx = { db, config, rates: createRates(config.rates), now: () => Date.now(), log: { warn: (...a) => console.error(...a) } };
++ const adapters = providers.createAdapters(config, {});
++ while (Date.now() < Number(process.env.AT)) { /* line up */ }
++ providers.process(ctx, adapters, Number(process.env.EVENT)).then((row) => { console.log(JSON.stringify({ processed: !!row.processed_at, result: row.result })); db.close(); });`;
++ const { spawn } = require('child_process');
++ const at = Date.now() + 1500;
++ const run = () => new Promise((resolve) => {
++ const c = spawn(process.execPath, ['-e', child], { cwd: require('path').join(__dirname, '..'), env: { ...process.env, DB: t.config.dbPath, AT: String(at), EVENT: String(q.json.event) } });
++ let out = '';
++ c.stdout.on('data', (d) => { out += d; });
++ c.stderr.on('data', (d) => { out += d; });
++ c.on('close', (code) => resolve({ code, out }));
++ });
++ const results = await Promise.all([run(), run(), run(), run()]);
++ for (const r of results) {
++ assert.strictEqual(r.code, 0, r.out);
++ assert.strictEqual(JSON.parse(r.out.trim().split('\n').pop()).processed, true, r.out);
++ }
++ assert.strictEqual(t.db.prepare("SELECT COUNT(*) AS n FROM transactions WHERE receipt_ref = 'powerchat:don-race-3'").get().n, 1);
++ const row = t.db.prepare('SELECT attempts, last_error, result FROM provider_events WHERE id = ?').get(q.json.event);
++ assert.deepStrictEqual([row.attempts, row.last_error, JSON.parse(row.result).effect], [1, null, 'settled'], 'applied once; the others waited for the lock and found it processed');
++ assert.strictEqual((await t.balances(buyer.id)).credit, credit + 500);
++ t.assertReconciled('after a multi-process race');
++ });
++
++ await check('an operator reprocess racing the retry job does not settle twice', async () => {
++ t.network.state.down = true;
++ const r = await t.powerchat(donation({ eventId: 'don-race-4', amountUsdCents: 300, appExternalRef: `pcdon:${streamerLiveId}:0` }));
++ assert.strictEqual(r.json.processed, false);
++ t.network.state.down = false;
++ const payable = (await t.balances(siteStreamer)).payable;
++ await Promise.all([
++ t.call('POST', `/api/v1/admin/provider-events/${r.json.event}/reprocess`, { key: null }),
++ providers.processPending(t.ctx, t.app.locals.adapters),
++ providers.processPending(t.ctx, t.app.locals.adapters),
++ ]);
++ assert.strictEqual(t.db.prepare("SELECT COUNT(*) AS n FROM transactions WHERE receipt_ref = 'powerchat:don-race-4'").get().n, 1);
++ assert.strictEqual((await t.balances(siteStreamer)).payable, payable + 300);
++ });
++
++ // ── Outages during settlement ────────────────────────────
++ let heldEvent;
++ await check('the Network down while a site tip resolves its creator: stored, 200 unprocessed, nothing booked, retried later', async () => {
++ t.network.state.down = true;
++ const before = txnCount();
++ const r = await t.powerchat(donation({ eventId: 'don-out-1', amountUsdCents: 400, appExternalRef: `pcdon:${streamerLiveId}:0` }), { deliveryId: 'dlv-out-1' });
++ assert.strictEqual(r.status, 200, 'acknowledged: the receipt is stored, PowerChat must not retry forever');
++ assert.strictEqual(r.json.processed, false);
++ heldEvent = r.json.event;
++ const row = t.db.prepare('SELECT * FROM provider_events WHERE id = ?').get(heldEvent);
++ assert.strictEqual(row.processed_at, null);
++ assert.ok(row.last_error, 'the failure is recorded');
++ assert.strictEqual(txnCount(), before);
++ const pending = await providers.processPending(t.ctx, t.app.locals.adapters);
++ assert.strictEqual(pending.processed, 0, 'still down: still pending');
++ const rec = t.assertReconciled('with a pending receipt');
++ assert.strictEqual(rec.warnings.unprocessed_events.length, 1);
++ });
++
++ await check('a receipt stuck past BILLING_RECONCILE_STALE_RECEIPT_MIN fails reconciliation (not while frozen)', async () => {
++ const { reconcile } = require('../server/reconcile');
++ t.clock.offset = 61 * 60_000;
++ const r = reconcile(t.ctx, { store: false });
++ const stale = r.checks.find((c) => c.id === 'receipts.stale');
++ assert.strictEqual(stale.ok, false);
++ assert.deepStrictEqual(stale.detail.offenders.map((o) => o.id), [heldEvent]);
++ require('../server/ops/admin').setFreeze(t.ctx, { on: true, reason: 'x' });
++ assert.strictEqual(reconcile(t.ctx, { store: false }).checks.find((c) => c.id === 'receipts.stale').ok, true, 'held on purpose while frozen');
++ require('../server/ops/admin').setFreeze(t.ctx, { on: false });
++ const m = await fetch(`${t.base}/metrics`).then((x) => x.text());
++ assert.match(m, /billing_provider_event_oldest_pending_seconds (36[6-9]\d|3[7-9]\d\d)/);
++ t.clock.offset = 0;
++ });
++
++ await check('the Network back: the retry job settles the held receipt once', async () => {
++ t.network.state.down = false;
++ const payable = (await t.balances(siteStreamer)).payable;
++ const out = await providers.processPending(t.ctx, t.app.locals.adapters);
++ assert.strictEqual(out.processed, 1);
++ await providers.processPending(t.ctx, t.app.locals.adapters);
++ assert.strictEqual((await t.balances(siteStreamer)).payable, payable + 400);
++ const row = t.db.prepare('SELECT attempts, last_error, result FROM provider_events WHERE id = ?').get(heldEvent);
++ assert.strictEqual(JSON.parse(row.result).effect, 'settled');
++ assert.strictEqual(row.last_error, null);
++ t.assertReconciled('after recovery');
++ });
++
++ await check('a failure inside the settlement transaction rolls everything back; the retry settles once', async () => {
++ const intents = require('../server/ops/intents');
++ const i = await intent({ kind: 'purchase', subject: buyer, bits: 100 });
++ const before = txnCount();
++ const credit = (await t.balances(buyer.id)).credit;
++ const real = intents.markSettled;
++ let fails = 1;
++ // Marking the intent settled comes after the journal posting, in the same transaction: failing
++ // it must take the posting, the outbox event and the processed mark down with it.
++ intents.markSettled = (...a) => { if (fails-- > 0) throw new Error('disk full'); return real(...a); };
++ const events = t.db.prepare('SELECT COUNT(*) AS n FROM outbox').get().n;
++ let r;
++ try { r = await t.powerchat(donation({ eventId: 'don-crash', amountUsdCents: 150, appExternalRef: i.checkout_ref })); }
++ finally { intents.markSettled = real; }
++ assert.strictEqual(t.db.prepare('SELECT COUNT(*) AS n FROM outbox').get().n, events, 'no event for a settlement that did not happen');
++ assert.strictEqual(r.json.processed, false);
++ assert.strictEqual(txnCount(), before);
++ assert.strictEqual((await t.balances(buyer.id)).credit, credit);
++ assert.strictEqual((await t.call('GET', `/api/v1/intents/${i.id}`)).json.intent.status, 'created');
++ assert.match(t.db.prepare('SELECT last_error FROM provider_events WHERE id = ?').get(r.json.event).last_error, /disk full/);
++ await providers.processPending(t.ctx, t.app.locals.adapters);
++ assert.strictEqual((await t.balances(buyer.id)).credit, credit + 100);
++ assert.strictEqual(t.db.prepare("SELECT COUNT(*) AS n FROM transactions WHERE receipt_ref = 'powerchat:don-crash'").get().n, 1);
++ t.assertReconciled('after a mid-settlement failure');
++ });
++
++ await check('OpenVibe.Events down: settlement is unaffected, events wait in the outbox and go out once it is back', async () => {
++ const { createRelay } = require('../server/outbox');
++ const log = { warn() {} };
++ const unsent = () => t.db.prepare('SELECT COUNT(*) AS n FROM outbox WHERE sent_at IS NULL').get().n;
++ const down = createRelay({ db: t.db, config: { ...t.config, events: { url: 'http://127.0.0.1:9', intervalMs: 1000 } }, log });
++ await fund(t, buyer, 200);
++ const r = await down.flush();
++ assert.strictEqual(r.sent, 0);
++ const waiting = unsent();
++ assert.ok(waiting > 0);
++ const m = await fetch(`${t.base}/metrics`).then((x) => x.text());
++ assert.match(m, new RegExp(`billing_outbox_pending ${waiting}\\b`));
++ assert.match(m, /billing_outbox_failing [1-9]/);
++ const events = await startEvents();
++ const up = createRelay({ db: t.db, config: { ...t.config, events: { url: events.url, intervalMs: 1000 } }, log });
++ await up.flush();
++ assert.strictEqual(unsent(), 0);
++ const ids = events.batches.flatMap((b) => b.events).map((e) => e.event_id);
++ assert.strictEqual(new Set(ids).size, ids.length, 'each event published once');
++ await events.close();
++ });
++
++ // ── Scheduled reconciliation, exposed ────────────────────
++ await check('the scheduled run is stored with its trigger, listed by the API and old passing runs are pruned', async () => {
++ const { runScheduled } = require('../server/reconcile');
++ const old = runScheduled({ ...t.ctx, now: () => Date.now() - 40 * 86_400_000 }, { keepDays: 0 });
++ assert.strictEqual(old.ok, true);
++ const rep = runScheduled(t.ctx, { keepDays: 30 });
++ assert.strictEqual(rep.trigger, 'scheduled');
++ assert.ok(rep.checks.some((c) => c.id === 'receipts.ledger' && c.ok));
++ assert.strictEqual(t.db.prepare('SELECT COUNT(*) AS n FROM reconciliation_runs WHERE id = ?').get(old.id).n, 0, 'a 40-day-old passing scheduled run is pruned');
++ const list = await t.call('GET', '/api/v1/admin/reconciliations?limit=5', { cap: ['billing.ledger.admin'] });
++ assert.strictEqual(list.status, 200, list.text);
++ assert.deepStrictEqual([list.json.runs[0].id, list.json.runs[0].trigger, list.json.runs[0].ok], [rep.id, 'scheduled', true]);
++ const latest = await t.call('GET', '/api/v1/admin/reconciliations/latest');
++ assert.strictEqual(latest.json.id, rep.id);
++ assert.ok(latest.json.checks.length >= 10);
++ assert.strictEqual((await t.call('GET', '/api/v1/admin/reconciliations/latest', { cap: ['billing.balance.read'] })).status, 403);
++ assert.strictEqual((await t.call('GET', '/api/v1/admin/reconciliations/rec_nope')).status, 404);
++ });
++
++ await check('a failed run is kept and shows on /metrics; the ledger-vs-receipts check catches a wrong clearing amount', async () => {
++ const { runScheduled } = require('../server/reconcile');
++ // A settlement whose clearing entry disagrees with the receipt it names.
++ const { post } = require('../server/ledger');
++ const { A, entry } = require('../server/ops/common');
++ const ev = t.db.prepare("INSERT INTO provider_events (provider, provider_event_id, type, payload_hash, payload, received_at, processed_at, result) VALUES ('powerchat', 'forged', 'donation.completed', 'x', '{}', ?, ?, ?)")
++ .run(new Date().toISOString(), new Date().toISOString(), JSON.stringify({ effect: 'settled' })).lastInsertRowid;
++ post(t.ctx, {
++ type: 'purchase', idempotencyKey: 'forged-1', provider: 'powerchat', receiptRef: 'powerchat:forged', sourceEventId: Number(ev), toSubject: buyer.id,
++ entries: [entry(A.clearing('powerchat'), -100), entry(A.fxCents(), 100), entry(A.fxBits(), -100), entry(A.credit(buyer.id), 100)],
++ metadata: { paid_cents: 900, bits: 100, rates: { bits_per_usd: 100 } },
++ });
++ const rep = runScheduled({ ...t.ctx, log: { warn() {} } }, { keepDays: 30 });
++ assert.strictEqual(rep.ok, false);
++ const c = rep.checks.find((x) => x.id === 'receipts.ledger');
++ assert.strictEqual(c.ok, false);
++ assert.deepStrictEqual(c.detail.offenders.map((o) => [o.receipt_ref, o.clearing, o.paid_cents]), [['powerchat:forged', -100, 900]]);
++ runScheduled({ ...t.ctx, log: { warn() {} } }, { keepDays: 0 });
++ assert.strictEqual(t.db.prepare('SELECT COUNT(*) AS n FROM reconciliation_runs WHERE id = ?').get(rep.id).n, 1, 'failed runs are never pruned');
++ const m = await fetch(`${t.base}/metrics`).then((x) => x.text());
++ assert.match(m, /^billing_reconciliation_ok 0$/m);
++ assert.match(m, /^billing_reconciliation_failed_checks 1$/m);
++ assert.match(m, /^billing_reconciliation_last_run_timestamp_seconds \d{10}$/m);
++ const failed = await t.call('GET', '/api/v1/admin/reconciliations?failed=1');
++ assert.ok(failed.json.runs.every((r) => !r.ok) && failed.json.runs.length >= 1);
++ });
++
++ await check('/metrics: golden signals and Billing gauges for loopback callers, 404 for anything relayed', async () => {
++ const res = await fetch(`${t.base}/metrics`);
++ assert.strictEqual(res.status, 200);
++ assert.match(res.headers.get('content-type'), /text\/plain/);
++ const m = await res.text();
++ for (const name of ['http_requests_total', 'release_info', 'billing_frozen 0', 'billing_authority{authority="live"} 1', 'billing_provider_events{provider="powerchat",state="settled"}',
++ 'billing_external_receipts{status="announced"} 0', 'billing_cashouts{status="requested"} 0']) {
++ assert.ok(m.includes(name), `missing ${name}`);
++ }
++ assert.ok(!/usr_[0-9A-Z]{26}/.test(m), 'no subject ever appears in metrics');
++ const relayed = await fetch(`${t.base}/metrics`, { headers: { 'X-Forwarded-For': '203.0.113.9' } });
++ assert.strictEqual(relayed.status, 404);
++ const real = await fetch(`${t.base}/metrics`, { headers: { 'X-Real-IP': '203.0.113.9' } });
++ assert.strictEqual(real.status, 404);
++ });
++
++ await t.close();
++ done();
++})();