zappier-edge/src/billing-nats.ts

139 lines
3.8 KiB
TypeScript

/** 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 }));
})();
}