release

server/src/external/stripe/webhookHandlers/handleStripeSubscriptionCreated/handleStripeSubscriptionCreated.ts

@@ -1,6 +1,6 @@
 import type { StripeWebhookContext } from "../../webhookMiddlewares/stripeWebhookContext.js";
 import { setupStripeSubscriptionCreatedContext } from "./setupStripeSubscriptionCreatedContext.js";
-import { autoSyncFromSubscription } from "./tasks/autoSyncFromSubscription.js";
+import { autoSyncFromSubscriptionWithLock } from "./tasks/autoSyncFromSubscription.js";
 import { linkScheduledCustomerProductsToSubscription } from "./tasks/linkScheduledCustomerProductsToSubscription.js";
 
 export const handleStripeSubscriptionCreated = async ({
@@ -17,5 +17,5 @@ export const handleStripeSubscriptionCreated = async ({
 		subscription: subscriptionCreatedContext.subscription,
 	});
 
-	await autoSyncFromSubscription({ ctx, subscriptionCreatedContext });
+	await autoSyncFromSubscriptionWithLock({ ctx, subscriptionCreatedContext });
 };

server/src/external/stripe/webhookHandlers/handleStripeSubscriptionCreated/tasks/autoSyncFromSubscription.ts

@@ -3,27 +3,35 @@ import { billingActions } from "@/internal/billing/v2/actions";
 import { canAutoSync } from "@/internal/billing/v2/actions/sync/canAutoSync/index.js";
 import { subscriptionToSyncParams } from "@/internal/billing/v2/actions/sync/subscriptionToSyncParams.js";
 import { isAutumnCheckoutSubscription } from "@/internal/billing/v2/actions/sync/utils/isAutumnCheckoutSubscription.js";
+import { withStripeSyncCustomerLock } from "@/internal/billing/v2/actions/sync/utils/withStripeSyncCustomerLock.js";
+import { CusService } from "@/internal/customers/CusService.js";
 import { shouldSkipSubscriptionSync } from "../../common/subscriptionSync/shouldSkipSubscriptionSync.js";
 import type { StripeSubscriptionCreatedContext } from "../setupStripeSubscriptionCreatedContext.js";
 
 /**
  * Custom base prices can auto-sync; custom feature prices and unresolved items
  * still abort via canAutoSync.
  */
-export const autoSyncFromSubscription = async ({
-	ctx,
-	subscriptionCreatedContext,
-}: {
+type AutoSyncFromSubscriptionParams = {
 	ctx: StripeWebhookContext;
 	subscriptionCreatedContext: StripeSubscriptionCreatedContext;
-}) => {
+};
+
+const autoSyncFromSubscription = async ({
+	ctx,
+	subscriptionCreatedContext,
+}: AutoSyncFromSubscriptionParams) => {
 	const { logger, stripeCli } = ctx;
 	const { subscription, fullCustomer } = subscriptionCreatedContext;
 	const customerId = fullCustomer.id ?? fullCustomer.internal_id;
+	const currentCustomer = await CusService.getFull({
+		ctx,
+		idOrInternalId: customerId,
+	});
 
 	const skip = shouldSkipSubscriptionSync({
 		subscription,
-		fullCustomer,
+		fullCustomer: currentCustomer,
 		requireRecent: false,
 	});
 	if (skip.skip) {
@@ -44,7 +52,7 @@ export const autoSyncFromSubscription = async ({
 		ctx,
 		customerId,
 		subscription,
-		customerProducts: fullCustomer.customer_products,
+		customerProducts: currentCustomer.customer_products,
 	});
 
 	const eligibility = canAutoSync({ match });
@@ -57,3 +65,17 @@ export const autoSyncFromSubscription = async ({
 
 	await billingActions.syncV2({ ctx, params });
 };
+
+export const autoSyncFromSubscriptionWithLock = async (
+	params: AutoSyncFromSubscriptionParams,
+) => {
+	const { ctx, subscriptionCreatedContext } = params;
+	const customerId =
+		subscriptionCreatedContext.fullCustomer.id ??
+		subscriptionCreatedContext.fullCustomer.internal_id;
+	await withStripeSyncCustomerLock({
+		ctx,
+		customerId,
+		run: () => autoSyncFromSubscription(params),
+	});
+};

server/src/internal/billing/v2/actions/sync/autoSyncStripeCustomer.ts

@@ -0,0 +1,44 @@
+import type { AutumnContext } from "@/honoUtils/HonoEnv";
+import { canAutoSync } from "./canAutoSync";
+import { prepareAutoSyncStripeCustomer } from "./setup/prepareAutoSyncStripeCustomer";
+import { syncV2 } from "./syncV2";
+import { withStripeSyncCustomerLock } from "./utils/withStripeSyncCustomerLock";
+
+const autoSyncStripeCustomer = async ({
+	ctx,
+	customerId,
+	stripeCustomerId,
+}: {
+	ctx: AutumnContext;
+	customerId: string;
+	stripeCustomerId: string;
+}) => {
+	const syncCandidates = await prepareAutoSyncStripeCustomer({
+		ctx,
+		customerId,
+		stripeCustomerId,
+	});
+	for (const syncCandidate of syncCandidates) {
+		if (!syncCandidate) continue;
+		const { match, params } = syncCandidate;
+		if (!canAutoSync({ match }).eligible) continue;
+		await syncV2({
+			ctx,
+			params,
+			tags: ["sync:customer.create"],
+		});
+	}
+};
+
+export const autoSyncStripeCustomerWithLock = (params: {
+	ctx: AutumnContext;
+	customerId: string;
+	stripeCustomerId: string;
+}) => {
+	const { ctx, customerId } = params;
+	return withStripeSyncCustomerLock({
+		ctx,
+		customerId,
+		run: () => autoSyncStripeCustomer(params),
+	});
+};

server/src/internal/billing/v2/actions/sync/setup/prepareAutoSyncStripeCustomer.ts

@@ -0,0 +1,82 @@
+import {
+	filterCustomerProductsByStripeSubscriptionId,
+	isCustomerProductOnStripeSubscriptionSchedule,
+} from "@autumn/shared";
+import { createStripeCli } from "@/external/connect/createStripeCli";
+import type { AutumnContext } from "@/honoUtils/HonoEnv";
+import { fetchStripeSyncSchedule } from "@/internal/billing/v2/providers/stripe/utils/sync/fetchStripeSyncObjects";
+import { CusService } from "@/internal/customers/CusService";
+import { ProductService } from "@/internal/products/ProductService";
+import { subscriptionToSyncParams } from "../subscriptionToSyncParams";
+
+export const prepareAutoSyncStripeCustomer = async ({
+	ctx,
+	customerId,
+	stripeCustomerId,
+}: {
+	ctx: AutumnContext;
+	customerId: string;
+	stripeCustomerId: string;
+}) => {
+	const stripeCli = createStripeCli({ org: ctx.org, env: ctx.env });
+	const [subscriptions, schedules, customer] = await Promise.all([
+		stripeCli.subscriptions
+			.list({ customer: stripeCustomerId, limit: 100 })
+			.autoPagingToArray({ limit: 10_000 }),
+		stripeCli.subscriptionSchedules
+			.list({ customer: stripeCustomerId, scheduled: true, limit: 100 })
+			.autoPagingToArray({ limit: 10_000 }),
+		CusService.getFull({ ctx, idOrInternalId: customerId }),
+	]);
+	const pendingSubscriptions = subscriptions.filter(
+		(subscription) =>
+			filterCustomerProductsByStripeSubscriptionId({
+				customerProducts: customer.customer_products,
+				stripeSubscriptionId: subscription.id,
+			}).length === 0,
+	);
+	const pendingSchedules = schedules.filter(
+		(schedule) =>
+			!customer.customer_products.some((customerProduct) =>
+				isCustomerProductOnStripeSubscriptionSchedule({
+					customerProduct,
+					stripeSubscriptionScheduleId: schedule.id,
+				}),
+			),
+	);
+	if (pendingSubscriptions.length === 0 && pendingSchedules.length === 0) {
+		return [];
+	}
+
+	const fullProducts = await ProductService.listFull({
+		db: ctx.db,
+		orgId: ctx.org.id,
+		env: ctx.env,
+	});
+	return Promise.all([
+		...pendingSubscriptions.map((subscription) =>
+			subscriptionToSyncParams({
+				ctx,
+				customerId,
+				subscription,
+				customerProducts: customer.customer_products,
+				fullProducts,
+			}),
+		),
+		...pendingSchedules.map(async ({ id }) => {
+			const schedule = await fetchStripeSyncSchedule({
+				stripeCli,
+				scheduleId: id,
+			});
+			return schedule
+				? subscriptionToSyncParams({
+						ctx,
+						customerId,
+						schedule,
+						customerProducts: customer.customer_products,
+						fullProducts,
+					})
+				: null;
+		}),
+	]);
+};

server/src/internal/billing/v2/actions/sync/utils/withStripeSyncCustomerLock.ts

@@ -0,0 +1,41 @@
+import { ErrCode, ms, RecaseError } from "@autumn/shared";
+import { acquireLock, clearLock } from "@/external/redis/redisUtils";
+import type { AutumnContext } from "@/honoUtils/HonoEnv";
+import { timeout } from "@/utils/genUtils";
+
+const LOCK_TIMEOUT_MS = ms.seconds(30);
+
+/** Serializes initial imports with subscription-created webhooks and waits for the winner. */
+export const withStripeSyncCustomerLock = async <T>({
+	ctx,
+	customerId,
+	run,
+}: {
+	ctx: AutumnContext;
+	customerId: string;
+	run: () => Promise<T>;
+}) => {
+	const lockKey = `lock:stripe-sync:${ctx.org.id}:${ctx.env}:${customerId}`;
+	const deadline = Date.now() + LOCK_TIMEOUT_MS;
+	while (true) {
+		try {
+			await acquireLock({ lockKey, ttlMs: LOCK_TIMEOUT_MS });
+			break;
+		} catch (error) {
+			if (
+				!(error instanceof RecaseError) ||
+				error.code !== ErrCode.LockAlreadyExists ||
+				Date.now() >= deadline
+			) {
+				throw error;
+			}
+			await timeout(250);
+		}
+	}
+
+	try {
+		return await run();
+	} finally {
+		await clearLock({ lockKey });
+	}
+};

server/src/internal/customers/actions/createWithDefaults/createCustomerWithDefaults.ts

@@ -1,4 +1,8 @@
-import type { CustomerData, FullCustomer } from "@autumn/shared";
+import {
+	type CustomerData,
+	type FullCustomer,
+	hasActivePaidSubscription,
+} from "@autumn/shared";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { evaluateStripeBillingPlan } from "@/internal/billing/v2/providers/stripe/actionBuilders/evaluateStripeBillingPlan.js";
 import { executeStripeBillingPlan } from "@/internal/billing/v2/providers/stripe/execute/executeStripeBillingPlan.js";
@@ -14,6 +18,7 @@ import {
 } from "./logs/logCreateCustomer.js";
 import { setupCreateCustomer } from "./setup/setupCreateCustomer.js";
 import { setupCreateCustomerBillingContext } from "./setup/setupCreateCustomerBillingContext.js";
+import { syncCreatedCustomerFromStripe } from "./syncCreatedCustomerFromStripe.js";
 
 /**
  * Create a customer and attach default products.
@@ -72,12 +77,22 @@ export const createCustomerWithDefaults = async ({
 	// drop the webhooks — a client retry lands on the "existing" path above and
 	// would never emit them.
 	try {
+		context.fullCustomer = await syncCreatedCustomerFromStripe({
+			ctx,
+			fullCustomer: context.fullCustomer,
+			stripeCustomerId: customerData?.stripe_id,
+		});
+
 		// 4. Setup billing context (creates Stripe customer)
 
 		const shouldCreateStripeCustomer =
 			customerData?.create_in_stripe || context.hasPaidProducts;
 
-		const shouldAttachPaidDefaults = context.hasPaidProducts;
+		const shouldAttachPaidDefaults =
+			context.hasPaidProducts &&
+			!hasActivePaidSubscription({
+				customerProducts: context.fullCustomer.customer_products,
+			});
 
 		if (!shouldCreateStripeCustomer) return context.fullCustomer;
 

server/src/internal/customers/actions/createWithDefaults/syncCreatedCustomerFromStripe.ts

@@ -0,0 +1,30 @@
+import { CustomerExpand, type FullCustomer } from "@autumn/shared";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import { autoSyncStripeCustomerWithLock } from "@/internal/billing/v2/actions/sync/autoSyncStripeCustomer.js";
+import { CusService } from "@/internal/customers/CusService.js";
+
+export const syncCreatedCustomerFromStripe = async ({
+	ctx,
+	fullCustomer,
+	stripeCustomerId,
+}: {
+	ctx: AutumnContext;
+	fullCustomer: FullCustomer;
+	stripeCustomerId?: string | null;
+}) => {
+	if (!stripeCustomerId) return fullCustomer;
+
+	const customerId = fullCustomer.id ?? fullCustomer.internal_id;
+	await autoSyncStripeCustomerWithLock({
+		ctx,
+		customerId,
+		stripeCustomerId,
+	});
+	return CusService.getFull({
+		ctx,
+		idOrInternalId: customerId,
+		withEntities: true,
+		withSubs: true,
+		expand: [CustomerExpand.Invoices],
+	});
+};