diff --git a/server/src/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/handleCheckoutSessionMetadataV2.ts b/server/src/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/handleCheckoutSessionMetadataV2.ts index b72217626..1124d1669 100644 --- a/server/src/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/handleCheckoutSessionMetadataV2.ts +++ b/server/src/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/handleCheckoutSessionMetadataV2.ts @@ -7,9 +7,9 @@ import { createStripeScheduleFromCheckout } from "@/external/stripe/webhookHandl import { modifyStripeSubscriptionFromCheckout } from "@/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/modifyStripeSubscriptionFromCheckout"; import { syncSubscriptionItemMetadataFromCheckout } from "@/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/syncSubscriptionItemMetadataFromCheckout"; import { updateBillingPlanFromCheckout } from "@/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/updateBillingPlanFromCheckout"; +import { withClaimedCheckoutSessionMetadata } from "@/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/withClaimedCheckoutSessionMetadata"; import type { StripeWebhookContext } from "@/external/stripe/webhookMiddlewares/stripeWebhookContext"; import { persistDeferredCreateSchedule } from "@/internal/billing/v2/actions/createSchedule/utils/persistDeferredCreateSchedule"; -import { checkoutSessionLock } from "@/internal/billing/v2/actions/locks/checkoutSessionLock/checkoutSessionLock"; import { addStripeSubscriptionScheduleIdToBillingPlan } from "@/internal/billing/v2/execute/addStripeSubscriptionScheduleIdToBillingPlan"; import { executeAutumnBillingPlan } from "@/internal/billing/v2/execute/executeAutumnBillingPlan"; import { logAutumnBillingPlan } from "@/internal/billing/v2/utils/logs/logAutumnBillingPlan"; @@ -34,6 +34,24 @@ export const handleCheckoutSessionMetadataV2 = async ({ `[checkout.completed] Handling checkout session metadata V2: ${metadata.id}`, ); + await withClaimedCheckoutSessionMetadata({ + ctx, + checkoutContext, + metadata, + execute: () => + executeCheckoutSessionMetadataV2({ ctx, checkoutContext, metadata }), + }); +}; + +const executeCheckoutSessionMetadataV2 = async ({ + ctx, + checkoutContext, + metadata, +}: { + ctx: StripeWebhookContext; + checkoutContext: CheckoutSessionCompletedContext; + metadata: NonNullable; +}): Promise => { const deferredData = metadata.data as DeferredAutumnBillingPlanData; // 1. Sync Autumn metadata onto subscription items created by checkout @@ -96,14 +114,6 @@ export const handleCheckoutSessionMetadataV2 = async ({ billingPlan: updatedDeferredData.billingPlan, }); - // Clear checkout session lock now that customer_product rows exist - const lockCustomerId = - updatedDeferredData.billingContext.fullCustomer.id ?? - updatedDeferredData.billingContext.fullCustomer.internal_id; - if (lockCustomerId) { - await checkoutSessionLock.clear({ ctx, customerId: lockCustomerId }); - } - // Queue customer.products.updated webhook (mirrors executeBillingPlan) await billingPlanToSendProductsUpdated({ ctx, diff --git a/server/src/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/withClaimedCheckoutSessionMetadata.ts b/server/src/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/withClaimedCheckoutSessionMetadata.ts new file mode 100644 index 000000000..745780ee1 --- /dev/null +++ b/server/src/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/tasks/handleCheckoutSessionMetadataV2/withClaimedCheckoutSessionMetadata.ts @@ -0,0 +1,84 @@ +import { + type DeferredAutumnBillingPlanData, + MetadataType, +} from "@autumn/shared"; +import { setStripeSubscriptionLock } from "@/external/stripe/subscriptions/utils/lockStripeSubscriptionUtils"; +import type { CheckoutSessionCompletedContext } from "@/external/stripe/webhookHandlers/handleStripeCheckoutSessionCompleted/setupCheckoutSessionCompletedContext"; +import type { StripeWebhookContext } from "@/external/stripe/webhookMiddlewares/stripeWebhookContext"; +import { checkoutSessionLock } from "@/internal/billing/v2/actions/locks/checkoutSessionLock/checkoutSessionLock"; +import { MetadataService } from "@/internal/metadata/MetadataService"; + +/** + * Runs `execute` exactly once across concurrent executors of the same deferred + * plan. The subscription lock marks the resulting subscription.updated events + * as Autumn-initiated; the checkout session lock is cleared even on failure + * since the Stripe session is already paid. + */ +export const withClaimedCheckoutSessionMetadata = async ({ + ctx, + checkoutContext, + metadata, + execute, +}: { + ctx: StripeWebhookContext; + checkoutContext: CheckoutSessionCompletedContext; + metadata: NonNullable; + execute: () => Promise; +}): Promise => { + const claimed = await MetadataService.claim({ + db: ctx.db, + id: metadata.id, + fromType: MetadataType.CheckoutSessionV2, + toType: MetadataType.CheckoutSessionV2Processing, + }); + + if (!claimed) { + ctx.logger.info( + `[checkout.completed] Metadata ${metadata.id} already claimed by another executor, skipping`, + ); + return; + } + + if (checkoutContext.stripeSubscription) { + await setStripeSubscriptionLock({ + stripeSubscriptionId: checkoutContext.stripeSubscription.id, + lockedAtMs: Date.now(), + }); + } + + const deferredData = metadata.data as DeferredAutumnBillingPlanData; + const lockCustomerId = + deferredData?.billingContext?.fullCustomer?.id ?? + deferredData?.billingContext?.fullCustomer?.internal_id; + + try { + await execute(); + } catch (error) { + await revertMetadataClaim({ ctx, metadataId: metadata.id }); + throw error; + } finally { + if (lockCustomerId) { + await checkoutSessionLock.clear({ ctx, customerId: lockCustomerId }); + } + } +}; + +const revertMetadataClaim = async ({ + ctx, + metadataId, +}: { + ctx: StripeWebhookContext; + metadataId: string; +}): Promise => { + await MetadataService.claim({ + db: ctx.db, + id: metadataId, + fromType: MetadataType.CheckoutSessionV2Processing, + toType: MetadataType.CheckoutSessionV2, + }).catch((revertError) => { + ctx.logger.error( + `[checkout.completed] Failed to revert metadata claim for ${metadataId}`, + { revertError }, + ); + }); +}; diff --git a/server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleStripeSubscriptionCanceled/handleStripeSubscriptionCanceled.ts b/server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleStripeSubscriptionCanceled/handleStripeSubscriptionCanceled.ts index 897947271..03d61a435 100644 --- a/server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleStripeSubscriptionCanceled/handleStripeSubscriptionCanceled.ts +++ b/server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleStripeSubscriptionCanceled/handleStripeSubscriptionCanceled.ts @@ -1,4 +1,10 @@ -import { AttachScenario, cp, type FullCusProduct } from "@autumn/shared"; +import { + AttachScenario, + cp, + type FullCusProduct, + notNullish, +} from "@autumn/shared"; +import { msToSeconds } from "@shared/utils/common/unixUtils"; import { getStripeSubscriptionLock } from "@/external/stripe/subscriptions/utils/lockStripeSubscriptionUtils"; import type { StripeWebhookContext } from "@/external/stripe/webhookMiddlewares/stripeWebhookContext"; import { addProductsUpdatedWebhookTask } from "@/internal/analytics/handlers/handleProductsUpdated"; @@ -68,6 +74,13 @@ export const handleStripeSubscriptionCanceled = async ({ if (!isActiveRecurringAndOnSub) continue; + // attach-set ends_at, not an external cancellation + const endedAtMatchesCancelAt = + notNullish(customerProduct.ended_at) && + notNullish(cancelsAtMs) && + msToSeconds(customerProduct.ended_at!) === msToSeconds(cancelsAtMs!); + if (endedAtMatchesCancelAt) continue; + const updates = { canceled_at: canceledAtMs ?? Date.now(), canceled: true, diff --git a/server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleStripeSubscriptionRenewed/handleStripeSubscriptionRenewed.ts b/server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleStripeSubscriptionRenewed/handleStripeSubscriptionRenewed.ts index e5ee4764a..b5e04a89a 100644 --- a/server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleStripeSubscriptionRenewed/handleStripeSubscriptionRenewed.ts +++ b/server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleStripeSubscriptionRenewed/handleStripeSubscriptionRenewed.ts @@ -77,6 +77,9 @@ export const handleStripeSubscriptionRenewed = async ({ if (!valid) continue; + // attach-set ends_at expiry, not a cancellation + if (!customerProduct.canceled && !customerProduct.canceled_at) continue; + // Clear cancellation fields const updates = { canceled_at: null, diff --git a/server/src/internal/metadata/MetadataService.ts b/server/src/internal/metadata/MetadataService.ts index 8679fc2f1..2f7c31d30 100644 --- a/server/src/internal/metadata/MetadataService.ts +++ b/server/src/internal/metadata/MetadataService.ts @@ -54,6 +54,31 @@ export class MetadataService { return meta as Metadata; } + /** + * Atomically transitions a metadata row from one type to another. + * Returns true only for the caller whose update matched the `fromType` + * predicate — concurrent executors racing on the same row get false. + */ + static async claim({ + db, + id, + fromType, + toType, + }: { + db: DrizzleCli; + id: string; + fromType: MetadataType; + toType: MetadataType; + }): Promise { + const claimedRows = await db + .update(metadata) + .set({ type: toType }) + .where(and(eq(metadata.id, id), eq(metadata.type, fromType))) + .returning({ id: metadata.id }); + + return claimedRows.length > 0; + } + static async delete({ db, id }: { db: DrizzleCli; id: string }) { await db.delete(metadata).where(eq(metadata.id, id)); } diff --git a/server/tests/integration/billing/attach/checkout/stripe-checkout/stripe-checkout-ends-at.test.ts b/server/tests/integration/billing/attach/checkout/stripe-checkout/stripe-checkout-ends-at.test.ts new file mode 100644 index 000000000..e272e890a --- /dev/null +++ b/server/tests/integration/billing/attach/checkout/stripe-checkout/stripe-checkout-ends-at.test.ts @@ -0,0 +1,156 @@ +/** + * Stripe Checkout + ends_at Tests (Attach V2) + * + * Regression tests for ends_at surviving the Stripe Checkout deferred flow. + * + * Previously, after checkout completion: + * - A concurrent execution of the same deferred plan crashed the + * checkout.session.completed handler on a duplicate customer_products insert + * - The handler took no Stripe subscription lock, so the subscription.updated + * events it generated were misread as customer-initiated cancel/renew and + * handleStripeSubscriptionRenewed wiped ended_at off the customer product + * + * Expected behavior: + * - cancel_at lands on the Stripe subscription and stays there + * - customer_product.ended_at persists with canceled=false (an Autumn-owned + * expiry, not a cancellation) + */ + +import { expect, test } from "bun:test"; +import { + type ApiCustomerV5, + type AttachParamsV1Input, + MetadataType, +} from "@autumn/shared"; +import { getCustomerProduct } from "@tests/integration/billing/attach/params/start-date/utils"; +import { expectProductActive } from "@tests/integration/billing/utils/expectCustomerProductCorrect"; +import { completeStripeCheckoutFormV2 as completeStripeCheckoutForm } from "@tests/utils/browserPool/completeStripeCheckoutFormV2"; +import { items } from "@tests/utils/fixtures/items"; +import { products } from "@tests/utils/fixtures/products"; +import { timeout } from "@tests/utils/genUtils"; +import testContext from "@tests/utils/testInitUtils/createTestContext"; +import { initScenario, s } from "@tests/utils/testInitUtils/initScenario"; +import chalk from "chalk"; +import { addDays } from "date-fns"; +import { MetadataService } from "@/internal/metadata/MetadataService"; + +// ═══════════════════════════════════════════════════════════════════════════════ +// TEST 1: Stripe Checkout attach with ends_at → cancel_at + ended_at persist +// ═══════════════════════════════════════════════════════════════════════════════ + +test.concurrent( + `${chalk.yellowBright("stripe-checkout: ends_at sets cancel_at and persists ended_at")}`, + async () => { + const customerId = "stripe-checkout-ends-at"; + + const pro = products.pro({ + id: "pro-checkout-ends-at", + items: [items.monthlyMessages({ includedUsage: 100 })], + }); + + const { autumnV2_2, ctx, advancedTo } = await initScenario({ + customerId, + setup: [ + s.customer({ testClock: true }), // No payment method → stripe_checkout + s.products({ list: [pro] }), + ], + actions: [], + }); + + const endsAt = addDays(advancedTo, 7).getTime(); + + // 1. Attach with ends_at — should defer to Stripe Checkout + const result = await autumnV2_2.billing.attach({ + customer_id: customerId, + plan_id: pro.id, + ends_at: endsAt, + }); + + expect(result.payment_url).toBeDefined(); + expect(result.payment_url).toContain("checkout.stripe.com"); + + // 2. Complete checkout, then wait past the trailing subscription.updated + // events that previously wiped the cancellation fields + await completeStripeCheckoutForm({ url: result.payment_url }); + await timeout(15000); + + // 3. Product attached + const customer = await autumnV2_2.customers.get(customerId); + await expectProductActive({ customer, productId: pro.id }); + + // 4. ended_at persisted as an Autumn-owned expiry — not a cancellation + const customerProduct = await getCustomerProduct({ + ctx, + customerId, + productId: pro.id, + }); + expect(customerProduct.ended_at).toBe(endsAt); + expect(customerProduct.canceled).toBe(false); + expect(customerProduct.canceled_at ?? null).toBeNull(); + expect(customerProduct.subscription_ids).toHaveLength(1); + + // 5. cancel_at propagated onto the Stripe subscription and not cleared + const stripeSubscription = await ctx.stripeCli.subscriptions.retrieve( + customerProduct.subscription_ids![0]!, + ); + expect(stripeSubscription.cancel_at).toBe(Math.floor(endsAt / 1000)); + }, +); + +// ═══════════════════════════════════════════════════════════════════════════════ +// TEST 2: Metadata claim — exactly one concurrent executor wins +// ═══════════════════════════════════════════════════════════════════════════════ + +test.concurrent( + `${chalk.yellowBright("checkout metadata claim: only one concurrent executor wins")}`, + async () => { + const metadataId = `meta_claim_race_${Date.now()}`; + + await MetadataService.insert({ + db: testContext.db, + data: { + id: metadataId, + type: MetadataType.CheckoutSessionV2, + data: {}, + }, + }); + + try { + const claimResults = await Promise.all([ + MetadataService.claim({ + db: testContext.db, + id: metadataId, + fromType: MetadataType.CheckoutSessionV2, + toType: MetadataType.CheckoutSessionV2Processing, + }), + MetadataService.claim({ + db: testContext.db, + id: metadataId, + fromType: MetadataType.CheckoutSessionV2, + toType: MetadataType.CheckoutSessionV2Processing, + }), + ]); + + expect(claimResults.filter(Boolean)).toHaveLength(1); + + // Reverting the claim re-arms it for exactly one retry + const reverted = await MetadataService.claim({ + db: testContext.db, + id: metadataId, + fromType: MetadataType.CheckoutSessionV2Processing, + toType: MetadataType.CheckoutSessionV2, + }); + expect(reverted).toBe(true); + + const reclaimed = await MetadataService.claim({ + db: testContext.db, + id: metadataId, + fromType: MetadataType.CheckoutSessionV2, + toType: MetadataType.CheckoutSessionV2Processing, + }); + expect(reclaimed).toBe(true); + } finally { + await MetadataService.delete({ db: testContext.db, id: metadataId }); + } + }, +); diff --git a/shared/models/otherModels/metadataTable.ts b/shared/models/otherModels/metadataTable.ts index 8683aa7e3..45103c1d8 100644 --- a/shared/models/otherModels/metadataTable.ts +++ b/shared/models/otherModels/metadataTable.ts @@ -9,6 +9,7 @@ export enum MetadataType { DeferredInvoice = "deferred_invoice", CheckoutSessionV2 = "checkout_session_v2", + CheckoutSessionV2Processing = "checkout_session_v2_processing", CheckoutSessionEnabledImmediately = "checkout_session_enabled_immediately", SetupPaymentV2 = "setup_payment_v2", }