fix(billing): compare membership-driven seat and cancel changes against the committed value, not a possibly-stale row

Author: waleedlatif1

apps/sim/lib/billing/organizations/membership.ts

@@ -48,7 +48,10 @@ import {
 import { toDecimal, toNumber } from '@/lib/billing/utils/decimal'
 import { validateSeatAvailability } from '@/lib/billing/validation/seat-management'
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
-import { enqueueCancelAtPeriodEndSync } from '@/lib/billing/webhooks/subscription-sync'
+import {
+  enqueueCancelAtPeriodEndSync,
+  readCommittedCancelAtPeriodEnd,
+} from '@/lib/billing/webhooks/subscription-sync'
 import { isBillingEnabled } from '@/lib/core/config/env-flags'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
@@ -259,7 +262,13 @@ export async function restoreUserProSubscription(userId: string): Promise<Restor
       .for('update')
       .limit(1)
 
-    if (!personalPro?.cancelAtPeriodEnd || !personalPro.stripeSubscriptionId) return
+    if (!personalPro?.stripeSubscriptionId) return
+    const pausing = await readCommittedCancelAtPeriodEnd(
+      tx,
+      personalPro.id,
+      Boolean(personalPro.cancelAtPeriodEnd)
+    )
+    if (!pausing) return
     result.subscriptionId = personalPro.id
 
     const organizationMemberships = await tx
@@ -405,7 +414,15 @@ export async function pauseProSubscriptionForOrgCoverage(
 
     result.subscriptionId = personalPro.id
 
-    if (personalPro.cancelAtPeriodEnd) return
+    if (
+      await readCommittedCancelAtPeriodEnd(
+        tx,
+        personalPro.id,
+        Boolean(personalPro.cancelAtPeriodEnd)
+      )
+    ) {
+      return
+    }
 
     await tx
       .update(subscriptionTable)
@@ -855,7 +872,14 @@ async function applyPaidOrgJoinBillingTx(
     .for('update')
     .limit(1)
 
-  if (personalPro && !personalPro.cancelAtPeriodEnd) {
+  const alreadyPausing =
+    personalPro &&
+    (await readCommittedCancelAtPeriodEnd(
+      tx,
+      personalPro.id,
+      Boolean(personalPro.cancelAtPeriodEnd)
+    ))
+  if (personalPro && !alreadyPausing) {
     await tx
       .update(subscriptionTable)
       .set({ cancelAtPeriodEnd: true })

apps/sim/lib/billing/organizations/provision-seat.ts

@@ -19,6 +19,8 @@ import { hasUsableSubscriptionStatus } from '@/lib/billing/subscriptions/utils'
 import {
   enqueueCancelAtPeriodEndSync,
   enqueueSubscriptionSeatsSync,
+  readCommittedCancelAtPeriodEnd,
+  readCommittedSeats,
   recordCancelAtPeriodEnd,
 } from '@/lib/billing/webhooks/subscription-sync'
 import type { DbOrTx, DbTransaction } from '@/lib/db/types'
@@ -267,13 +269,13 @@ async function activateTeamSubscription(
   if (planChanged) {
     await enqueueSubscriptionSeatsSync(tx, {
       subscriptionId: sub.id,
-      seats: locked?.seats ?? 1,
+      seats: await readCommittedSeats(tx, sub.id, locked?.seats ?? 1),
       reason: 'pro-to-team-conversion',
     })
   }
 
   if (!locked?.stripeSubscriptionId) return
-  if (locked.cancelAtPeriodEnd) {
+  if (await readCommittedCancelAtPeriodEnd(tx, sub.id, Boolean(locked.cancelAtPeriodEnd))) {
     await enqueueCancelAtPeriodEndSync(tx, {
       stripeSubscriptionId: locked.stripeSubscriptionId,
       subscriptionId: sub.id,

apps/sim/lib/billing/organizations/seats.ts

@@ -6,7 +6,10 @@ import { and, count, desc, eq, inArray } from 'drizzle-orm'
 import { syncSubscriptionUsageLimits } from '@/lib/billing/organization'
 import { isTeam } from '@/lib/billing/plan-helpers'
 import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/utils'
-import { enqueueSubscriptionSeatsSync } from '@/lib/billing/webhooks/subscription-sync'
+import {
+  enqueueSubscriptionSeatsSync,
+  readCommittedSeats,
+} from '@/lib/billing/webhooks/subscription-sync'
 import { isBillingEnabled } from '@/lib/core/config/env-flags'
 import { captureServerEvent } from '@/lib/posthog/server'
 
@@ -100,7 +103,11 @@ export async function reconcileOrganizationSeats({
       .where(eq(member.organizationId, organizationId))
 
     const targetSeats = Math.max(1, memberCountRow?.value ?? 1)
-    const currentSeats = orgSubscription.seats ?? 1
+    const currentSeats = await readCommittedSeats(
+      tx,
+      orgSubscription.id,
+      orgSubscription.seats ?? 1
+    )
 
     if (targetSeats === currentSeats) {
       return { kind: 'noop', seats: currentSeats }

apps/sim/lib/billing/webhooks/stripe-sync-convergence.integration.ts

@@ -853,6 +853,26 @@ describe('cancel_at_period_end sync', () => {
     expect(stripe.subscription(pro.stripeSubscriptionId).cancel_at_period_end).toBe(true)
   })
 
+  it('restores the personal Pro when its member leaves while the plugin has overwritten the row', async () => {
+    const pro = await createProUserInPaidOrganization()
+    await pauseProSubscriptionForOrgCoverage(pro.userId)
+    const pauseSync = await latestOutboxEventId(
+      OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
+      pro.subscriptionId
+    )
+
+    beforeReconcile = async () => {
+      beforeReconcile = undefined
+      await leaveOrganization(pro.userId, pro.paidOrganization.organizationId)
+      await restoreUserProSubscription(pro.userId)
+    }
+    await deliverUnrelatedUpdate(pro.stripeSubscriptionId)
+
+    expect((await storedSubscription(pro.subscriptionId)).cancelAtPeriodEnd).toBe(false)
+    await expect(processEvent(pauseSync)).resolves.toBe('completed')
+    expect(stripe.subscription(pro.stripeSubscriptionId).cancel_at_period_end).toBe(false)
+  })
+
   it('does not revive an older value when a retry path resets its sync without re-committing', async () => {
     const pro = await createProUserInPaidOrganization()
     await pauseProSubscriptionForOrgCoverage(pro.userId)
@@ -1237,6 +1257,32 @@ describe('Team seat sync', () => {
     expect(stripe.subscription(org.stripeSubscriptionId).items.data[0].quantity).toBe(2)
   })
 
+  it('drops a seat when a member leaves while the plugin has overwritten the row', async () => {
+    const [owner, joiner] = await Promise.all([createUser('owner'), createUser('joiner')])
+    const org = await createOrganizationWithPlan('team', 1)
+    await addMember(org.organizationId, owner.id, 'owner')
+    await addMember(org.organizationId, joiner.id)
+    await reconcileOrganizationSeats({ organizationId: org.organizationId, reason: 'member-added' })
+    const growSync = await latestOutboxEventId(
+      OUTBOX_EVENT_TYPES.STRIPE_SYNC_SUBSCRIPTION_SEATS,
+      org.subscriptionId
+    )
+
+    beforeReconcile = async () => {
+      beforeReconcile = undefined
+      await leaveOrganization(joiner.id, org.organizationId)
+      await reconcileOrganizationSeats({
+        organizationId: org.organizationId,
+        reason: 'member-removed',
+      })
+    }
+    await deliverUnrelatedUpdate(org.stripeSubscriptionId)
+
+    expect((await storedSubscription(org.subscriptionId)).seats).toBe(1)
+    await expect(processEvent(growSync)).resolves.toBe('completed')
+    expect(stripe.subscription(org.stripeSubscriptionId).items.data[0].quantity).toBe(1)
+  })
+
   it('does not revive an older seat count when its dead-lettered sync is requeued', async () => {
     const [owner, joiner] = await Promise.all([createUser('owner'), createUser('joiner')])
     const org = await createOrganizationWithPlan('team', 1)

apps/sim/lib/billing/webhooks/subscription-sync.ts

@@ -215,18 +215,18 @@ export async function recommitSubscriptionSync(
     .limit(1)
   if (!current) return
 
-  const pending = await readSyncIntents(tx, subscriptionId)
   if (eventType === CANCEL_SYNC) {
-    const intent = pending.cancelAtPeriodEnd
     await commitIntent(tx, eventType, subscriptionId, {
-      cancelAtPeriodEnd:
-        intent.status === 'value' ? intent.value : Boolean(current.cancelAtPeriodEnd),
+      cancelAtPeriodEnd: await readCommittedCancelAtPeriodEnd(
+        tx,
+        subscriptionId,
+        Boolean(current.cancelAtPeriodEnd)
+      ),
     })
     return
   }
-  const intent = pending.seats
   await commitIntent(tx, eventType, subscriptionId, {
-    seats: intent.status === 'value' ? intent.value : (current.seats ?? 1),
+    seats: await readCommittedSeats(tx, subscriptionId, current.seats ?? 1),
   })
 }
 
@@ -304,6 +304,31 @@ export async function readRecordedSyncValue(
   }
 }
 
+/**
+ * The subscription's latest committed `cancelAtPeriodEnd`: the newest value an in-flight sync
+ * records, else `stored` (the row). Until the reconcile step runs, the row can hold the Stripe
+ * plugin's stale webhook payload, so a writer deciding whether a change is needed compares
+ * against this, never the row alone. The caller holds the subscription row lock.
+ */
+export async function readCommittedCancelAtPeriodEnd(
+  tx: DbOrTx,
+  subscriptionId: string,
+  stored: boolean
+): Promise<boolean> {
+  const intent = (await readSyncIntents(tx, subscriptionId)).cancelAtPeriodEnd
+  return intent.status === 'value' ? intent.value : stored
+}
+
+/** The seat-count counterpart of {@link readCommittedCancelAtPeriodEnd}. */
+export async function readCommittedSeats(
+  tx: DbOrTx,
+  subscriptionId: string,
+  stored: number
+): Promise<number> {
+  const intent = (await readSyncIntents(tx, subscriptionId)).seats
+  return intent.status === 'value' ? intent.value : stored
+}
+
 /** A fresh key per Stripe write: the SDK reuses it across its own network retries of that call. */
 export function cancelAtPeriodEndSyncIdempotencyKey(eventId: string): string {
   return `${CANCEL_AT_PERIOD_END_SYNC_KEY_PREFIX}${eventId}:${generateShortId()}`

packages/testing/src/mocks/billing-subscription-sync.mock.ts

@@ -3,7 +3,8 @@ import { vi } from 'vitest'
 /**
  * Controllable mock functions for `@/lib/billing/webhooks/subscription-sync`. The enqueue
  * functions resolve to a fixed event id; drive them with `mockResolvedValueOnce`.
- * `mockIsSubscriptionSyncEventType` keeps the real logic.
+ * `mockIsSubscriptionSyncEventType` keeps the real logic; the `mockReadCommitted*` readers
+ * return the stored value they are given, as when nothing is in flight.
  *
  * @example
  * ```ts
@@ -32,6 +33,12 @@ export const billingSubscriptionSyncMockFns = {
   mockReconcileSubscriptionSyncFromStripe: vi.fn(async () => undefined),
   mockRecordCustomerRestoreAfterHook: vi.fn(async () => undefined),
   mockReadRecordedSyncValue: vi.fn(async () => undefined),
+  mockReadCommittedCancelAtPeriodEnd: vi.fn(
+    async (_tx: unknown, _subscriptionId: string, stored: boolean) => stored
+  ),
+  mockReadCommittedSeats: vi.fn(
+    async (_tx: unknown, _subscriptionId: string, stored: number) => stored
+  ),
 }
 
 /**
@@ -55,4 +62,6 @@ export const billingSubscriptionSyncMock = {
     billingSubscriptionSyncMockFns.mockReconcileSubscriptionSyncFromStripe,
   recordCustomerRestoreAfterHook: billingSubscriptionSyncMockFns.mockRecordCustomerRestoreAfterHook,
   readRecordedSyncValue: billingSubscriptionSyncMockFns.mockReadRecordedSyncValue,
+  readCommittedCancelAtPeriodEnd: billingSubscriptionSyncMockFns.mockReadCommittedCancelAtPeriodEnd,
+  readCommittedSeats: billingSubscriptionSyncMockFns.mockReadCommittedSeats,
 }