fix(billing): a membership writer skips only when both the row and the committed value already match

Author: waleedlatif1

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

@@ -50,7 +50,7 @@ import { validateSeatAvailability } from '@/lib/billing/validation/seat-manageme
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
 import {
   enqueueCancelAtPeriodEndSync,
-  readCommittedCancelAtPeriodEnd,
+  isCancelAtPeriodEndSettled,
 } from '@/lib/billing/webhooks/subscription-sync'
 import { isBillingEnabled } from '@/lib/core/config/env-flags'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
@@ -263,12 +263,16 @@ export async function restoreUserProSubscription(userId: string): Promise<Restor
       .limit(1)
 
     if (!personalPro?.stripeSubscriptionId) return
-    const pausing = await readCommittedCancelAtPeriodEnd(
-      tx,
-      personalPro.id,
-      Boolean(personalPro.cancelAtPeriodEnd)
-    )
-    if (!pausing) return
+    if (
+      await isCancelAtPeriodEndSettled(
+        tx,
+        personalPro.id,
+        Boolean(personalPro.cancelAtPeriodEnd),
+        false
+      )
+    ) {
+      return
+    }
     result.subscriptionId = personalPro.id
 
     const organizationMemberships = await tx
@@ -415,10 +419,11 @@ export async function pauseProSubscriptionForOrgCoverage(
     result.subscriptionId = personalPro.id
 
     if (
-      await readCommittedCancelAtPeriodEnd(
+      await isCancelAtPeriodEndSettled(
         tx,
         personalPro.id,
-        Boolean(personalPro.cancelAtPeriodEnd)
+        Boolean(personalPro.cancelAtPeriodEnd),
+        true
       )
     ) {
       return
@@ -874,10 +879,11 @@ async function applyPaidOrgJoinBillingTx(
 
   const alreadyPausing =
     personalPro &&
-    (await readCommittedCancelAtPeriodEnd(
+    (await isCancelAtPeriodEndSettled(
       tx,
       personalPro.id,
-      Boolean(personalPro.cancelAtPeriodEnd)
+      Boolean(personalPro.cancelAtPeriodEnd),
+      true
     ))
   if (personalPro && !alreadyPausing) {
     await tx

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

@@ -19,7 +19,7 @@ import { hasUsableSubscriptionStatus } from '@/lib/billing/subscriptions/utils'
 import {
   enqueueCancelAtPeriodEndSync,
   enqueueSubscriptionSeatsSync,
-  readCommittedCancelAtPeriodEnd,
+  isCancelAtPeriodEndSettled,
   readCommittedSeats,
   recordCancelAtPeriodEnd,
 } from '@/lib/billing/webhooks/subscription-sync'
@@ -275,7 +275,7 @@ async function activateTeamSubscription(
   }
 
   if (!locked?.stripeSubscriptionId) return
-  if (await readCommittedCancelAtPeriodEnd(tx, sub.id, Boolean(locked.cancelAtPeriodEnd))) {
+  if (!(await isCancelAtPeriodEndSettled(tx, sub.id, Boolean(locked.cancelAtPeriodEnd), false))) {
     await enqueueCancelAtPeriodEndSync(tx, {
       stripeSubscriptionId: locked.stripeSubscriptionId,
       subscriptionId: sub.id,

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

@@ -109,7 +109,7 @@ export async function reconcileOrganizationSeats({
       orgSubscription.seats ?? 1
     )
 
-    if (targetSeats === currentSeats) {
+    if (targetSeats === currentSeats && targetSeats === (orgSubscription.seats ?? 1)) {
       return { kind: 'noop', seats: currentSeats }
     }
 

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

@@ -853,6 +853,33 @@ describe('cancel_at_period_end sync', () => {
     expect(stripe.subscription(pro.stripeSubscriptionId).cancel_at_period_end).toBe(true)
   })
 
+  it('keeps a pause committed after the reconcile read a customer restore from Stripe', async () => {
+    const pro = await createProUserInPaidOrganization()
+    await pauseProSubscriptionForOrgCoverage(pro.userId)
+    const pauseSync = await latestOutboxEventId(
+      OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
+      pro.subscriptionId
+    )
+    stripe.failNextUpdateAfterApplying('subscriptions')
+    await expect(processEvent(pauseSync)).resolves.toBe('pending')
+
+    stripe.updateOutsideSim(pro.stripeSubscriptionId, { cancel_at_period_end: false })
+    const liveRead = stripe.holdNextRequest('subscriptions.retrieve')
+    const reconcilingRestore = deliver(stripe.events.at(-1) as Stripe.Event)
+    await liveRead.reached
+    await pauseProSubscriptionForOrgCoverage(pro.userId)
+    liveRead.release()
+    await reconcilingRestore
+
+    expect((await storedSubscription(pro.subscriptionId)).cancelAtPeriodEnd).toBe(true)
+    const latestSync = await latestOutboxEventId(
+      OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
+      pro.subscriptionId
+    )
+    await expect(processEvent(latestSync)).resolves.toBe('completed')
+    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)

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

@@ -310,7 +310,7 @@ export async function readRecordedSyncValue(
  * 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(
+async function readCommittedCancelAtPeriodEnd(
   tx: DbOrTx,
   subscriptionId: string,
   stored: boolean
@@ -319,6 +319,24 @@ export async function readCommittedCancelAtPeriodEnd(
   return intent.status === 'value' ? intent.value : stored
 }
 
+/**
+ * True when both the row and the latest committed `cancelAtPeriodEnd` already hold `desired`, so
+ * a writer has nothing to record. A writer that sees either one differ writes the row and
+ * commits: a redundant sync of the same value is harmless, while skipping on the committed value
+ * alone could let an older Stripe read, accepted after this writer, override it.
+ */
+export async function isCancelAtPeriodEndSettled(
+  tx: DbOrTx,
+  subscriptionId: string,
+  stored: boolean,
+  desired: boolean
+): Promise<boolean> {
+  return (
+    stored === desired &&
+    (await readCommittedCancelAtPeriodEnd(tx, subscriptionId, stored)) === desired
+  )
+}
+
 /** The seat-count counterpart of {@link readCommittedCancelAtPeriodEnd}. */
 export async function readCommittedSeats(
   tx: DbOrTx,
@@ -363,9 +381,10 @@ function latestIntent<T>(
 }
 
 /**
- * One indexed read of the subscription's in-flight syncs. Dead letters are not intents: they are
- * failed syncs awaiting an operator, kept current by every commit so a retry pushes the latest
- * value, but never a reason to override Stripe.
+ * One read of the subscription's in-flight syncs, through the status index and bounded by the
+ * in-flight backlog. Dead letters are not intents: they are failed syncs awaiting an operator,
+ * kept current by every commit so a retry pushes the latest value, but never a reason to override
+ * Stripe.
  */
 async function readSyncIntents(executor: DbOrTx, subscriptionId: string) {
   const events = await listInflightOutboxEvents(

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

@@ -3,8 +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; the `mockReadCommitted*` readers
- * return the stored value they are given, as when nothing is in flight.
+ * `mockIsSubscriptionSyncEventType` keeps the real logic; `mockReadCommittedSeats` and
+ * `mockIsCancelAtPeriodEndSettled` answer from the stored value, as when nothing is in flight.
  *
  * @example
  * ```ts
@@ -33,8 +33,9 @@ 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
+  mockIsCancelAtPeriodEndSettled: vi.fn(
+    async (_tx: unknown, _subscriptionId: string, stored: boolean, desired: boolean) =>
+      stored === desired
   ),
   mockReadCommittedSeats: vi.fn(
     async (_tx: unknown, _subscriptionId: string, stored: number) => stored
@@ -62,6 +63,6 @@ export const billingSubscriptionSyncMock = {
     billingSubscriptionSyncMockFns.mockReconcileSubscriptionSyncFromStripe,
   recordCustomerRestoreAfterHook: billingSubscriptionSyncMockFns.mockRecordCustomerRestoreAfterHook,
   readRecordedSyncValue: billingSubscriptionSyncMockFns.mockReadRecordedSyncValue,
-  readCommittedCancelAtPeriodEnd: billingSubscriptionSyncMockFns.mockReadCommittedCancelAtPeriodEnd,
+  isCancelAtPeriodEndSettled: billingSubscriptionSyncMockFns.mockIsCancelAtPeriodEndSettled,
   readCommittedSeats: billingSubscriptionSyncMockFns.mockReadCommittedSeats,
 }