merge: account-balance main with JetStream VERAE_BILLING consumer

Includes billing-jetstream work on main for ns1 production deploy.
Co-authored ops credit: Scott R00tPiG Lindsey (rootpig).

README.md

@@ -2,54 +2,42 @@
 
 [![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...
+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
 
-Zapier cloud never connects here. verae-edge-billing, customer-service, sales-pricing, and accounting-export do.
+- [Background](#background)
+- [Install](#install)
+- [Usage](#usage)
+- [Contributing](#contributing)
+- [License](#license)
 
-## Subjects
+## Background
 
-| 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 | verae-edge-billing meter |
-| `verae.billing.payment.recorded` | pub | portal reload, invoice paid |
-| `verae.billing.credit.applied` | pub | CS |
+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.
 
-Queue group: `account-balance`.
 
-## Run
-
-```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[] }`.
+**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 ci
 ```
 
 ## Usage
 
 ```sh
-npm install
-npm start
+NATS_URL=nats://nats.verae.ai:4222 node src/server.js
 ```
 
 ## 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)

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"

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' };
+}

src/billing-store.js

@@ -0,0 +1,453 @@
+/**
+ * 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',
+  };
+}
+
+/** Legacy name — JetStream path applies usage into billing.db. */
+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-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,
+};
+
+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 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 endpointId = String(payload.endpointId || payload.endpoint_id || 'timestamp');
+  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 (
+        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
+      );
+    `);
+    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, 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, ts);
+      if (cents !== 0) {
+        db.prepare(`UPDATE customers SET balance_cents = balance_cents - ? WHERE id = ?`).run(
+          cents,
+          customerId,
+        );
+      }
+      applied = true;
+    }
+  } finally {
+    db.close();
+  }
+  process.stderr.write(
+    `[facade] usage ${auditOnly ? 'audited' : 'applied'} customerId=${customerId} endpoint=${endpointId} cents=${cents} source=${source} eventId=${eventId}\n`,
+  );
+  return {
+    ok: true,
+    customerId,
+    endpointId,
+    cents,
+    eventId,
+    applied,
+    auditOnly,
+    source: 'billing-js',
+  };
+}
+
+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',
+  };
+}

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() {

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;
+}

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`;

src/server.js

@@ -1,27 +1,65 @@
 #!/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,
+  applyUsageEvent,
+  billingDbPath,
+  edgeBaseUrl,
+  facadeEnabled,
+  getStatement,
+  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';
-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.includes('usage.')) return applyUsageEvent(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;
+  // 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, fn) => {
+  const reply = async (sub, subject) => {
     for await (const m of sub) {
       let payload = {};
       try {
@@ -29,30 +67,18 @@ async function startNats() {
       } catch {
         payload = {};
       }
-      const out = fn(payload);
-      saveFrom(books);
+      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),
-  );
-  (async () => {
-    for await (const m of nc.subscribe(SUBJECTS.USAGE_RECORDED)) {
-      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.
+  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);
   natsOk = true;
   process.stdout.write(`account-balance nats ${url}\n`);
 }
@@ -79,19 +105,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 +138,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`);

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 });
   }
 });