Merge pull request #1397 from useautumn/charlie/webhook-immediate-response

chore: add early ack on webhooks

knip.json

@@ -11,6 +11,7 @@
 		"enumMembers",
 		"duplicates"
 	],
+	"ignore": ["ai/**"],
 	"ignoreWorkspaces": [
 		"packages/atmn",
 		"packages/autumn-js",

package.json

@@ -84,6 +84,7 @@
 		"setup": "node scripts/setup/setup.js",
 		"setup:s3-admin": "bun scripts/setup/setupS3Admin.ts",
 		"setup:test": "infisical run --env=dev --recursive -- bun scripts/setup/setup-test.ts",
+		"stripe:link-test": "ENV_FILE=.env infisical run --env=dev --recursive -- bun scripts/setup/link-test-stripe-account.ts",
 		"agent:bootstrap": "bash scripts/setup/agent-bootstrap.sh",
 		"dev:agent": "bash scripts/setup/devAgent.sh",
 		"migrate-functions": "infisical run --env=dev --recursive -- bun scripts/migrations/migrate-functions.ts",

scripts/setup/STRIPE_TEST_OAUTH.md

@@ -0,0 +1,47 @@
+# Stripe Test OAuth Linking
+
+Use this when local tests say the test org has no linked Stripe account, or when Stripe Connect webhooks are visible in Stripe but Autumn cannot map events back to `unit-test-org`.
+
+The Connect webhook destination should be:
+
+```txt
+https://c.autumn.ngrok.app/webhooks/connect/sandbox
+```
+
+OAuth still needs the Autumn org row to store the connected account ID:
+
+```json
+{ "test_stripe_connect": { "account_id": "acct_..." } }
+```
+
+## Commands
+
+List recent connected accounts for the test org email:
+
+```sh
+bun stripe:link-test -- --list --email=unit-test-org@test.com
+```
+
+Link an explicit account:
+
+```sh
+bun stripe:link-test -- --account-id=acct_...
+```
+
+Link the newest account matching the test org email:
+
+```sh
+bun stripe:link-test -- --latest --email=unit-test-org@test.com
+```
+
+If the org has a direct Stripe secret key, `createStripeCli` will prefer that over OAuth Connect. To force the OAuth account for sandbox tests:
+
+```sh
+bun stripe:link-test -- --account-id=acct_... --clear-secret-key
+```
+
+After linking, rerun a focused checkout test before the full suite:
+
+```sh
+ENV_FILE=.env infisical run --env=dev --recursive -- bun test --timeout 0 server/tests/integration/billing/attach/checkout/stripe-checkout/prepaid/stripe-checkout-prepaid-basic.test.ts
+```

scripts/setup/link-test-stripe-account.ts

@@ -0,0 +1,169 @@
+#!/usr/bin/env bun
+import { AppEnv, organizations } from "@autumn/shared";
+import { eq } from "drizzle-orm";
+import { initDrizzle } from "@server/db/initDrizzle.js";
+import { createStripeCli } from "@server/external/connect/createStripeCli.js";
+import { initMasterStripe } from "@server/external/connect/initStripeCli.js";
+import { OrgService } from "@server/internal/orgs/OrgService.js";
+import { clearOrgCache } from "@server/internal/orgs/orgUtils/clearOrgCache.js";
+import { loadLocalEnv } from "@server/utils/envUtils.js";
+
+loadLocalEnv();
+
+const args = process.argv.slice(2);
+
+const readFlag = (name: string) => {
+	const inline = args.find((arg) => arg.startsWith(`${name}=`));
+	if (inline) return inline.slice(name.length + 1);
+
+	const idx = args.indexOf(name);
+	return idx === -1 ? undefined : args[idx + 1];
+};
+
+const hasFlag = (name: string) => args.includes(name);
+
+const usage = () => {
+	console.log(`Usage:
+  bun stripe:link-test -- --account-id=acct_...
+  bun stripe:link-test -- --latest --email=unit-test-org@test.com
+  bun stripe:link-test -- --list --email=unit-test-org@test.com
+
+Options:
+  --org=<slug-or-id>       Autumn org to update. Defaults to TESTS_ORG.
+  --env=<sandbox|live>     Stripe environment. Defaults to sandbox.
+  --account-id=<acct_...>  Connected Stripe account ID to link.
+  --email=<email>          Filter Stripe connected accounts by email.
+  --latest                 Link the newest connected account matching --email.
+  --clear-secret-key       Clear the org's direct Stripe key for this env so Connect is used.
+  --list                   Print matching connected accounts without updating.
+`);
+};
+
+if (hasFlag("--help") || hasFlag("-h")) {
+	usage();
+	process.exit(0);
+}
+
+const env =
+	(readFlag("--env") || "sandbox").toLowerCase() === "live"
+		? AppEnv.Live
+		: AppEnv.Sandbox;
+const orgRef = readFlag("--org") || process.env.TESTS_ORG;
+const email = readFlag("--email");
+const accountIdArg = readFlag("--account-id");
+
+if (!orgRef) {
+	throw new Error("Missing org. Pass --org=<slug-or-id> or set TESTS_ORG.");
+}
+
+const { db, client } = initDrizzle();
+
+const getOrg = async () => {
+	const bySlug = await OrgService.getBySlug({ db, slug: orgRef });
+	if (bySlug) return bySlug;
+
+	return await OrgService.get({ db, orgId: orgRef });
+};
+
+const listAccounts = async () => {
+	const stripe = initMasterStripe({ env, skipInstrumentation: true });
+	const accounts = await stripe.accounts.list({ limit: 100 });
+
+	return accounts.data
+		.filter((account) => !email || account.email === email)
+		.sort((a, b) => b.created - a.created);
+};
+
+try {
+	const org = await getOrg();
+	const accounts = await listAccounts();
+
+	if (hasFlag("--list")) {
+		console.log(
+			JSON.stringify(
+				accounts.map((account) => ({
+					id: account.id,
+					email: account.email,
+					created: new Date(account.created * 1000).toISOString(),
+					charges_enabled: account.charges_enabled,
+					details_submitted: account.details_submitted,
+				})),
+				null,
+				2,
+			),
+		);
+		process.exit(0);
+	}
+
+	const accountId =
+		accountIdArg || (hasFlag("--latest") ? accounts[0]?.id : undefined);
+
+	if (!accountId) {
+		throw new Error(
+			"Missing account. Pass --account-id=acct_... or use --latest with --email=...",
+		);
+	}
+
+	const directKeyField =
+		env === AppEnv.Sandbox ? "test_api_key" : "live_api_key";
+	const directWebhookSecretField =
+		env === AppEnv.Sandbox ? "test_webhook_secret" : "live_webhook_secret";
+	const hasDirectKey = Boolean(org.stripe_config?.[directKeyField]);
+
+	if (hasDirectKey && !hasFlag("--clear-secret-key")) {
+		throw new Error(
+			`${org.slug} has stripe_config.${directKeyField}; createStripeCli will prefer that over Connect. Re-run with --clear-secret-key to use the OAuth account.`,
+		);
+	}
+
+	const stripe = initMasterStripe({ env, accountId, skipInstrumentation: true });
+	await stripe.accounts.retrieve();
+
+	await OrgService.updateStripeConnect({
+		db,
+		orgId: org.id,
+		accountId,
+		env,
+	});
+
+	if (hasDirectKey) {
+		await db
+			.update(organizations)
+			.set({
+				stripe_config: {
+					...(org.stripe_config || {}),
+					[directKeyField]: null,
+					[directWebhookSecretField]: null,
+				},
+			})
+			.where(eq(organizations.id, org.id));
+		await clearOrgCache({ db, orgId: org.id });
+	}
+
+	const updatedOrg = await OrgService.get({ db, orgId: org.id });
+	const resolvedStripe = createStripeCli({
+		org: updatedOrg,
+		env,
+		skipInstrumentation: true,
+	});
+	const resolvedAccount = await resolvedStripe.accounts.retrieve();
+
+	console.log(
+		JSON.stringify(
+			{
+				org: { id: updatedOrg.id, slug: updatedOrg.slug },
+				env,
+				linked_account_id: accountId,
+				resolved_account_id: resolvedAccount.id,
+				test_stripe_connect: updatedOrg.test_stripe_connect,
+				live_stripe_connect: updatedOrg.live_stripe_connect,
+			},
+			null,
+			2,
+		),
+	);
+} finally {
+	await client.end();
+}
+
+process.exit(0);

server/src/external/stripe/stripeWebhookRouter.ts

@@ -8,6 +8,7 @@ import { stripeLegacySeederMiddleware } from "./webhookMiddlewares/stripeLegacyS
 import { stripeSyncMiddleware } from "./webhookMiddlewares/stripeSyncMiddleware.js";
 import { stripeToAutumnCustomerMiddleware } from "./webhookMiddlewares/stripeToAutumnCustomerMiddleware.js";
 import type { StripeWebhookHonoEnv } from "./webhookMiddlewares/stripeWebhookContext.js";
+import { stripeWebhookEarlyAckMiddleware } from "./webhookMiddlewares/stripeWebhookEarlyAckMiddleware.js";
 import { stripeWebhookRefreshMiddleware } from "./webhookMiddlewares/stripeWebhookRefreshMiddleware.js";
 
 export const stripeWebhookRouter = new Hono<StripeWebhookHonoEnv>();
@@ -16,12 +17,13 @@ export const stripeWebhookRouter = new Hono<StripeWebhookHonoEnv>();
 stripeWebhookRouter.post(
 	"/webhooks/stripe/:orgId/:env",
 	stripeLegacySeederMiddleware,
+	stripeIdempotencyMiddleware,
+	stripeWebhookEarlyAckMiddleware,
 	stripeWebhookRefreshMiddleware,
 	stripeSyncMiddleware,
 	stripeToAutumnCustomerMiddleware,
 	stripeLoggerMiddleware,
 	traceEnrichMiddleware,
-	stripeIdempotencyMiddleware,
 	handleStripeWebhookEvent,
 );
 
@@ -29,11 +31,12 @@ stripeWebhookRouter.post(
 stripeWebhookRouter.post(
 	"/webhooks/connect/:env",
 	stripeConnectSeederMiddleware,
+	stripeIdempotencyMiddleware,
+	stripeWebhookEarlyAckMiddleware,
 	stripeWebhookRefreshMiddleware,
 	stripeSyncMiddleware,
 	stripeToAutumnCustomerMiddleware,
 	stripeLoggerMiddleware,
 	traceEnrichMiddleware,
-	stripeIdempotencyMiddleware,
 	handleStripeWebhookEvent,
 );

server/src/external/stripe/webhookMiddlewares/stripeWebhookEarlyAckMiddleware.ts

@@ -0,0 +1,38 @@
+import type { Context, Next } from "hono";
+import type { StripeWebhookHonoEnv } from "./stripeWebhookContext";
+
+const getWaitUntil = (c: Context<StripeWebhookHonoEnv>) => {
+	try {
+		return c.executionCtx.waitUntil.bind(c.executionCtx);
+	} catch {
+		return undefined;
+	}
+};
+
+export const stripeWebhookEarlyAckMiddleware = async (
+	c: Context<StripeWebhookHonoEnv>,
+	next: Next,
+) => {
+	const ctx = c.get("ctx");
+	const runWebhook = () =>
+		Promise.resolve()
+			.then(next)
+			.catch((error) => {
+				ctx.logger.error(`Stripe webhook background processing failed: ${error}`, {
+					error,
+				});
+			});
+
+	const waitUntil = getWaitUntil(c);
+	if (waitUntil) {
+		try {
+			waitUntil(runWebhook());
+		} catch (error) {
+			ctx.logger.error(`Stripe webhook waitUntil failed: ${error}`, { error });
+		}
+	} else {
+		setImmediate(() => void runWebhook());
+	}
+
+	return c.json({ received: true }, 200);
+};

server/src/internal/billing/v2/providers/stripe/setup/fetchStripeSubscriptionForBilling.ts

@@ -56,10 +56,7 @@ export const fetchStripeSubscriptionForBilling = async ({
 	if (!subId) return undefined;
 
 	const sub = await stripeCli.subscriptions.retrieve(subId, {
-		expand: [
-			"discounts.source.coupon.applies_to",
-			"latest_invoice.lines.data.discount_amounts",
-		],
+		expand: ["discounts.source.coupon.applies_to"],
 	});
 
 	if (!sub) {

server/tests/unit/webhooks/stripe-webhook-early-ack.test.ts

@@ -0,0 +1,125 @@
+import { describe, expect, test } from "bun:test";
+import { Hono } from "hono";
+import { stripeWebhookEarlyAckMiddleware } from "@/external/stripe/webhookMiddlewares/stripeWebhookEarlyAckMiddleware";
+
+const waitForImmediate = () => new Promise((resolve) => setImmediate(resolve));
+
+const createApp = () => {
+	const app = new Hono();
+
+	app.use("*", async (c, next) => {
+		(c as any).set("ctx", {
+			logger: {
+				error: () => {},
+			},
+		});
+		await next();
+	});
+
+	return app;
+};
+
+describe("stripeWebhookEarlyAckMiddleware", () => {
+	test("uses executionCtx.waitUntil when the runtime provides it", async () => {
+		const waits: Promise<unknown>[] = [];
+		let processed = false;
+		const response = await stripeWebhookEarlyAckMiddleware(
+			{
+				get: () => ({
+					logger: { error: () => {} },
+				}),
+				json: (body: unknown, status: number) =>
+					new Response(JSON.stringify(body), { status }),
+				executionCtx: {
+					waitUntil: (promise: Promise<unknown>) => waits.push(promise),
+				},
+			} as never,
+			async () => {
+				processed = true;
+			},
+		);
+
+		expect(response.status).toBe(200);
+		expect(await response.json()).toEqual({ received: true });
+		expect(waits).toHaveLength(1);
+
+		await waits[0];
+		expect(processed).toBe(true);
+	});
+
+	test("does not run downstream twice when waitUntil throws", async () => {
+		const errors: unknown[] = [];
+		let runs = 0;
+		const response = await stripeWebhookEarlyAckMiddleware(
+			{
+				get: () => ({
+					logger: { error: (_message: string, meta: unknown) => errors.push(meta) },
+				}),
+				json: (body: unknown, status: number) =>
+					new Response(JSON.stringify(body), { status }),
+				executionCtx: {
+					waitUntil: () => {
+						throw new Error("waitUntil failed");
+					},
+				},
+			} as never,
+			async () => {
+				runs++;
+			},
+		);
+
+		expect(response.status).toBe(200);
+		await Promise.resolve();
+		await waitForImmediate();
+		expect(runs).toBe(1);
+		expect(errors).toHaveLength(1);
+	});
+
+	test("returns 200 before downstream webhook processing completes", async () => {
+		const app = createApp();
+		let resolveProcessing!: () => void;
+		let processed = false;
+		const processing = new Promise<void>((resolve) => {
+			resolveProcessing = resolve;
+		});
+
+		app.post(
+			"/webhook",
+			stripeWebhookEarlyAckMiddleware as never,
+			async (c) => {
+				await processing;
+				processed = true;
+				return c.json({ processed: true }, 200);
+			},
+		);
+
+		const response = await app.request("/webhook", { method: "POST" });
+
+		expect(response.status).toBe(200);
+		expect(await response.json()).toEqual({ received: true });
+		expect(processed).toBe(false);
+
+		resolveProcessing();
+		await waitForImmediate();
+		expect(processed).toBe(true);
+	});
+
+	test("does not run downstream webhook processing before returning", async () => {
+		const app = createApp();
+		let started = false;
+
+		app.post("/webhook", stripeWebhookEarlyAckMiddleware as never, (c) => {
+			started = true;
+			return c.json({ processed: true }, 200);
+		});
+
+		const response = await app.request("/webhook", { method: "POST" });
+
+		expect(response.status).toBe(200);
+		expect(await response.json()).toEqual({ received: true });
+		expect(started).toBe(false);
+
+		await waitForImmediate();
+		expect(started).toBe(true);
+	});
+});