Add NATS account-balance SoT and statement review for customers, CS, and sales

Internal billing now uses verae.billing.* request-reply and pubs. zappier-account-balance tracks prepaid, credits, usage, and payments. Portal, admin, CS, and sales all review the same statement. Independent Forgejo repos stay split via push-module-repos.

packages/zappier/src/admin.ts

@@ -12,6 +12,8 @@ import { billingRows, toCsv, usageTrend } from './reports';
 import { UsageRepo } from './usage';
 import { CreditLedger } from './credits';
 import { invoicesToAccountingCsv, invoicesToQuickBooksIif } from './accounting-export';
+import { composeStatement } from './statement';
+import { BILLING_SUBJECTS, natsAdjust, natsPublish, natsStatement } from './billing-nats';
 
 // Issued login tokens (in-memory; a restart simply requires logging in again).
 const sessions = new Map<string, number>();
@@ -246,6 +248,14 @@ export function adminRouter(
       agent: typeof agent === 'string' ? agent : 'admin',
     });
     customers.save({ ...customer, balanceCents: (customer.balanceCents ?? 0) + delta });
+    natsPublish(BILLING_SUBJECTS.CREDIT_APPLIED, rec);
+    void natsAdjust({
+      customerId,
+      cents: delta,
+      reason: rec.reason,
+      agent: rec.agent,
+      kind: 'credit',
+    });
     res.status(201).json(rec);
   });
 
@@ -254,6 +264,31 @@ export function adminRouter(
     res.json({ credits: credits.list(customerId) });
   });
 
+  router.get('/statement/:id', async (req, res) => {
+    const customer = customers.list().find((c) => c.id === req.params.id);
+    if (!customer) {
+      res.status(404).json({ error: 'customer not found' });
+      return;
+    }
+    const fromNats = await natsStatement(customer.id);
+    if (fromNats) {
+      res.json({ ...fromNats, name: customer.name, tierId: customer.tierId, source: 'nats' });
+      return;
+    }
+    res.json({
+      ...composeStatement({
+        customerId: customer.id,
+        prepaidCents: customer.balanceCents ?? 0,
+        credits: credits.list(customer.id),
+        usage: accounting.usage.listFor(customer.id),
+        invoices: accounting.invoices.list({ customerId: customer.id }),
+      }),
+      name: customer.name,
+      tierId: customer.tierId,
+      source: 'local',
+    });
+  });
+
   router.get('/sales/quote/:id', (req, res) => {
     const customer = customers.list().find((c) => c.id === req.params.id);
     if (!customer) {

packages/zappier/src/app.ts

@@ -26,6 +26,9 @@ import { InMemoryInvoiceRepo, InvoiceRepo } from './invoicing';
 import { PROJECT_ROOT } from './paths';
 import { PaymentClient, portalRouter } from './portal';
 import { proxyVerae } from './upstream';
+import { CreditLedger } from './credits';
+import { composeStatement } from './statement';
+import { BILLING_SUBJECTS, natsAdjust, natsPublish, natsStatement } from './billing-nats';
 
 export interface StoredItem {
   id: string;
@@ -98,6 +101,9 @@ export function buildApp(deps: AppDeps = {}): {
     },
   };
   const items: StoredItem[] = [];
+  const credits = new CreditLedger();
+  const onUsage = (e: { customerId: string; endpointId: string; cents: number }) =>
+    natsPublish(BILLING_SUBJECTS.USAGE_RECORDED, { ...e, at: new Date().toISOString() });
   const hashIndex = new Map<
     string,
     { jobId: string; sha256: string; data?: string; timestamp: string }
@@ -114,7 +120,7 @@ export function buildApp(deps: AppDeps = {}): {
     '/admin/api',
     adminLoginRouter(adminUsers),
     adminAuth(),
-    adminRouter(pricingStore, customers, { usage, invoices, users: adminUsers }),
+    adminRouter(pricingStore, customers, { usage, invoices, users: adminUsers }, credits),
   );
   app.use('/admin', express.static(path.join(PROJECT_ROOT, 'admin')));
 
@@ -140,6 +146,7 @@ export function buildApp(deps: AppDeps = {}): {
       rateCard: () => pricingStore.getRateCard(),
       payments,
       qr,
+      credits,
     }),
   );
   app.use('/portal', express.static(path.join(PROJECT_ROOT, 'portal')));
@@ -158,16 +165,16 @@ export function buildApp(deps: AppDeps = {}): {
     }),
   );
 
