From d4092caa6a70adbde1f76964235e1e276b438ea2 Mon Sep 17 00:00:00 2001 From: George Lambert Date: Sat, 3 Oct 2026 23:02:02 -0400 Subject: [PATCH 1/5] WIP snapshot before drive cleanup 2026-10-03 --- README.md | 56 ++++---- package.json | 2 +- src/billing-store.js | 304 +++++++++++++++++++++++++++++++++++++++++++ src/books.js | 3 +- src/edge-env.js | 34 +++++ src/persist.js | 4 + src/server.js | 94 ++++++++----- test/health.test.js | 63 +++++++-- 8 files changed, 489 insertions(+), 71 deletions(-) create mode 100644 src/billing-store.js create mode 100644 src/edge-env.js diff --git a/README.md b/README.md index ee9c754..a8a5216 100644 --- a/README.md +++ b/README.md @@ -2,13 +2,41 @@ [![standard-readme compliant](https://img.shields.io/badge/readme%20style-standard-brightgreen.svg?style=flat-square)](https://github.com/RichardLitt/standard-readme) -Source of truth for **customer prepaid balances** (persisted `books.json`). Internal traffic is **NATS... +NATS/HTTP facade for customer prepaid balances over edge BusinessStore (D006). **Forgejo:** https://git.georgelambert.org/marchon/zappier-account-balance **Catalog:** https://zapier.georgelambert.org/packages/zappier-account-balance/README.pdf +Under decision **D006**, this package is a **facade**, not a competing system of record. Lab SoT is zappier-edge `billing.db` (`customers.balance_cents` + `usage_entries`); multi-host SoT becomes **libSQL** later. `books.json` is no longer written as SoT. + Zapier cloud never connects here. zappier-edge, customer-service, sales-pricing, and accounting-export do. +## Install + +```sh +git clone ssh://git@git.georgelambert.org:2223/marchon/zappier-account-balance.git +cd zappier-account-balance +npm install +``` + +## Usage + +```sh +EDGE_URL=http://127.0.0.1:3040 \ +BILLING_DB=/opt/verae/zappier-edge/billing.db \ +EDGE_ENV_FILE=./.env.edge \ +NATS_URL=nats://127.0.0.1:4222 \ +PORT=3010 npm start + +curl http://127.0.0.1:3010/health +curl http://127.0.0.1:3010/statement/cust_1 +``` + +Statement JSON: `{ customerId, prepaidCents, credits[], usage[], payments[], source }`. +`balance.get` / `statement.get` read `billing.db` (fallback: edge `/admin/api/statement/:id`). +`balance.adjust` proxies to edge `POST /admin/api/credits` (fallback: `sudo -u verae-zapier` sqlite write). +`usage.recorded` is a no-op; edge meter already appends `usage_entries`. + ## Subjects | Subject | Kind | Who | @@ -16,35 +44,15 @@ Zapier cloud never connects here. zappier-edge, customer-service, sales-pricing, | `verae.billing.statement.get` | request-reply | portal, CS, sales, admin | | `verae.billing.balance.get` | request-reply | anyone internal | | `verae.billing.balance.adjust` | request-reply | CS credits, portal reload | -| `verae.billing.usage.recorded` | pub | zappier-edge meter | +| `verae.billing.usage.recorded` | pub | zappier-edge meter (ignored by facade) | | `verae.billing.payment.recorded` | pub | portal reload, invoice paid | | `verae.billing.credit.applied` | pub | CS | Queue group: `account-balance`. -## Run +## Background -```bash -NATS_URL=nats://127.0.0.1:4222 PORT=3010 npm start -curl http://127.0.0.1:3010/health -curl http://127.0.0.1:3010/statement/cust_1 -``` - -Statement JSON: `{ customerId, prepaidCents, credits[], usage[], payments[] }`. - -## Install - -```sh -git clone ssh://git@git.georgelambert.org:2223/marchon/zappier-account-balance.git -cd zappier-account-balance -``` - -## Usage - -```sh -npm install -npm start -``` +D006 freeze: edge BusinessStore is the authoritative meter/balance writer on the single-host lab. This service keeps the historical NATS/HTTP surface so callers do not break while prepaid moves to libSQL. ## Contributing diff --git a/package.json b/package.json index 4906264..6a1ade3 100644 --- a/package.json +++ b/package.json @@ -3,7 +3,7 @@ "version": "0.1.0", "private": true, "type": "module", - "description": "NATS source of truth for customer prepaid balances, credits, usage, payments", + "description": "NATS/HTTP facade for prepaid balances over edge BusinessStore (D006)", "scripts": { "start": "node src/server.js", "test": "node --test test/*.test.js" diff --git a/src/billing-store.js b/src/billing-store.js new file mode 100644 index 0000000..9bb6469 --- /dev/null +++ b/src/billing-store.js @@ -0,0 +1,304 @@ +/** + * D006 facade: edge BusinessStore / billing.db is the lab SoT until libSQL. + * This package reads balance + usage from billing.db (or edge admin HTTP) and + * proxies balance.adjust to edge /admin/api/credits. It must not write books.json. + */ +import { spawnSync } from 'node:child_process'; +import fs from 'node:fs'; +import { DatabaseSync } from 'node:sqlite'; +import { loadEdgeEnv } from './edge-env.js'; + +export function billingDbPath() { + return process.env.BILLING_DB || '/opt/verae/zappier-edge/billing.db'; +} + +export function edgeBaseUrl() { + return (process.env.EDGE_URL || 'http://127.0.0.1:3040').replace(/\/$/, ''); +} + +export function facadeEnabled() { + return Boolean(process.env.BILLING_DB || process.env.EDGE_URL || fs.existsSync(billingDbPath())); +} + +function openDb() { + const p = billingDbPath(); + if (!fs.existsSync(p)) return null; + return new DatabaseSync(p, { readOnly: true }); +} + +function mapUsage(rows) { + return rows.map((r) => ({ + customerId: r.customer_id, + endpointId: r.endpoint_id, + cents: r.cents, + at: new Date(r.timestamp_ms).toISOString(), + })); +} + +function mapPayments(rows) { + return rows.map((r) => ({ + id: r.id, + customerId: r.customer_id, + cents: r.billable_cents, + kind: r.status === 'paid' ? 'invoice-paid' : 'invoice-issued', + reason: r.period, + at: new Date(r.paid_at_ms || r.issued_at_ms || Date.now()).toISOString(), + })); +} + +function findCustomer(db, idOrName) { + const want = String(idOrName || ''); + let row = db + .prepare( + `SELECT id, name, verae_user_id, balance_cents + FROM customers WHERE id = ?`, + ) + .get(want); + if (row) return row; + row = db + .prepare( + `SELECT id, name, verae_user_id, balance_cents + FROM customers WHERE lower(name) = lower(?)`, + ) + .get(want); + return row || null; +} + +/** Read statement from billing.db (customers.balance_cents + usage_entries). */ +export function statementFromDb(customerId) { + const db = openDb(); + if (!db) return null; + try { + const cust = findCustomer(db, customerId); + if (!cust) { + return { + customerId, + prepaidCents: 0, + credits: [], + usage: [], + payments: [], + source: 'billing.db-miss', + }; + } + const usage = db + .prepare( + `SELECT customer_id, endpoint_id, cents, timestamp_ms + FROM usage_entries + WHERE customer_id = ? + ORDER BY id DESC + LIMIT 100`, + ) + .all(cust.id); + let payments = []; + try { + payments = db + .prepare( + `SELECT id, customer_id, period, status, billable_cents, issued_at_ms, paid_at_ms + FROM invoices + WHERE customer_id = ? AND status IN ('paid', 'issued') + ORDER BY COALESCE(paid_at_ms, issued_at_ms) DESC + LIMIT 100`, + ) + .all(cust.id); + } catch { + payments = []; + } + let credits = []; + try { + credits = db + .prepare( + `SELECT id, customer_id, cents, reason, agent, at_ms + FROM credit_adjustments + WHERE customer_id = ? + ORDER BY at_ms DESC + LIMIT 100`, + ) + .all(cust.id) + .map((r) => ({ + id: r.id, + customerId: r.customer_id, + cents: r.cents, + reason: r.reason, + agent: r.agent, + at: new Date(r.at_ms).toISOString(), + })); + } catch { + // Table may not exist until edge migrates schema on first credit write. + credits = []; + } + return { + customerId: cust.id, + name: cust.name, + veraeUserId: cust.verae_user_id || undefined, + prepaidCents: cust.balance_cents ?? 0, + credits, + usage: mapUsage(usage), + payments: mapPayments(payments), + source: 'billing.db', + }; + } finally { + db.close(); + } +} + +export async function statementFromEdge(customerId) { + const { ADMIN_KEY } = loadEdgeEnv(); + if (!ADMIN_KEY) return null; + try { + const r = await fetch(`${edgeBaseUrl()}/admin/api/statement/${encodeURIComponent(customerId)}`, { + headers: { 'x-admin-key': ADMIN_KEY }, + }); + if (!r.ok) { + process.stderr.write(`[facade] edge statement ${r.status} for ${customerId}\n`); + return null; + } + const body = await r.json(); + return { ...body, source: body.source ? `edge:${body.source}` : 'edge-admin' }; + } catch (err) { + process.stderr.write(`[facade] edge statement error: ${err.message}\n`); + return null; + } +} + +export async function getStatement(customerId) { + const fromDb = statementFromDb(customerId); + if (fromDb && fromDb.source === 'billing.db') return fromDb; + const fromEdge = await statementFromEdge(customerId); + if (fromEdge) return fromEdge; + return ( + fromDb || { + customerId, + prepaidCents: 0, + credits: [], + usage: [], + payments: [], + source: 'empty', + } + ); +} + +/** + * Prefer edge /admin/api/credits (authoritative writer under D006 freeze). + * Fall back to sudo -u verae-zapier sqlite UPDATE on billing.db. + */ +export async function adjustBalance(payload) { + const customerId = payload.customerId; + const delta = Math.trunc(Number(payload.cents) || 0); + const reason = payload.reason || payload.kind || 'credit'; + const agent = payload.agent || 'account-balance-facade'; + const { ADMIN_KEY } = loadEdgeEnv(); + + if (ADMIN_KEY) { + try { + const r = await fetch(`${edgeBaseUrl()}/admin/api/credits`, { + method: 'POST', + headers: { + 'content-type': 'application/json', + 'x-admin-key': ADMIN_KEY, + }, + body: JSON.stringify({ customerId, cents: delta, reason, agent }), + }); + if (r.ok) { + const rec = await r.json(); + process.stderr.write( + `[facade] adjust via edge credits customerId=${customerId} delta=${delta} prepaid=${rec.prepaidCents}\n`, + ); + return { + id: rec.id, + customerId: rec.customerId, + cents: rec.cents, + reason: rec.reason, + agent: rec.agent, + kind: payload.kind || 'credit', + at: rec.at, + prepaidCents: rec.prepaidCents, + source: 'edge-admin', + }; + } + process.stderr.write(`[facade] edge credits HTTP ${r.status}; trying billing.db write\n`); + } catch (err) { + process.stderr.write(`[facade] edge credits error: ${err.message}; trying billing.db write\n`); + } + } else { + process.stderr.write('[facade] no ADMIN_KEY; adjust via billing.db write\n'); + } + + return adjustViaSqliteCli({ customerId, delta, reason, agent, kind: payload.kind || 'credit' }); +} + +function adjustViaSqliteCli({ customerId, delta, reason, agent, kind }) { + const dbPath = billingDbPath(); + const safeId = String(customerId).replace(/'/g, "''"); + const safeReason = String(reason || 'credit').replace(/'/g, "''"); + const safeAgent = String(agent || 'account-balance-facade').replace(/'/g, "''"); + const adjId = `adj_${Date.now().toString(36)}`; + const atMs = Date.now(); + const sql = [ + `CREATE TABLE IF NOT EXISTS credit_adjustments ( + id TEXT PRIMARY KEY, + customer_id TEXT NOT NULL, + cents INTEGER NOT NULL, + reason TEXT NOT NULL, + agent TEXT NOT NULL, + at_ms INTEGER NOT NULL + );`, + `UPDATE customers SET balance_cents = balance_cents + (${delta}) WHERE id = '${safeId}';`, + `INSERT INTO credit_adjustments (id, customer_id, cents, reason, agent, at_ms) + VALUES ('${adjId}', '${safeId}', ${delta}, '${safeReason}', '${safeAgent}', ${atMs});`, + `SELECT id, balance_cents FROM customers WHERE id = '${safeId}';`, + ].join(' '); + const r = spawnSync( + 'sudo', + ['-n', '-u', 'verae-zapier', 'sqlite3', dbPath, sql], + { encoding: 'utf8' }, + ); + if (r.status !== 0) { + const msg = (r.stderr || r.stdout || 'sqlite adjust failed').trim(); + process.stderr.write(`[facade] billing.db write failed: ${msg}\n`); + throw new Error(`billing.db adjust failed: ${msg}`); + } + const line = (r.stdout || '').trim().split('\n').filter(Boolean).pop() || ''; + const parts = line.split('|'); + const prepaidCents = Number(parts[1] ?? 0); + process.stderr.write( + `[facade] adjust via billing.db customerId=${customerId} delta=${delta} prepaid=${prepaidCents} reason=${reason}\n`, + ); + return { + id: adjId, + customerId, + cents: delta, + reason, + agent, + kind, + at: new Date(atMs).toISOString(), + prepaidCents, + source: 'billing.db', + }; +} + +/** usage.recorded is telemetry only; edge meter already wrote usage_entries. */ +export function ignoreUsageRecorded(payload) { + process.stderr.write( + `[facade] usage.recorded noop (D006 edge SoT) customerId=${payload?.customerId} cents=${payload?.cents}\n`, + ); + return { + ok: true, + ignored: true, + reason: 'D006 edge BusinessStore meters usage_entries; account-balance is facade', + customerId: payload?.customerId, + source: 'facade-noop', + }; +} + +export async function putCustomer(payload) { + // Name/id already live on edge customers table; return current statement. + if (payload?.customerId) return getStatement(payload.customerId); + return { + customerId: payload?.customerId, + prepaidCents: 0, + credits: [], + usage: [], + payments: [], + source: 'facade', + }; +} diff --git a/src/books.js b/src/books.js index 5dee26b..9e42b67 100644 --- a/src/books.js +++ b/src/books.js @@ -1,5 +1,6 @@ /** - * Source of truth for prepaid balances, CS credits, usage, and payments. + * In-memory books helper for unit tests. + * Live SoT is edge BusinessStore / billing.db (D006 facade in billing-store.js). */ export class AccountBooks { constructor() { diff --git a/src/edge-env.js b/src/edge-env.js new file mode 100644 index 0000000..6f6579e --- /dev/null +++ b/src/edge-env.js @@ -0,0 +1,34 @@ +import fs from 'node:fs'; + +/** + * Load ADMIN_KEY (and friends) from EDGE_ENV_FILE without printing secrets. + * Fleet cannot read /SSD2/secrets; a mode-600 copy is used on the host. + */ +const DEFAULT_EDGE_ENV_FILE = + '/SSD3/moved/home-marchon/verae-stack/packages/zappier-account-balance/.env.edge'; + +export function loadEdgeEnv( + file = process.env.EDGE_ENV_FILE || DEFAULT_EDGE_ENV_FILE, +) { + const out = {}; + if (process.env.ADMIN_KEY) out.ADMIN_KEY = process.env.ADMIN_KEY; + if (!file || !fs.existsSync(file)) return out; + try { + for (const line of fs.readFileSync(file, 'utf8').split('\n')) { + const m = line.match(/^([A-Z0-9_]+)=(.*)$/); + if (!m) continue; + const key = m[1]; + let val = m[2]; + if ( + (val.startsWith('"') && val.endsWith('"')) || + (val.startsWith("'") && val.endsWith("'")) + ) { + val = val.slice(1, -1); + } + if (out[key] == null) out[key] = val; + } + } catch { + /* keep env-only */ + } + return out; +} diff --git a/src/persist.js b/src/persist.js index 9b857ff..46e5899 100644 --- a/src/persist.js +++ b/src/persist.js @@ -20,6 +20,10 @@ export function loadInto(books) { } export function saveFrom(books) { + // D006: books.json is not the SoT. Skip writes when the edge facade is active. + if (process.env.BILLING_DB || process.env.EDGE_URL || process.env.FACADE === '1') { + return; + } const p = booksPath(); fs.mkdirSync(path.dirname(p), { recursive: true }); const tmp = `${p}.tmp`; diff --git a/src/server.js b/src/server.js index c3cd64e..d6ac775 100644 --- a/src/server.js +++ b/src/server.js @@ -1,27 +1,48 @@ #!/usr/bin/env node +/** + * D006 facade over edge BusinessStore / billing.db. + * HTTP + NATS surface unchanged; books.json is no longer the SoT. + */ import http from 'node:http'; -import { AccountBooks, handle } from './books.js'; import { SUBJECTS } from './subjects.js'; -import { loadInto, saveFrom } from './persist.js'; +import { + adjustBalance, + billingDbPath, + edgeBaseUrl, + facadeEnabled, + getStatement, + ignoreUsageRecorded, + putCustomer, +} from './billing-store.js'; +import { loadEdgeEnv } from './edge-env.js'; const PORT = Number(process.env.PORT || process.env.FLEET_HEALTH_PORT || 3010); const BIND = process.env.FLEET_HEALTH_BIND || '0.0.0.0'; -const books = loadInto(new AccountBooks()); - -function apply(subject, payload) { - const out = handle(subject, payload, books); - saveFrom(books); - return out; -} let natsOk = false; +async function apply(subject, payload) { + const p = payload || {}; + if (subject.endsWith('customer.put')) return putCustomer(p); + if (subject.endsWith('balance.get') || subject.endsWith('statement.get')) { + return getStatement(p.customerId || p.name || p.id); + } + if (subject.endsWith('balance.adjust') || subject.endsWith('credit.applied')) { + return adjustBalance(p); + } + if (subject.endsWith('usage.recorded')) return ignoreUsageRecorded(p); + if (subject.endsWith('payment.recorded')) { + return adjustBalance({ ...p, kind: p.kind || 'payment' }); + } + throw new Error(`unknown billing subject ${subject}`); +} + async function startNats() { const url = process.env.NATS_URL; if (!url) return; const { connect, StringCodec } = await import('nats'); const nc = await connect({ servers: url.split(','), name: 'zappier-account-balance' }); const sc = StringCodec(); - const reply = async (sub, fn) => { + const reply = async (sub, subject) => { for await (const m of sub) { let payload = {}; try { @@ -29,30 +50,23 @@ async function startNats() { } catch { payload = {}; } - const out = fn(payload); - saveFrom(books); - if (m.reply) m.respond(sc.encode(JSON.stringify(out))); + try { + const out = await apply(subject, payload); + if (m.reply) m.respond(sc.encode(JSON.stringify(out))); + } catch (err) { + if (m.reply) m.respond(sc.encode(JSON.stringify({ error: err.message }))); + } } }; - reply(nc.subscribe(SUBJECTS.STATEMENT_GET, { queue: SUBJECTS.QUEUE }), (p) => - apply(SUBJECTS.STATEMENT_GET, p), - ); - reply(nc.subscribe(SUBJECTS.BALANCE_GET, { queue: SUBJECTS.QUEUE }), (p) => - apply(SUBJECTS.BALANCE_GET, p), - ); - reply(nc.subscribe(SUBJECTS.BALANCE_ADJUST, { queue: SUBJECTS.QUEUE }), (p) => - apply(SUBJECTS.BALANCE_ADJUST, p), - ); - reply(nc.subscribe(SUBJECTS.CUSTOMER_PUT, { queue: SUBJECTS.QUEUE }), (p) => - apply(SUBJECTS.CUSTOMER_PUT, p), - ); + reply(nc.subscribe(SUBJECTS.STATEMENT_GET, { queue: SUBJECTS.QUEUE }), SUBJECTS.STATEMENT_GET); + reply(nc.subscribe(SUBJECTS.BALANCE_GET, { queue: SUBJECTS.QUEUE }), SUBJECTS.BALANCE_GET); + reply(nc.subscribe(SUBJECTS.BALANCE_ADJUST, { queue: SUBJECTS.QUEUE }), SUBJECTS.BALANCE_ADJUST); + reply(nc.subscribe(SUBJECTS.CUSTOMER_PUT, { queue: SUBJECTS.QUEUE }), SUBJECTS.CUSTOMER_PUT); (async () => { for await (const m of nc.subscribe(SUBJECTS.USAGE_RECORDED)) { - apply(SUBJECTS.USAGE_RECORDED, JSON.parse(sc.decode(m.data) || '{}')); + await apply(SUBJECTS.USAGE_RECORDED, JSON.parse(sc.decode(m.data) || '{}')); } })(); - // payment.recorded / credit.applied are fan-out events. Prepaid mutations - // go through balance.adjust so publishers can both pub and request-reply. natsOk = true; process.stdout.write(`account-balance nats ${url}\n`); } @@ -79,19 +93,31 @@ const server = http.createServer(async (req, res) => { }; try { if (req.method === 'GET' && url.pathname === '/health') { - return json(200, { ok: true, role: 'zappier-account-balance', nats: natsOk, subjects: SUBJECTS }); + const edgeEnv = loadEdgeEnv(); + return json(200, { + ok: true, + role: 'zappier-account-balance', + facade: true, + d006: true, + nats: natsOk, + billingDb: billingDbPath(), + edgeUrl: edgeBaseUrl(), + adminKeyConfigured: Boolean(edgeEnv.ADMIN_KEY), + facadeEnabled: facadeEnabled(), + subjects: SUBJECTS, + }); } const st = url.pathname.match(/^\/statement\/([^/]+)$/); if (req.method === 'GET' && st) { - return json(200, books.lookup(decodeURIComponent(st[1]))); + return json(200, await getStatement(decodeURIComponent(st[1]))); } if (req.method === 'POST' && url.pathname === '/customer') { const body = await readBody(req); - return json(200, apply(SUBJECTS.CUSTOMER_PUT, body)); + return json(200, await apply(SUBJECTS.CUSTOMER_PUT, body)); } if (req.method === 'POST' && url.pathname === '/adjust') { const body = await readBody(req); - return json(200, apply(SUBJECTS.BALANCE_ADJUST, body)); + return json(200, await apply(SUBJECTS.BALANCE_ADJUST, body)); } json(404, { error: 'not found' }); } catch (err) { @@ -100,7 +126,9 @@ const server = http.createServer(async (req, res) => { }); server.listen(PORT, BIND, () => { - process.stdout.write(`zappier-account-balance http://${BIND}:${PORT}/\n`); + process.stdout.write( + `zappier-account-balance facade http://${BIND}:${PORT}/ billingDb=${billingDbPath()} edge=${edgeBaseUrl()}\n`, + ); }); startNats().catch((err) => { process.stderr.write(`nats optional: ${err.message}\n`); diff --git a/test/health.test.js b/test/health.test.js index 49bcc20..999f7b4 100644 --- a/test/health.test.js +++ b/test/health.test.js @@ -5,30 +5,69 @@ import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; import { fileURLToPath } from 'node:url'; +import { DatabaseSync } from 'node:sqlite'; const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..'); -test('account-balance health and statement', async () => { +function seedBillingDb(dbPath) { + const db = new DatabaseSync(dbPath); + db.exec(` + CREATE TABLE customers ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + verae_user_id TEXT, + balance_cents INTEGER NOT NULL DEFAULT 0 + ); + CREATE TABLE usage_entries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + customer_id TEXT NOT NULL, + endpoint_id TEXT NOT NULL, + cents INTEGER NOT NULL, + metadata_bytes INTEGER NOT NULL DEFAULT 0, + attachment_bytes INTEGER NOT NULL DEFAULT 0, + timestamp_ms INTEGER NOT NULL + ); + `); + db.prepare( + `INSERT INTO customers (id, name, verae_user_id, balance_cents) VALUES (?, ?, ?, ?)`, + ).run('cust_1', 'Ada (free)', 'vu_ada', 0); + const now = Date.now(); + const ins = db.prepare( + `INSERT INTO usage_entries (customer_id, endpoint_id, cents, timestamp_ms) VALUES (?, ?, ?, ?)`, + ); + for (let i = 0; i < 3; i++) ins.run('cust_1', 'timestamp', 10, now - i * 1000); + db.close(); +} + +test('account-balance health and statement from billing.db facade', async () => { const port = 18010; - const books = path.join(os.tmpdir(), `books-health-${Date.now()}.json`); + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ab-facade-')); + const billingDb = path.join(dir, 'billing.db'); + seedBillingDb(billingDb); const child = spawn(process.execPath, ['src/server.js'], { cwd: root, - env: { ...process.env, PORT: String(port), BOOKS_PATH: books }, + env: { + ...process.env, + PORT: String(port), + BILLING_DB: billingDb, + EDGE_URL: '', + FACADE: '1', + }, stdio: ['ignore', 'pipe', 'pipe'], }); - await new Promise((r) => setTimeout(r, 400)); + await new Promise((r) => setTimeout(r, 500)); try { const h = await (await fetch(`http://127.0.0.1:${port}/health`)).json(); assert.equal(h.role, 'zappier-account-balance'); - await fetch(`http://127.0.0.1:${port}/adjust`, { - method: 'POST', - headers: { 'content-type': 'application/json' }, - body: JSON.stringify({ customerId: 'c1', cents: 200, reason: 'test' }), - }); - const st = await (await fetch(`http://127.0.0.1:${port}/statement/c1`)).json(); - assert.equal(st.prepaidCents, 200); + assert.equal(h.facade, true); + const st = await (await fetch(`http://127.0.0.1:${port}/statement/cust_1`)).json(); + assert.equal(st.customerId, 'cust_1'); + assert.equal(st.name, 'Ada (free)'); + assert.equal(st.prepaidCents, 0); + assert.equal(st.usage.length, 3); + assert.equal(st.source, 'billing.db'); } finally { child.kill('SIGTERM'); - fs.rmSync(books, { force: true }); + fs.rmSync(dir, { recursive: true, force: true }); } }); From 23e32f3b8283bd85d70f0edb628d064bac2c7083 Mon Sep 17 00:00:00 2001 From: George Lambert Date: Mon, 12 Oct 2026 16:37:14 -0400 Subject: [PATCH 2/5] feat: JetStream pull consumer for VERAE_BILLING usage events Durable account-balance-usage on nats.verae.ai applies usage into billing.db idempotently by eventId. --- src/billing-js-consumer.js | 100 +++++++++++++++++++++++++++++++++++++ src/billing-store.js | 75 ++++++++++++++++++++++++---- src/server.js | 30 +++++++---- 3 files changed, 186 insertions(+), 19 deletions(-) create mode 100644 src/billing-js-consumer.js diff --git a/src/billing-js-consumer.js b/src/billing-js-consumer.js new file mode 100644 index 0000000..8224f13 --- /dev/null +++ b/src/billing-js-consumer.js @@ -0,0 +1,100 @@ +/** + * Durable JetStream **pull** consumer for VERAE_BILLING on nats.verae.ai. + */ +import { SUBJECTS } from './subjects.js'; +import { applyUsageEvent } from './billing-store.js'; + +const STREAM = 'VERAE_BILLING'; +const DURABLE = 'account-balance-usage'; +const FILTER = 'verae.billing.usage.>'; + +export function billingNatsUrl() { + return ( + process.env.VERAE_BILLING_NATS_URL || + process.env.NATS_URL || + 'nats://nats.verae.ai:4222' + ); +} + +function billingToken() { + return (process.env.VERAE_BILLING_NATS_TOKEN || process.env.NATS_TOKEN || '').trim(); +} + +/** + * @returns {Promise<{ ok: boolean, mode?: string }>} + */ +export async function startBillingJetStreamConsumer() { + const url = billingNatsUrl(); + if (!url) return { ok: false }; + const { connect, StringCodec } = await import('nats'); + const token = billingToken(); + const nc = await connect({ + servers: url.split(','), + name: 'zappier-account-balance-js', + ...(token ? { token } : {}), + }); + const sc = StringCodec(); + const jsm = await nc.jetstreamManager(); + try { + await jsm.streams.info(STREAM); + } catch { + await jsm.streams.add({ + name: STREAM, + subjects: ['verae.billing.usage.>'], + retention: 'limits', + storage: 'file', + max_age: 90 * 24 * 60 * 60 * 1e9, + discard: 'old', + num_replicas: 1, + }); + } + + try { + await jsm.consumers.info(STREAM, DURABLE); + } catch { + await jsm.consumers.add(STREAM, { + durable_name: DURABLE, + ack_policy: 'explicit', + deliver_policy: 'all', + filter_subject: FILTER, + max_ack_pending: 256, + // pull consumer: no deliver_subject + }); + } + + const js = nc.jetstream(); + const consumer = await js.consumers.get(STREAM, DURABLE); + + (async () => { + const messages = await consumer.consume({ max_messages: 100 }); + for await (const m of messages) { + try { + const payload = JSON.parse(sc.decode(m.data) || '{}'); + await applyUsageEvent(payload); + m.ack(); + } catch (err) { + process.stderr.write(`[billing-js] apply failed: ${err?.message || err}\n`); + try { + m.nak(); + } catch { + /* ignore */ + } + } + } + })().catch((err) => { + process.stderr.write(`[billing-js] consumer stopped: ${err?.message || err}\n`); + }); + + (async () => { + for await (const m of nc.subscribe(SUBJECTS.USAGE_RECORDED)) { + try { + await applyUsageEvent(JSON.parse(sc.decode(m.data) || '{}')); + } catch { + /* ignore */ + } + } + })(); + + process.stdout.write(`account-balance jetstream ${url} stream=${STREAM} durable=${DURABLE}\n`); + return { ok: true, mode: 'jetstream-pull' }; +} diff --git a/src/billing-store.js b/src/billing-store.js index 9bb6469..c954c5c 100644 --- a/src/billing-store.js +++ b/src/billing-store.js @@ -276,18 +276,73 @@ function adjustViaSqliteCli({ customerId, delta, reason, agent, kind }) { }; } -/** usage.recorded is telemetry only; edge meter already wrote usage_entries. */ +/** Legacy name — JetStream path applies usage into billing.db. */ export function ignoreUsageRecorded(payload) { - process.stderr.write( - `[facade] usage.recorded noop (D006 edge SoT) customerId=${payload?.customerId} cents=${payload?.cents}\n`, + return applyUsageEvent(payload); +} + +/** + * Apply a durable usage event (from timestamping-svc / peergos / edge) into billing.db. + * Idempotent on eventId. Runs as the service user (no sudo — NoNewPrivileges). + */ +export async function applyUsageEvent(payload) { + const customerId = payload?.customerId || payload?.customer_id; + if (!customerId) { + process.stderr.write('[facade] usage event missing customerId\n'); + return { ok: false, error: 'customerId required' }; + } + const cents = Number(payload.cents ?? 0); + const endpointId = String(payload.endpointId || payload.endpoint_id || 'timestamp'); + const eventId = String( + payload.eventId || payload.jobId || `${customerId}:${endpointId}:${payload.at || Date.now()}`, ); - return { - ok: true, - ignored: true, - reason: 'D006 edge BusinessStore meters usage_entries; account-balance is facade', - customerId: payload?.customerId, - source: 'facade-noop', - }; + const atMs = payload.at ? Date.parse(payload.at) : Date.now(); + const source = String(payload.source || 'stream'); + const dbPath = billingDbPath(); + const db = new DatabaseSync(dbPath); + try { + db.exec(` + CREATE TABLE IF NOT EXISTS billing_usage_events ( + event_id TEXT PRIMARY KEY, + customer_id TEXT NOT NULL, + endpoint_id TEXT NOT NULL, + cents INTEGER NOT NULL, + source TEXT, + at_ms INTEGER NOT NULL + ); + CREATE TABLE IF NOT EXISTS usage_entries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + customer_id TEXT NOT NULL, + endpoint_id TEXT NOT NULL, + cents INTEGER NOT NULL, + metadata_bytes INTEGER NOT NULL DEFAULT 0, + attachment_bytes INTEGER NOT NULL DEFAULT 0, + timestamp_ms INTEGER NOT NULL, + prev_hash TEXT, + row_hash TEXT + ); + `); + const ins = db.prepare( + `INSERT OR IGNORE INTO billing_usage_events (event_id, customer_id, endpoint_id, cents, source, at_ms) + VALUES (?, ?, ?, ?, ?, ?)`, + ); + const info = ins.run(eventId, customerId, endpointId, cents, source, Number.isFinite(atMs) ? atMs : Date.now()); + if (info.changes > 0) { + db.prepare( + `INSERT INTO usage_entries (customer_id, endpoint_id, cents, metadata_bytes, attachment_bytes, timestamp_ms) + VALUES (?, ?, ?, 0, 0, ?)`, + ).run(customerId, endpointId, cents, Number.isFinite(atMs) ? atMs : Date.now()); + if (cents !== 0) { + db.prepare(`UPDATE customers SET balance_cents = balance_cents - ? WHERE id = ?`).run(cents, customerId); + } + } + } finally { + db.close(); + } + process.stderr.write( + `[facade] usage applied customerId=${customerId} endpoint=${endpointId} cents=${cents} source=${source} eventId=${eventId}\n`, + ); + return { ok: true, customerId, endpointId, cents, eventId, source: 'billing-js' }; } export async function putCustomer(payload) { diff --git a/src/server.js b/src/server.js index d6ac775..d33e395 100644 --- a/src/server.js +++ b/src/server.js @@ -7,14 +7,15 @@ import http from 'node:http'; import { SUBJECTS } from './subjects.js'; import { adjustBalance, + applyUsageEvent, billingDbPath, edgeBaseUrl, facadeEnabled, getStatement, - ignoreUsageRecorded, putCustomer, } from './billing-store.js'; import { loadEdgeEnv } from './edge-env.js'; +import { startBillingJetStreamConsumer } from './billing-js-consumer.js'; const PORT = Number(process.env.PORT || process.env.FLEET_HEALTH_PORT || 3010); const BIND = process.env.FLEET_HEALTH_BIND || '0.0.0.0'; @@ -29,7 +30,7 @@ async function apply(subject, payload) { if (subject.endsWith('balance.adjust') || subject.endsWith('credit.applied')) { return adjustBalance(p); } - if (subject.endsWith('usage.recorded')) return ignoreUsageRecorded(p); + if (subject.includes('usage.')) return applyUsageEvent(p); if (subject.endsWith('payment.recorded')) { return adjustBalance({ ...p, kind: p.kind || 'payment' }); } @@ -37,10 +38,26 @@ async function apply(subject, payload) { } async function startNats() { - const url = process.env.NATS_URL; + // Prefer private Verae JetStream (nats.verae.ai) for durable usage. + try { + const js = await startBillingJetStreamConsumer(); + if (js.ok) natsOk = true; + } catch (err) { + process.stderr.write(`[account-balance] jetstream consumer failed: ${err.message}\n`); + } + + const url = + process.env.VERAE_BILLING_NATS_URL || + process.env.NATS_URL || + 'nats://nats.verae.ai:4222'; if (!url) return; const { connect, StringCodec } = await import('nats'); - const nc = await connect({ servers: url.split(','), name: 'zappier-account-balance' }); + const token = (process.env.VERAE_BILLING_NATS_TOKEN || process.env.NATS_TOKEN || '').trim(); + const nc = await connect({ + servers: url.split(','), + name: 'zappier-account-balance', + ...(token ? { token } : {}), + }); const sc = StringCodec(); const reply = async (sub, subject) => { for await (const m of sub) { @@ -62,11 +79,6 @@ async function startNats() { reply(nc.subscribe(SUBJECTS.BALANCE_GET, { queue: SUBJECTS.QUEUE }), SUBJECTS.BALANCE_GET); reply(nc.subscribe(SUBJECTS.BALANCE_ADJUST, { queue: SUBJECTS.QUEUE }), SUBJECTS.BALANCE_ADJUST); reply(nc.subscribe(SUBJECTS.CUSTOMER_PUT, { queue: SUBJECTS.QUEUE }), SUBJECTS.CUSTOMER_PUT); - (async () => { - for await (const m of nc.subscribe(SUBJECTS.USAGE_RECORDED)) { - await apply(SUBJECTS.USAGE_RECORDED, JSON.parse(sc.decode(m.data) || '{}')); - } - })(); natsOk = true; process.stdout.write(`account-balance nats ${url}\n`); } From 8dd221e711deca2ecadf0356ed5d9c2644885c8b Mon Sep 17 00:00:00 2001 From: George Lambert Date: Mon, 12 Oct 2026 16:40:09 -0400 Subject: [PATCH 3/5] fix(billing): integrate JetStream usage into customer ledger Resolve customer by id/veraeUserId/email; default cents from rate card; source=edge is audit-only (no double usage_entries); other sources write usage_entries and draw prepaid balance. --- src/billing-store.js | 122 ++++++++++++++++++++++++++++++++++++------- 1 file changed, 104 insertions(+), 18 deletions(-) diff --git a/src/billing-store.js b/src/billing-store.js index c954c5c..c25e733 100644 --- a/src/billing-store.js +++ b/src/billing-store.js @@ -281,25 +281,77 @@ export function ignoreUsageRecorded(payload) { return applyUsageEvent(payload); } +/** Default list credits when publisher omits cents (matches edge DEFAULT_CHANNEL_SKU_CREDITS). */ +const DEFAULT_CENTS = { + timestamp: 10, + 'timestamp-batch': 10, + 'timestamp-metadata': 15, + 'timestamp-attachment': 25, + 'timestamp-re': 5, + 'timestamp-verify': 0, + 'storage-gb-month': 1, + 'storage-gb-month-byo': 1, +}; + +function defaultCentsFor(endpointId, existing, units) { + if (existing) return 0; + const per = DEFAULT_CENTS[endpointId] ?? DEFAULT_CENTS.timestamp; + const n = Number(units); + return per * (Number.isFinite(n) && n > 0 ? Math.ceil(n) : 1); +} + +function resolveCustomerId(db, payload) { + let id = payload?.customerId || payload?.customer_id || ''; + if (id) { + const row = db.prepare(`SELECT id FROM customers WHERE id = ?`).get(id); + if (row) return row.id; + } + const vuid = payload?.veraeUserId || payload?.verae_user_id; + if (vuid) { + const row = db + .prepare(`SELECT id FROM customers WHERE verae_user_id = ? LIMIT 1`) + .get(String(vuid)); + if (row) return row.id; + } + const uname = payload?.veraeUsername || payload?.username; + if (uname) { + const row = db + .prepare(`SELECT id FROM customers WHERE lower(verae_username) = lower(?) OR lower(email) = lower(?) LIMIT 1`) + .get(String(uname), String(uname)); + if (row) return row.id; + } + return id || ''; +} + /** - * Apply a durable usage event (from timestamping-svc / peergos / edge) into billing.db. - * Idempotent on eventId. Runs as the service user (no sudo — NoNewPrivileges). + * Apply a durable usage event into the customer billing solution (billing.db). + * + * - source=edge: edge already wrote usage_entries via meter(); only audit the stream event + * (avoids double-count). Still records billing_usage_events for the JetStream ledger. + * - source=timestamping-svc|peergos|mcp|manual|…: write usage_entries + optional balance drawdown. + * - Idempotent on eventId. + * - Resolves customerId from veraeUserId / email when needed. + * - Fills cents from default rate card when omitted (API publishers often send units only). */ export async function applyUsageEvent(payload) { - const customerId = payload?.customerId || payload?.customer_id; - if (!customerId) { - process.stderr.write('[facade] usage event missing customerId\n'); - return { ok: false, error: 'customerId required' }; - } - const cents = Number(payload.cents ?? 0); const endpointId = String(payload.endpointId || payload.endpoint_id || 'timestamp'); - const eventId = String( - payload.eventId || payload.jobId || `${customerId}:${endpointId}:${payload.at || Date.now()}`, - ); - const atMs = payload.at ? Date.parse(payload.at) : Date.now(); + const existing = Boolean(payload.existing); const source = String(payload.source || 'stream'); + const units = Number(payload.units ?? 1); + let cents = payload.cents; + if (cents === undefined || cents === null || cents === '') { + cents = defaultCentsFor(endpointId, existing, units); + } else { + cents = Number(cents); + if (!Number.isFinite(cents)) cents = defaultCentsFor(endpointId, existing, units); + } + const dbPath = billingDbPath(); const db = new DatabaseSync(dbPath); + let customerId = ''; + let eventId = ''; + let applied = false; + let auditOnly = false; try { db.exec(` CREATE TABLE IF NOT EXISTS billing_usage_events ( @@ -322,27 +374,61 @@ export async function applyUsageEvent(payload) { row_hash TEXT ); `); + customerId = resolveCustomerId(db, payload); + if (!customerId) { + process.stderr.write( + `[facade] usage event missing resolvable customerId veraeUserId=${payload?.veraeUserId || ''}\n`, + ); + return { ok: false, error: 'customerId required' }; + } + eventId = String( + payload.eventId || + payload.jobId || + `${customerId}:${endpointId}:${payload.digest || payload.at || Date.now()}`, + ); + const atMs = payload.at ? Date.parse(payload.at) : Date.now(); + const ts = Number.isFinite(atMs) ? atMs : Date.now(); + const ins = db.prepare( `INSERT OR IGNORE INTO billing_usage_events (event_id, customer_id, endpoint_id, cents, source, at_ms) VALUES (?, ?, ?, ?, ?, ?)`, ); - const info = ins.run(eventId, customerId, endpointId, cents, source, Number.isFinite(atMs) ? atMs : Date.now()); - if (info.changes > 0) { + const info = ins.run(eventId, customerId, endpointId, cents, source, ts); + if (info.changes === 0) { + return { ok: true, customerId, endpointId, cents, eventId, duplicate: true, source: 'billing-js' }; + } + + // Edge meter already appended usage_entries — stream is the durable bus, not a second write. + auditOnly = source === 'edge'; + if (!auditOnly) { db.prepare( `INSERT INTO usage_entries (customer_id, endpoint_id, cents, metadata_bytes, attachment_bytes, timestamp_ms) VALUES (?, ?, ?, 0, 0, ?)`, - ).run(customerId, endpointId, cents, Number.isFinite(atMs) ? atMs : Date.now()); + ).run(customerId, endpointId, cents, ts); if (cents !== 0) { - db.prepare(`UPDATE customers SET balance_cents = balance_cents - ? WHERE id = ?`).run(cents, customerId); + db.prepare(`UPDATE customers SET balance_cents = balance_cents - ? WHERE id = ?`).run( + cents, + customerId, + ); } + applied = true; } } finally { db.close(); } process.stderr.write( - `[facade] usage applied customerId=${customerId} endpoint=${endpointId} cents=${cents} source=${source} eventId=${eventId}\n`, + `[facade] usage ${auditOnly ? 'audited' : 'applied'} customerId=${customerId} endpoint=${endpointId} cents=${cents} source=${source} eventId=${eventId}\n`, ); - return { ok: true, customerId, endpointId, cents, eventId, source: 'billing-js' }; + return { + ok: true, + customerId, + endpointId, + cents, + eventId, + applied, + auditOnly, + source: 'billing-js', + }; } export async function putCustomer(payload) { From c75fa5f5fe453d80cb78f725d54e600bf2b992f8 Mon Sep 17 00:00:00 2001 From: George Lambert Date: Mon, 12 Oct 2026 16:52:21 -0400 Subject: [PATCH 4/5] feat(billing): default cents for readonly/lookup/history/scope SKUs --- src/billing-store.js | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/billing-store.js b/src/billing-store.js index c25e733..b3b2244 100644 --- a/src/billing-store.js +++ b/src/billing-store.js @@ -288,7 +288,15 @@ const DEFAULT_CENTS = { 'timestamp-metadata': 15, 'timestamp-attachment': 25, 'timestamp-re': 5, + 'timestamp-metadata-update': 15, 'timestamp-verify': 0, + 'timestamp-readonly': 0, + 'timestamp-lookup': 0, + 'timestamp-scope-public': 0, + 'timestamp-scope-private': 3, + 'timestamp-customer-id': 2, + 'timestamp-history': 5, + 'timestamp-history-item': 2, 'storage-gb-month': 1, 'storage-gb-month-byo': 1, }; From 1d0db82ccf6e49d8d1887b4f3f67facd10d936e7 Mon Sep 17 00:00:00 2001 From: George Lambert Date: Mon, 12 Oct 2026 19:24:30 -0400 Subject: [PATCH 5/5] docs: Standard Readme update for production board pack --- README.md | 60 +++++++++++++++++++------------------------------------ 1 file changed, 20 insertions(+), 40 deletions(-) diff --git a/README.md b/README.md index a8a5216..1763623 100644 --- a/README.md +++ b/README.md @@ -2,62 +2,42 @@ [![standard-readme compliant](https://img.shields.io/badge/readme%20style-standard-brightgreen.svg?style=flat-square)](https://github.com/RichardLitt/standard-readme) -NATS/HTTP facade for customer prepaid balances over edge BusinessStore (D006). +Durable JetStream consumer applying VERAE_BILLING usage into the shared customer ledger. -**Forgejo:** https://git.georgelambert.org/marchon/zappier-account-balance -**Catalog:** https://zapier.georgelambert.org/packages/zappier-account-balance/README.pdf +## Table of Contents -Under decision **D006**, this package is a **facade**, not a competing system of record. Lab SoT is zappier-edge `billing.db` (`customers.balance_cents` + `usage_entries`); multi-host SoT becomes **libSQL** later. `books.json` is no longer written as SoT. +- [Background](#background) +- [Install](#install) +- [Usage](#usage) +- [Contributing](#contributing) +- [License](#license) -Zapier cloud never connects here. zappier-edge, customer-service, sales-pricing, and accounting-export do. +## Background + +Consumes stream VERAE_BILLING (subjects verae.billing.usage.>) on the private Verae NATS edge. +Writes usage_entries and prepaid balance drawdowns into billing.db (SoT for portal/CS/sales statements). +Job-complete events stay on core NATS — not JetStream. + + +**Co-contributors:** Grok (xAI), George Lambert, Scott "R00tPiG Security" Lindsey — Verae Inc. ([verae.com](https://verae.com)). ## Install ```sh -git clone ssh://git@git.georgelambert.org:2223/marchon/zappier-account-balance.git -cd zappier-account-balance -npm install +npm ci ``` ## Usage ```sh -EDGE_URL=http://127.0.0.1:3040 \ -BILLING_DB=/opt/verae/zappier-edge/billing.db \ -EDGE_ENV_FILE=./.env.edge \ -NATS_URL=nats://127.0.0.1:4222 \ -PORT=3010 npm start - -curl http://127.0.0.1:3010/health -curl http://127.0.0.1:3010/statement/cust_1 +NATS_URL=nats://nats.verae.ai:4222 node src/server.js ``` -Statement JSON: `{ customerId, prepaidCents, credits[], usage[], payments[], source }`. -`balance.get` / `statement.get` read `billing.db` (fallback: edge `/admin/api/statement/:id`). -`balance.adjust` proxies to edge `POST /admin/api/credits` (fallback: `sudo -u verae-zapier` sqlite write). -`usage.recorded` is a no-op; edge meter already appends `usage_entries`. - -## Subjects - -| Subject | Kind | Who | -|---------|------|-----| -| `verae.billing.statement.get` | request-reply | portal, CS, sales, admin | -| `verae.billing.balance.get` | request-reply | anyone internal | -| `verae.billing.balance.adjust` | request-reply | CS credits, portal reload | -| `verae.billing.usage.recorded` | pub | zappier-edge meter (ignored by facade) | -| `verae.billing.payment.recorded` | pub | portal reload, invoice paid | -| `verae.billing.credit.applied` | pub | CS | - -Queue group: `account-balance`. - -## Background - -D006 freeze: edge BusinessStore is the authoritative meter/balance writer on the single-host lab. This service keeps the historical NATS/HTTP surface so callers do not break while prepaid moves to libSQL. - ## Contributing -Questions: open an issue at https://git.georgelambert.org/marchon/zappier-account-balance/issues. Pull requests are accepted. +Questions: open an issue on the project Forgejo/GitHub remote, or contact Verae engineering. +PRs are accepted when they include tests and do not leak secrets. ## License -UNLICENSED © George Lambert +UNLICENSED © Verae Inc. (verae.com)