diff --git a/apps/cloud/src/account/org-api-key-revoke.node.test.ts b/apps/cloud/src/account/org-api-key-revoke.node.test.ts index 97a3816ba..0de1c64d2 100644 --- a/apps/cloud/src/account/org-api-key-revoke.node.test.ts +++ b/apps/cloud/src/account/org-api-key-revoke.node.test.ts @@ -108,6 +108,7 @@ const stubUsers = Layer.succeed(UserStoreService)({ const stubAutumn = Layer.succeed(AutumnService)({ use: () => Effect.die("revoke does not touch billing"), + ensureCustomer: () => Effect.die("revoke does not touch billing"), checkExecutionBalance: () => Effect.die("revoke does not touch billing"), trackExecution: () => Effect.void, }); diff --git a/apps/cloud/src/auth/handlers.ts b/apps/cloud/src/auth/handlers.ts index 03b44f5a0..05bc5ec17 100644 --- a/apps/cloud/src/auth/handlers.ts +++ b/apps/cloud/src/auth/handlers.ts @@ -22,6 +22,7 @@ import { env } from "cloudflare:workers"; import { WorkOSError } from "./errors"; import { WorkOSClient } from "./workos"; import { AutumnService } from "../extensions/billing/service"; +import { captureCauseEffect } from "../observability"; import { hasPaidOrganizationSubscription, isOverFreeOrganizationLimit, @@ -415,7 +416,9 @@ export const CloudSessionAuthHandlers = HttpApiBuilder.group( ), { concurrency: 3 }, ).pipe( - Effect.catchTag("AutumnError", () => Effect.fail(new WorkOSError())), + // Any Autumn failure here (outage or missing customer) leaves the + // paid/free split unknown, and the limit must fail closed. + Effect.mapError(() => new WorkOSError()), Effect.map((ids) => new Set(ids.filter(Predicate.isNotNull))), ); @@ -431,6 +434,24 @@ export const CloudSessionAuthHandlers = HttpApiBuilder.group( s.upsertOrganization({ id: org.id, name: org.name }), ); + // Provision the org's billing customer while we're the ones creating + // the org. Without this the first billing call an org ever makes is a + // non-creating one (balance check / usage track), which 404s and keeps + // 404ing — unlimited unbilled executions. Non-fatal: a billing blip + // must not block signup, and the billing seam heals a customer that + // is still missing later. + yield* autumn.ensureCustomer(org.id).pipe( + Effect.catch((error) => + Effect.gen(function* () { + yield* Effect.logWarning( + "createOrganization: could not provision the Autumn customer", + { organizationId: org.id, error }, + ); + yield* captureCauseEffect(error); + }), + ), + ); + // Try to attach the new org to the current session. This can fail // (or silently return a session still scoped to the old org) when // the caller's current session is stale — most commonly after the @@ -515,7 +536,9 @@ export const CloudSessionAuthHandlers = HttpApiBuilder.group( yield* autumn .use((client) => client.customers.delete({ customerId: organizationId })) .pipe( - Effect.catchTag("AutumnError", (error) => + // Includes the "customer never existed" answer: nothing to cancel + // is a fine outcome for a deleted org, and it is still worth a line. + Effect.catch((error) => Effect.logWarning("deleteOrganization: failed to delete Autumn customer", { organizationId, error, diff --git a/apps/cloud/src/extensions/billing/service.ts b/apps/cloud/src/extensions/billing/service.ts index 1b33b203c..e96c8ed84 100644 --- a/apps/cloud/src/extensions/billing/service.ts +++ b/apps/cloud/src/extensions/billing/service.ts @@ -17,15 +17,61 @@ export class AutumnError extends Data.TaggedError("AutumnError")<{ cause?: unknown; }> {} +/** + * Autumn has no customer record for the organization. Split out from + * `AutumnError` because it is not an outage and must not be treated like one: + * an outage is transient and the right answer is to fail open and page, while + * a missing customer is a PERMANENT provisioning gap — every subsequent + * balance call 404s the same way, so the org runs unbilled and unmetered + * forever. Callers that own an organization id repair it (see + * `withProvisionedCustomer`); everything else still surfaces as `AutumnError`. + */ +export class AutumnCustomerNotFoundError extends Data.TaggedError("AutumnCustomerNotFoundError")<{ + message: string; + cause?: unknown; +}> {} + +export type AutumnFailure = AutumnError | AutumnCustomerNotFoundError; + +// Autumn's own error code for "no such customer", carried in the JSON body of +// a 404 (`{"message":"Customer not found","code":"customer_not_found"}`). +const CUSTOMER_NOT_FOUND_CODE = "customer_not_found"; + +/** + * True when `cause` is Autumn's "no such customer" answer. The autumn-js SDK + * throws an `AutumnError` carrying the raw HTTP `statusCode` and `body`; match + * on the code rather than the status alone so an unrelated 404 (a removed + * endpoint, a proxy) is still reported as a genuine failure. + */ +const isCustomerNotFoundCause = (cause: unknown): boolean => { + if (typeof cause !== "object" || cause === null) return false; + const { statusCode, body } = cause as { readonly statusCode?: unknown; readonly body?: unknown }; + if (statusCode !== 404 || typeof body !== "string") return false; + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: classifying a third-party SDK's raw response body; a body that isn't JSON simply isn't this error + try { + // oxlint-disable-next-line executor/no-json-parse -- boundary: the autumn-js SDK hands back the response body as an unvalidated string + const parsed: unknown = JSON.parse(body); + if (typeof parsed !== "object" || parsed === null) return false; + return (parsed as { readonly code?: unknown }).code === CUSTOMER_NOT_FOUND_CODE; + } catch { + return false; + } +}; + // --------------------------------------------------------------------------- // Service interface // --------------------------------------------------------------------------- export type IAutumnService = Readonly<{ - use: (fn: (client: Autumn) => Promise) => Effect.Effect; + use: (fn: (client: Autumn) => Promise) => Effect.Effect; + /** + * Provision the organization's Autumn customer, creating it if Autumn has + * never seen it. Idempotent — safe to call on every org creation. + */ + ensureCustomer: (organizationId: string) => Effect.Effect; checkExecutionBalance: ( organizationId: string, - ) => Effect.Effect<{ readonly allowed: boolean }, AutumnError, never>; + ) => Effect.Effect<{ readonly allowed: boolean }, AutumnFailure, never>; /** * Fire-and-forget-safe execution usage tracker. Errors are caught and * logged; the returned Effect never fails. Callers typically @@ -48,6 +94,7 @@ const make = Effect.sync(() => { ); return { use: () => notConfigured, + ensureCustomer: () => notConfigured, checkExecutionBalance: () => notConfigured, trackExecution: () => Effect.void, } satisfies IAutumnService; @@ -62,16 +109,53 @@ const make = Effect.sync(() => { const use = (fn: (client: Autumn) => Promise) => Effect.tryPromise({ try: () => fn(client), - catch: (cause) => new AutumnError({ message: "Autumn SDK request failed", cause }), - }).pipe(Effect.withSpan(`autumn.${fn.name ?? "use"}`)); + catch: (cause): AutumnFailure => + isCustomerNotFoundCause(cause) + ? new AutumnCustomerNotFoundError({ + message: "Autumn has no customer for this organization", + cause, + }) + : new AutumnError({ message: "Autumn SDK request failed", cause }), + // An inline arrow's `name` is "" — not nullish — so `??` left every + // Autumn call tracing as the bare span `autumn.`. + }).pipe(Effect.withSpan(`autumn.${fn.name || "use"}`)); + + const ensureCustomer = (organizationId: string) => + Effect.asVoid(use((c) => c.customers.getOrCreate({ customerId: organizationId }))); + + /** + * Run `operation`; if Autumn answers "no such customer", provision the + * organization's customer and run it ONCE more. + * + * This is the seam that closes the provisioning hole. Both billing paths use + * non-creating endpoints, so an organization Autumn never learned about + * 404s here forever: the balance gate fails open (correct for an outage, + * catastrophic as a steady state) and every usage track is lost. Repairing + * the customer makes the retry land the call — and a genuine Autumn outage + * still fails with `AutumnError` and still pages, unretried. + */ + const withProvisionedCustomer = ( + organizationId: string, + operation: Effect.Effect, + ) => + operation.pipe( + Effect.catchTag("AutumnCustomerNotFoundError", () => + Effect.gen(function* () { + yield* Effect.annotateCurrentSpan({ "autumn.customer.provisioned": true }); + yield* ensureCustomer(organizationId); + return yield* operation; + }), + ), + ); const trackExecution = (organizationId: string) => Effect.gen(function* () { yield* Effect.annotateCurrentSpan({ "autumn.customer.id": organizationId }); - yield* use((c) => - c.track({ customerId: organizationId, featureId: "executions", value: 1 }), + yield* withProvisionedCustomer( + organizationId, + use((c) => c.track({ customerId: organizationId, featureId: "executions", value: 1 })), ).pipe( - Effect.catchTag("AutumnError", (error) => + Effect.catch((error) => Effect.gen(function* () { // Silent billing data loss is worth paging on: autumn.trackExecution // is fire-and-forget so the caller doesn't handle it themselves. @@ -88,13 +172,14 @@ const make = Effect.sync(() => { const checkExecutionBalance = (organizationId: string) => Effect.gen(function* () { yield* Effect.annotateCurrentSpan({ "autumn.customer.id": organizationId }); - const check = yield* use((c) => - c.check({ customerId: organizationId, featureId: "executions" }), + const check = yield* withProvisionedCustomer( + organizationId, + use((c) => c.check({ customerId: organizationId, featureId: "executions" })), ); return { allowed: check.allowed }; }).pipe(Effect.withSpan("autumn.checkExecutionBalance")); - return { use, checkExecutionBalance, trackExecution } satisfies IAutumnService; + return { use, ensureCustomer, checkExecutionBalance, trackExecution } satisfies IAutumnService; }); export class AutumnService extends Context.Service()( diff --git a/e2e/cloud/billing-customer-provisioning.test.ts b/e2e/cloud/billing-customer-provisioning.test.ts new file mode 100644 index 000000000..ab98cfd27 --- /dev/null +++ b/e2e/cloud/billing-customer-provisioning.test.ts @@ -0,0 +1,281 @@ +// Cloud-only (billing): an organization must EXIST as a customer at the billing +// provider, or every billing call for it answers `customer_not_found` forever. +// +// That state is invisible from the product side and catastrophic underneath it: +// the balance gate fails open (by design — a billing outage must not stop +// executions), and usage tracking is fire-and-forget, so the org runs unlimited +// executions and NONE of them reach the meter. The product looks perfectly +// healthy while every execution is unbilled and unmetered. +// +// Three guarantees are pinned here, all read from Autumn's own state rather than +// from the execution's response (which can never show any of them): +// +// 1. creating an organization provisions its billing customer up front, +// 2. if a customer is missing anyway, the billing seam heals it in place and +// the execution still lands on the meter, and +// 3. any OTHER billing failure is left alone — the repair is scoped to the +// "no such customer" answer, so a real provider failure is still reported +// rather than quietly retried away. +// +// The missing customer is produced with the emulator's fault injector — one +// 404 `customer_not_found` per billing endpoint — because the emulator, like +// real Autumn's SDK flow, otherwise auto-creates customers on contact. +import { expect } from "@effect/vitest"; +import { Effect, Schedule } from "effect"; + +import { scenario } from "../src/scenario"; +import { Autumn, Billing, Mcp, Target } from "../src/services"; +import type { Identity } from "../src/target"; + +const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label; + +/** The org the bearer is scoped to — the Autumn customer id every billing call + * is made against — read from the JWT's public claims. */ +const orgIdOf = (bearer: string): string => { + const claims = JSON.parse(Buffer.from(bearer.split(".")[1] ?? "", "base64url").toString()) as { + readonly org_id?: string; + }; + if (!claims.org_id) throw new Error("orgIdOf: bearer carries no org_id claim"); + return claims.org_id; +}; + +/** A 404 `customer_not_found`, byte-for-byte the shape Autumn answers with for + * an organization it has no customer record for. */ +const CUSTOMER_NOT_FOUND = { + status: 404, + body: { message: "Customer not found", code: "customer_not_found" }, +} as const; + +/** A 404 that is NOT a missing customer — a moved route, a proxy, a gateway. + * Autumn carries a different code, and that code is the entire safety margin + * between "provision and retry" and "quietly retry away a real failure". */ +const UNRELATED_NOT_FOUND = { + status: 404, + body: { message: "Not Found", code: "not_found" }, +} as const; + +scenario( + "Billing · creating an organization provisions it as a billing customer", + { timeout: 180_000 }, + Effect.gen(function* () { + yield* Billing; + const autumn = yield* Autumn; + const target = yield* Target; + const mcp = yield* Mcp; + + // A brand-new user creating their first organization: none of the + // opportunistic billing lookups (the over-the-free-limit check, the members + // page's seat count, the rate limiter's paid-plan exemption) are reached on + // this journey, so org creation is the only chance to provision. + const identity = yield* target.newIdentity(); + const bearer = yield* mcp.mintBearer(emailOf(identity)); + const organizationId = orgIdOf(bearer); + + const customerIds = yield* autumn.customerIds(); + expect( + customerIds, + "the new organization exists as a customer at the billing provider", + ).toContain(organizationId); + }), +); + +scenario( + "Billing · a missing billing customer is healed in place and the execution is still metered", + { timeout: 180_000 }, + Effect.gen(function* () { + yield* Billing; + const autumn = yield* Autumn; + const target = yield* Target; + const mcp = yield* Mcp; + + const identity = yield* target.newIdentity(); + const bearer = yield* mcp.mintBearer(emailOf(identity)); + const customerId = orgIdOf(bearer); + + const before = yield* autumn.usageEvents({ customerId, featureId: "executions" }); + expect(before.length, "a brand-new org starts with zero metered executions").toBe(0); + + yield* Effect.gen(function* () { + // One 404 each: the next balance check and the next usage track both + // answer "this customer does not exist". A single `times: 1` fault means + // a seam that heals and retries gets through, while one that gives up + // loses the usage permanently — exactly the production failure. + yield* autumn.armFault({ + match: { operationId: "balances.check" }, + response: CUSTOMER_NOT_FOUND, + times: 1, + }); + yield* autumn.armFault({ + match: { operationId: "balances.track" }, + response: CUSTOMER_NOT_FOUND, + times: 1, + }); + + const session = mcp.session(identity); + const result = yield* session.call("execute", { code: "return 6 * 7;" }); + + // The gate still fails open: a billing problem never blocks a customer. + expect(result.ok, "the execution runs despite the missing billing customer").toBe(true); + expect(result.text, "it returns its value").toContain("42"); + + // Both billing calls really did reach Autumn and really were rejected — + // without this the scenario could pass on a fault that never armed. The + // ledger is shared by the whole run, so attribute to THIS org: another + // scenario's faulted call must not stand in for this one's. + const checks = (yield* autumn.ledgerFor("balances.check")).filter( + (entry) => entry.customerId === customerId, + ); + expect( + checks.some((entry) => entry.faulted), + "the balance check reached Autumn and was answered customer_not_found", + ).toBe(true); + const tracks = (yield* autumn.ledgerFor("balances.track")).filter( + (entry) => entry.customerId === customerId, + ); + expect( + tracks.some((entry) => entry.faulted), + "the usage track reached Autumn and was answered customer_not_found", + ).toBe(true); + + // The guarantee: the seam provisions the customer and retries, so the + // execution lands on the meter. Before the fix this ledger stays empty + // forever — the execution ran, and nobody was ever billed for it. + const metered = yield* autumn.expectUsage({ + customerId, + featureId: "executions", + count: 1, + }); + expect(metered.length, "the execution is metered exactly once").toBe(1); + expect(metered[0]?.value, "it meters a single unit").toBe(1); + + // Healing the customer is what made the retry possible, so the customer + // must exist at the provider afterwards. + const customerIds = yield* autumn.customerIds(); + expect(customerIds, "the organization has a billing customer afterwards").toContain( + customerId, + ); + }).pipe(Effect.ensuring(autumn.clearFaults().pipe(Effect.ignore))); + }), +); + +scenario( + "Billing · an unrelated billing failure is reported, not retried away", + { timeout: 180_000 }, + Effect.gen(function* () { + yield* Billing; + const autumn = yield* Autumn; + const target = yield* Target; + const mcp = yield* Mcp; + + const identity = yield* target.newIdentity(); + const bearer = yield* mcp.mintBearer(emailOf(identity)); + const customerId = orgIdOf(bearer); + + yield* Effect.gen(function* () { + // A 404 that is not "no such customer". If the seam treated every 404 as + // a provisioning gap it would create a customer and retry, turning a real + // provider failure into a silent success — and the alert never fires. + yield* autumn.armFault({ + match: { operationId: "balances.track" }, + response: UNRELATED_NOT_FOUND, + times: 1, + }); + + const session = mcp.session(identity); + const faulted = yield* session.call("execute", { code: "return 1 + 1;" }); + expect(faulted.ok, "the execution runs — billing never blocks a customer").toBe(true); + + // A second, unfaulted execution. Its usage landing is the barrier: the + // first execution's track (and any retry of it) is already done by the + // time this one is on the meter, so the counts below are settled without + // waiting on a clock. + const clean = yield* session.call("execute", { code: "return 2 + 2;" }); + expect(clean.ok, "the follow-up execution runs too").toBe(true); + yield* autumn.expectUsage({ customerId, featureId: "executions", count: 1 }); + + const tracks = (yield* autumn.ledgerFor("balances.track")).filter( + (entry) => entry.customerId === customerId, + ); + expect( + tracks.filter((entry) => entry.faulted).length, + "the first usage track was answered with the unrelated 404", + ).toBe(1); + // Two executions, two attempts: the failed one was reported and dropped, + // not repaired and replayed. + expect(tracks.length, "the rejected usage track is not retried").toBe(2); + + const metered = yield* autumn.usageEvents({ customerId, featureId: "executions" }); + expect(metered.length, "only the unfaulted execution reaches the meter").toBe(1); + }).pipe(Effect.ensuring(autumn.clearFaults().pipe(Effect.ignore))); + }), +); + +scenario( + "Billing · a customer that cannot be provisioned is given up on, not retried in a loop", + { timeout: 180_000 }, + Effect.gen(function* () { + yield* Billing; + const autumn = yield* Autumn; + const target = yield* Target; + const mcp = yield* Mcp; + + const identity = yield* target.newIdentity(); + const bearer = yield* mcp.mintBearer(emailOf(identity)); + const customerId = orgIdOf(bearer); + + // Attempts against THIS org's meter, polled until the seam has settled on + // `atLeast` of them — the ledger is the only place the retry is visible. + const trackAttempts = (atLeast: number) => + autumn.ledgerFor("balances.track").pipe( + Effect.map((entries) => entries.filter((entry) => entry.customerId === customerId)), + Effect.filterOrFail( + (entries) => entries.length >= atLeast, + (entries) => `only ${entries.length}/${atLeast} usage-track attempts for ${customerId}`, + ), + Effect.retry(Schedule.both(Schedule.spaced("250 millis"), Schedule.recurs(40))), + ); + + yield* Effect.gen(function* () { + // "No such customer" that does NOT go away when the customer is created — + // a provisioning gap the repair genuinely cannot close (a rejected id, a + // provider that never materializes the record). The seam must repair, + // retry ONCE, report, and stop: a repair loop here runs in a forked, + // untimed fibre, so an unbounded one hammers the provider forever for a + // single execution. + yield* autumn.armFault({ + match: { operationId: "balances.track" }, + response: CUSTOMER_NOT_FOUND, + times: 20, + }); + + const session = mcp.session(identity); + const stuck = yield* session.call("execute", { code: "return 7 * 6;" }); + expect(stuck.ok, "the execution runs — billing never blocks a customer").toBe(true); + expect(stuck.text, "it returns its value").toContain("42"); + + // The original attempt plus the one post-repair retry. + yield* trackAttempts(2); + + // Clearing the fault re-opens the meter, and the next execution's usage + // landing is the barrier: whatever the first execution was going to do to + // the ledger is finished by then, so the count below is settled without + // waiting on a clock. + yield* autumn.clearFaults(); + const clean = yield* session.call("execute", { code: "return 2 + 2;" }); + expect(clean.ok, "the follow-up execution runs too").toBe(true); + yield* autumn.expectUsage({ customerId, featureId: "executions", count: 1 }); + + const tracks = (yield* autumn.ledgerFor("balances.track")).filter( + (entry) => entry.customerId === customerId, + ); + expect( + tracks.filter((entry) => entry.faulted).length, + "the unrepairable customer is attempted exactly twice: once, then once after the repair", + ).toBe(2); + expect(tracks.length, "and no further attempt beyond the next execution's own").toBe(3); + + const metered = yield* autumn.usageEvents({ customerId, featureId: "executions" }); + expect(metered.length, "only the execution Autumn accepted reaches the meter").toBe(1); + }).pipe(Effect.ensuring(autumn.clearFaults().pipe(Effect.ignore))); + }), +); diff --git a/e2e/src/surfaces/autumn.ts b/e2e/src/surfaces/autumn.ts index a401cf548..2b3c32b27 100644 --- a/e2e/src/surfaces/autumn.ts +++ b/e2e/src/surfaces/autumn.ts @@ -51,11 +51,19 @@ export interface LedgerEntry { readonly method: string; readonly path: string; readonly faulted: boolean; + /** The customer the request was made against, read from its body. Autumn's + * ledger is shared by every scenario in the run, so attempts have to be + * attributed to one organization before they can be counted. */ + readonly customerId?: string; } export interface AutumnSurface { /** One-shot read of the matching usage events from the ledger. */ readonly usageEvents: (query: UsageQuery) => Effect.Effect; + /** Every customer id Autumn currently holds (`customers.list`). An org that + * was never provisioned as a billing customer is simply absent — the state + * in which every balance call 404s `customer_not_found` forever. */ + readonly customerIds: () => Effect.Effect; /** Poll until at least `count` matching events have landed. The track is * forked and the worker drains it on `waitUntil` shortly after the execution * returns, so arrival is eventually-consistent — polling IS the contract: @@ -129,6 +137,26 @@ export const makeAutumnSurface = (autumnUrl: string): AutumnSurface => { })); }); + const customerIds = () => + Effect.gen(function* () { + const response = yield* Effect.promise(() => + fetch(`${autumnUrl}/v1/customers.list`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: "{}", + }), + ); + if (!response.ok) { + return yield* Effect.fail( + `autumn customers.list responded ${response.status}: ${yield* Effect.promise(() => response.text())}`, + ); + } + const body = (yield* Effect.promise(() => response.json())) as { + readonly list?: ReadonlyArray<{ readonly id?: string }>; + }; + return (body.list ?? []).map((customer) => customer.id ?? ""); + }); + const settleCheckout = (sessionId: string) => Effect.gen(function* () { const response = yield* Effect.promise(() => @@ -232,20 +260,30 @@ export const makeAutumnSurface = (autumnUrl: string): AutumnSurface => { readonly method?: string; readonly path?: string; readonly faulted?: boolean; + readonly request?: { readonly body?: unknown }; }>; }; return (body.entries ?? []) .filter((entry) => entry.operationId === operationId) - .map((entry) => ({ - operationId: entry.operationId, - method: entry.method ?? "", - path: entry.path ?? "", - faulted: entry.faulted === true, - })); + .map((entry) => { + const requestBody = entry.request?.body; + const customerId = + typeof requestBody === "object" && requestBody !== null + ? (requestBody as { readonly customer_id?: unknown }).customer_id + : undefined; + return { + operationId: entry.operationId, + method: entry.method ?? "", + path: entry.path ?? "", + faulted: entry.faulted === true, + customerId: typeof customerId === "string" ? customerId : undefined, + }; + }); }); return { usageEvents, + customerIds, settleCheckout, exhaustExecutions, attachPlan,