-  app.get('/v1/status', meter('status', usage, pricing), (req, res) => {
+  app.get('/v1/status', meter('status', usage, pricing, onUsage), (req, res) => {
     res.json({ status: 'ok', quote: res.locals.quote });
   });
 
-  app.post('/v1/transform', meter('transform', usage, pricing), (req, res) => {
+  app.post('/v1/transform', meter('transform', usage, pricing, onUsage), (req, res) => {
     const text = String(req.body?.text ?? '');
     res.json({ output: text.toUpperCase(), quote: res.locals.quote });
   });
 
-  app.post('/v1/timestamp', meter('timestamp', usage, pricing), async (req, res) => {
+  app.post('/v1/timestamp', meter('timestamp', usage, pricing, onUsage), async (req, res) => {
     if (await proxyVerae(req, res, '/zapier/v1/timestamp')) return;
     const data = req.body?.data != null ? String(req.body.data) : '';
     const sha256 =
@@ -190,7 +197,7 @@ export function buildApp(deps: AppDeps = {}): {
     res.status(202).json({ jobId, sha256, existing: false, timestamp });
   });
 
-  app.get('/v1/receipts/:jobId', meter('receipt', usage, pricing), async (req, res) => {
+  app.get('/v1/receipts/:jobId', meter('receipt', usage, pricing, onUsage), async (req, res) => {
     if (await proxyVerae(req, res, `/zapier/v1/receipts/${req.params.jobId}`)) return;
     const rec = jobIndex.get(String(req.params.jobId));
     if (!rec) {
@@ -208,7 +215,7 @@ export function buildApp(deps: AppDeps = {}): {
     });
   });
 
-  app.get('/v1/hashes/:sha256', meter('hash-lookup', usage, pricing), async (req, res) => {
+  app.get('/v1/hashes/:sha256', meter('hash-lookup', usage, pricing, onUsage), async (req, res) => {
     if (await proxyVerae(req, res, `/zapier/v1/hashes/${req.params.sha256}`)) return;
     const sha256 = String(req.params.sha256 || '').toLowerCase();
     const rec = hashIndex.get(sha256);
@@ -219,7 +226,7 @@ export function buildApp(deps: AppDeps = {}): {
     res.json({ exists: true, ...rec });
   });
 
-  app.post('/v1/add', meter('add', usage, pricing), (req, res) => {
+  app.post('/v1/add', meter('add', usage, pricing, onUsage), (req, res) => {
     const number1 = Number(req.body?.number1);
     const number2 = Number(req.body?.number2);
     if (!Number.isFinite(number1) || !Number.isFinite(number2)) {
@@ -229,7 +236,7 @@ export function buildApp(deps: AppDeps = {}): {
     res.json({ number1, number2, sum: number1 + number2, quote: res.locals.quote });
   });
 
-  app.post('/v1/storage', parseMetadata, meter('storage', usage, pricing), (req, res) => {
+  app.post('/v1/storage', parseMetadata, meter('storage', usage, pricing, onUsage), (req, res) => {
     const metadata = res.locals.parsedMetadata as Record<string, unknown>;
     const files = (req.files as Express.Multer.File[]) ?? [];
     const item: StoredItem = {
@@ -243,7 +250,7 @@ export function buildApp(deps: AppDeps = {}): {
     res.json({ id: item.id, quote: res.locals.quote });
   });
 
-  app.get('/v1/storage', meter('storage-list', usage, pricing), (req, res) => {
+  app.get('/v1/storage', meter('storage-list', usage, pricing, onUsage), (req, res) => {
     res.json({ items: items.filter((i) => i.customerId === req.customer!.id) });
   });
 

packages/zappier/src/billing-nats.ts

@@ -0,0 +1,73 @@
+/** Internal NATS billing bus. No-op when NATS_URL is unset (tests). */
+
+export const BILLING_SUBJECTS = {
+  STATEMENT_GET: 'verae.billing.statement.get',
+  BALANCE_ADJUST: 'verae.billing.balance.adjust',
+  USAGE_RECORDED: 'verae.billing.usage.recorded',
+  PAYMENT_RECORDED: 'verae.billing.payment.recorded',
+  CREDIT_APPLIED: 'verae.billing.credit.applied',
+};
+
+export type BillingStatement = {
+  customerId: string;
+  prepaidCents: number;
+  credits: unknown[];
+  usage: unknown[];
+  payments: unknown[];
+};
+
+type Nc = {
+  request: (s: string, d: Uint8Array, o: { timeout: number }) => Promise<{ data: Uint8Array }>;
+  publish: (s: string, d: Uint8Array) => void;
+};
+
+let ncPromise: Promise<Nc | null> | null = null;
+
+async function nc(): Promise<Nc | null> {
+  const url = process.env.NATS_URL;
+  if (!url) return null;
+  if (!ncPromise) {
+    ncPromise = (async () => {
+      try {
+        const nats = await import('nats');
+        return (await nats.connect({ servers: url.split(','), name: 'zappier-edge' })) as unknown as Nc;
+      } catch {
+        return null;
+      }
+    })();
+  }
+  return ncPromise;
+}
+
+function encode(obj: unknown): Uint8Array {
+  return new TextEncoder().encode(JSON.stringify(obj));
+}
+function decode(buf: Uint8Array): unknown {
+  return JSON.parse(new TextDecoder().decode(buf) || '{}');
+}
+
+export async function natsStatement(customerId: string): Promise<BillingStatement | null> {
+  const c = await nc();
+  if (!c) return null;
+  const m = await c.request(BILLING_SUBJECTS.STATEMENT_GET, encode({ customerId }), { timeout: 2000 });
+  return decode(m.data) as BillingStatement;
+}
+
+export async function natsAdjust(payload: {
+  customerId: string;
+  cents: number;
+  reason: string;
+  agent: string;
+  kind?: string;
+}): Promise<unknown | null> {
+  const c = await nc();
+  if (!c) return null;
+  const m = await c.request(BILLING_SUBJECTS.BALANCE_ADJUST, encode(payload), { timeout: 2000 });
+  return decode(m.data);
+}
+
+export function natsPublish(subject: string, payload: unknown): void {
+  void nc().then((c) => {
+    if (c) c.publish(subject, encode(payload));
+  });
+}

packages/zappier/src/meter.ts

@@ -6,6 +6,7 @@ export function meter(
   endpointId: string,
   repo: UsageRepo,
   pricing: PricingContext,
+  onRecord?: (entry: { customerId: string; endpointId: string; cents: number }) => void,
 ): RequestHandler {
   return (req, res, next) => {
     const customer = req.customer;
@@ -34,14 +35,16 @@ export function meter(
       return;
     }
 
-    repo.record({
+    const entry = {
       customerId: customer.id,
       endpointId,
       cents: quote.totalCents,
       metadataBytes,
       attachmentBytes,
       timestamp: new Date(),
-    });
+    };
+    repo.record(entry);
+    onRecord?.({ customerId: customer.id, endpointId, cents: quote.totalCents });
     res.locals.quote = quote;
     next();
   };

packages/zappier/src/nats-shim.d.ts

@@ -0,0 +1,11 @@
+declare module 'nats' {
+  export function connect(opts: unknown): Promise<{
+    request(s: string, d: Uint8Array, o: { timeout: number }): Promise<{ data: Uint8Array }>;
+    publish(s: string, d: Uint8Array): void;
+    close(): Promise<void>;
+  }>;
+  export function StringCodec(): {
+    encode(s: string): Uint8Array;
+    decode(u: Uint8Array): string;
+  };
+}

packages/zappier/src/portal.ts

@@ -14,6 +14,9 @@ import { applyMonthlyCredit } from './billing/credit';
 import { InvoiceRepo } from './invoicing';
 import { RateCard, TierConfig } from './pricing';
 import { UsageRepo } from './usage';
+import { CreditLedger } from './credits';
+import { composeStatement } from './statement';
+import { BILLING_SUBJECTS, natsPublish, natsStatement, natsAdjust } from './billing-nats';
 
 /**
  * Customer portal API (/portal/api): signup, login with optional TOTP 2FA,
@@ -44,6 +47,7 @@ export interface PortalDeps {
   qr: (uri: string) => Promise<string>;
   sessionTtlMs?: number;
   now?: () => number;
+  credits?: CreditLedger;
 }
 
 const DEFAULT_SESSION_TTL_MS = 7 * 24 * 60 * 60 * 1000;
@@ -198,6 +202,24 @@ export function portalRouter(deps: PortalDeps): Router {
     res.json({ tiers: deps.tiers(), rateCard: deps.rateCard() });
   });
 
+  router.get('/statement', async (req, res) => {
+    const fromNats = await natsStatement(req.customer!.id);
+    if (fromNats) {
+      res.json({ ...fromNats, source: 'nats' });
+      return;
+    }
+    res.json({
+      ...composeStatement({
+        customerId: req.customer!.id,
+        prepaidCents: req.customer!.balanceCents ?? 0,
+        credits: (deps.credits || new CreditLedger()).list(req.customer!.id),
+        usage: deps.usage.listFor(req.customer!.id),
+        invoices: deps.invoices.list({ customerId: req.customer!.id }),
+      }),
+      source: 'local',
+    });
+  });
+
   router.get('/invoices/:id', (req, res) => {
     const invoice = deps.invoices.get(req.params.id);
     if (!invoice || invoice.customerId !== req.customer!.id) {
@@ -263,6 +285,20 @@ export function portalRouter(deps: PortalDeps): Router {
       balanceCents: (req.customer!.balanceCents ?? 0) + result.creditedCents,
     };
     if (result.creditedCents > 0) save(deps, customer);
+    if (result.creditedCents > 0) {
+      natsPublish(BILLING_SUBJECTS.PAYMENT_RECORDED, {
+        customerId: customer.id,
+        cents: result.creditedCents,
+        reason: 'reload',
+      });
+      void natsAdjust({
+        customerId: customer.id,
+        cents: result.creditedCents,
+        reason: 'reload',
+        agent: 'portal',
+        kind: 'payment',
+      });
+    }
     res.json({
       balanceCents: customer.balanceCents,
       mode: result.mode,

packages/zappier/src/statement.ts

@@ -0,0 +1,38 @@
+import { CreditAdjustment } from './credits';
+import { Invoice } from './invoicing';
+import { UsageEntry } from './usage';
+
+export function composeStatement(args: {
+  customerId: string;
+  prepaidCents: number;
+  credits: CreditAdjustment[];
+  usage: UsageEntry[];
+  invoices: Invoice[];
+}) {
+  const payments = args.invoices
+    .filter((i) => i.customerId === args.customerId && (i.status === 'paid' || i.status === 'issued'))
+    .map((i) => ({
+      id: i.id,
+      customerId: i.customerId,
+      cents: i.billableCents,
+      kind: i.status === 'paid' ? 'invoice-paid' : 'invoice-issued',
+      reason: i.period,
+      at: new Date(i.paidAtMs || i.issuedAtMs || Date.now()).toISOString(),
+    }));
+  return {
+    customerId: args.customerId,
+    prepaidCents: args.prepaidCents,
+    credits: args.credits.filter((c) => c.customerId === args.customerId),
+    usage: args.usage
+      .filter((u) => u.customerId === args.customerId)
+      .slice(-50)
+      .reverse()
+      .map((u) => ({
+        customerId: u.customerId,
+        endpointId: u.endpointId,
+        cents: u.cents,
+        at: u.timestamp.toISOString(),
+      })),
+    payments,
+  };
+}