Clean prepaid SoT, identity mailbox, public access planes, leaf policy, fleet spawn

Persist account-balance books; edge caches prepaid from books. Add zappier-identity, verae-nats-accounts, verae-jobs-events, verae-access-staff, zapier-decisions. Edge binds loopback; lan-134 stays off; HTTP services prefer local spawn.

packages/zappier-account-balance/README.md

@@ -1,6 +1,6 @@
 # zappier-account-balance
 
-Source of truth for **customer prepaid balances**. Internal traffic is **NATS request-reply**; HTTP is health plus a fallback.
+Source of truth for **customer prepaid balances** (persisted `books.json`). Internal traffic is **NATS request-reply**; HTTP is health plus a fallback. zappier-edge `balanceCents` is a cache of this book.
 
 **Forgejo:** https://git.georgelambert.org/marchon/zappier-account-balance  
 **Catalog:** https://zapier.georgelambert.org/packages/zappier-account-balance/README.pdf

packages/zappier-account-balance/src/books.js

@@ -78,6 +78,26 @@ export class AccountBooks {
       payments: match(this.payments),
     };
   }
+
+  dump() {
+    return {
+      prepaid: this.prepaid,
+      veraeUserIds: this.veraeUserIds,
+      credits: this.credits,
+      usage: this.usage,
+      payments: this.payments,
+    };
+  }
+
+  load(raw) {
+    if (!raw || typeof raw !== 'object') return this;
+    this.prepaid = raw.prepaid || {};
+    this.veraeUserIds = raw.veraeUserIds || {};
+    this.credits = Array.isArray(raw.credits) ? raw.credits : [];
+    this.usage = Array.isArray(raw.usage) ? raw.usage : [];
+    this.payments = Array.isArray(raw.payments) ? raw.payments : [];
+    return this;
+  }
 }
 
 export function handle(subject, payload, books) {

packages/zappier-account-balance/src/persist.js

@@ -0,0 +1,28 @@
+import fs from 'node:fs';
+import path from 'node:path';
+
+export function booksPath() {
+  return (
+    process.env.BOOKS_PATH ||
+    path.join(process.env.FLEET_STATE_DIR || process.cwd(), 'data', 'books.json')
+  );
+}
+
+export function loadInto(books) {
+  const p = booksPath();
+  if (!fs.existsSync(p)) return books;
+  try {
+    books.load(JSON.parse(fs.readFileSync(p, 'utf8')));
+  } catch {
+    /* keep empty */
+  }
+  return books;
+}
+
+export function saveFrom(books) {
+  const p = booksPath();
+  fs.mkdirSync(path.dirname(p), { recursive: true });
+  const tmp = `${p}.tmp`;
+  fs.writeFileSync(tmp, JSON.stringify(books.dump(), null, 2));
+  fs.renameSync(tmp, p);
+}

packages/zappier-account-balance/src/server.js

@@ -2,10 +2,17 @@
 import http from 'node:http';
 import { AccountBooks, handle } from './books.js';
 import { SUBJECTS } from './subjects.js';
+import { loadInto, saveFrom } from './persist.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 = new AccountBooks();
+const books = loadInto(new AccountBooks());
+
+function apply(subject, payload) {
+  const out = handle(subject, payload, books);
+  saveFrom(books);
+  return out;
+}
 let natsOk = false;
 
 async function startNats() {
@@ -23,21 +30,22 @@ async function startNats() {
         payload = {};
       }
       const out = fn(payload);
+      saveFrom(books);
       if (m.reply) m.respond(sc.encode(JSON.stringify(out)));
     }
   };
   reply(nc.subscribe(SUBJECTS.STATEMENT_GET, { queue: SUBJECTS.QUEUE }), (p) =>
-    handle(SUBJECTS.STATEMENT_GET, p, books),
+    apply(SUBJECTS.STATEMENT_GET, p),
   );
   reply(nc.subscribe(SUBJECTS.BALANCE_GET, { queue: SUBJECTS.QUEUE }), (p) =>
-    handle(SUBJECTS.BALANCE_GET, p, books),
+    apply(SUBJECTS.BALANCE_GET, p),
   );
   reply(nc.subscribe(SUBJECTS.BALANCE_ADJUST, { queue: SUBJECTS.QUEUE }), (p) =>
-    handle(SUBJECTS.BALANCE_ADJUST, p, books),
+    apply(SUBJECTS.BALANCE_ADJUST, p),
   );
   (async () => {
     for await (const m of nc.subscribe(SUBJECTS.USAGE_RECORDED)) {
-      handle(SUBJECTS.USAGE_RECORDED, JSON.parse(sc.decode(m.data) || '{}'), books);
+      apply(SUBJECTS.USAGE_RECORDED, JSON.parse(sc.decode(m.data) || '{}'));
     }
   })();
   // payment.recorded / credit.applied are fan-out events. Prepaid mutations
@@ -76,7 +84,7 @@ const server = http.createServer(async (req, res) => {
     }
     if (req.method === 'POST' && url.pathname === '/adjust') {
       const body = await readBody(req);
-      return json(200, books.adjust(body));
+      return json(200, apply(SUBJECTS.BALANCE_ADJUST, body));
     }
     json(404, { error: 'not found' });
   } catch (err) {

packages/zappier-account-balance/test/books.test.js

@@ -1,7 +1,11 @@
 import { test } from 'node:test';
 import assert from 'node:assert/strict';
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
 import { AccountBooks, handle } from '../src/books.js';
 import { SUBJECTS } from '../src/subjects.js';
+import { loadInto, saveFrom, booksPath } from '../src/persist.js';
 
 test('adjust credits prepaid and statement lists credits usage payments', () => {
   const books = new AccountBooks();
@@ -19,3 +23,15 @@ test('adjust credits prepaid and statement lists credits usage payments', () =>
   assert.equal(st.usage[0].endpointId, 'timestamp');
   assert.equal(st.payments[0].kind, 'payment');
 });
+
+test('persist round-trip keeps prepaid', () => {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'books-'));
+  process.env.BOOKS_PATH = path.join(dir, 'books.json');
+  const a = new AccountBooks();
+  handle(SUBJECTS.BALANCE_ADJUST, { customerId: 'c9', cents: 120, kind: 'reload' }, a);
+  saveFrom(a);
+  assert.ok(fs.existsSync(booksPath()));
+  const b = loadInto(new AccountBooks());
+  assert.equal(b.prepaidCents('c9'), 120);
+  delete process.env.BOOKS_PATH;
+});