fix: ends at propogation

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<CheckoutSessionCompletedContext["metadata"]>;
+}): Promise<void> => {
 	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,

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<CheckoutSessionCompletedContext["metadata"]>;
+	execute: () => Promise<void>;
+}): Promise<void> => {
+	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<void> => {
+	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 },
+		);
+	});
+};

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,

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,

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<boolean> {
+		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));
 	}

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<AttachParamsV1Input>({
+			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<ApiCustomerV5>(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 });
+		}
+	},
+);

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",
 }