Initial import of zappier-edge from zapier monorepo

src/billing-nats.ts

@@ -0,0 +1,139 @@
+/** Internal NATS billing bus. No-op when NATS_URL is unset (tests).
+ *  When NATS is on, every hop must pass verae.access.authz.check for a plane.
+ */
+
+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',
+  CUSTOMER_PUT: 'verae.billing.customer.put',
+};
+
+export const AUTHZ_CHECK = 'verae.access.authz.check';
+export type AccessPlane = 'zapier' | 'web' | 'api' | 'leaf' | 'staff';
+
+export type BillingStatement = {
+  customerId: string;
+  name?: string;
+  veraeUserId?: 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) || '{}');
+}
+
+async function authzAllow(
+  plane: AccessPlane,
+  subject: string,
+  extra: { principal?: string; kind?: string; veraeUserId?: string } = {},
+): Promise<boolean> {
+  const http = process.env.AUTHZ_URL;
+  if (!http && !(await nc())) return true;
+  try {
+    const c = await nc();
+    if (c) {
+      const m = await c.request(AUTHZ_CHECK, encode({ plane, subject, ...extra }), { timeout: 1500 });
+      const d = decode(m.data) as { allow?: boolean };
+      return Boolean(d.allow);
+    }
+    if (http) {
+      const r = await fetch(`${http.replace(/\/$/, '')}/check`, {
+        method: 'POST',
+        headers: { 'content-type': 'application/json' },
+        body: JSON.stringify({ plane, subject, ...extra }),
+      });
+      const d = (await r.json()) as { allow?: boolean };
+      return Boolean(d.allow);
+    }
+  } catch {
+    return false;
+  }
+  return true;
+}
+
+export async function natsStatement(
+  customerId: string,
+  plane: AccessPlane = 'web',
+  veraeUserId?: string,
+): Promise<BillingStatement | null> {
+  const c = await nc();
+  if (!c) return null;
+  if (
+    !(await authzAllow(plane, BILLING_SUBJECTS.STATEMENT_GET, { principal: customerId, veraeUserId }))
+  ) {
+    return null;
+  }
+  const m = await c.request(
+    BILLING_SUBJECTS.STATEMENT_GET,
+    encode({ customerId, plane, veraeUserId }),
+    { timeout: 2000 },
+  );
+  return decode(m.data) as BillingStatement;
+}
+
+export async function natsAdjust(
+  payload: {
+    customerId: string;
+    veraeUserId?: string;
+    cents: number;
+    reason: string;
+    agent: string;
+    kind?: string;
+  },
+  plane: AccessPlane = 'staff',
+): Promise<unknown | null> {
+  const c = await nc();
+  if (!c) return null;
+  if (
+    !(await authzAllow(plane, BILLING_SUBJECTS.BALANCE_ADJUST, {
+      principal: payload.agent,
+      kind: payload.kind,
+      veraeUserId: payload.veraeUserId,
+    }))
+  ) {
+    return null;
+  }
+  const m = await c.request(BILLING_SUBJECTS.BALANCE_ADJUST, encode({ ...payload, plane }), { timeout: 2000 });
+  return decode(m.data);
+}
+
+export function natsPublish(subject: string, payload: unknown, plane: AccessPlane = 'api'): void {
+  void (async () => {
+    const c = await nc();
+    if (!c) return;
+    if (!(await authzAllow(plane, subject))) return;
+    c.publish(subject, encode({ ...(payload as object), plane }));
+  })();
+}