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.

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
 

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,359 @@
+/**
+ * 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);
+}
+
+/**
+ * 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()}`,
+  );
+  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) {
+  // 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);
-      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),
-  );
-  (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 });
   }
 });