fix(billing): push each sync's recorded value and reject Stripe reads older than a newer Sim commit

Author: waleedlatif1

apps/sim/lib/admin/subscription-lifecycle.test.ts

@@ -2,10 +2,7 @@ import { outboxEvent, subscription } from '@sim/db/schema'
 import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing'
 import { auditMock, auditMockFns } from '@sim/testing/mocks/audit.mock'
 import { billingOutboxHandlersMock } from '@sim/testing/mocks/billing-outbox-handlers.mock'
-import {
-  billingSubscriptionSyncMock,
-  billingSubscriptionSyncMockFns,
-} from '@sim/testing/mocks/billing-subscription-sync.mock'
+import { billingSubscriptionSyncMock } from '@sim/testing/mocks/billing-subscription-sync.mock'
 import { organizationMembershipMock } from '@sim/testing/mocks/organization-membership.mock'
 import { outboxServiceMock, outboxServiceMockFns } from '@sim/testing/mocks/outbox-service.mock'
 import { stripeClientMock } from '@sim/testing/mocks/stripe.mock'
@@ -125,15 +122,6 @@ describe('admin subscription cancellation', () => {
       expect.objectContaining({ status: 'pending', attempts: 0, lastError: null })
     )
     expect(dbChainMockFns.set).toHaveBeenCalledWith({ cancelAtPeriodEnd: true })
-    expect(billingSubscriptionSyncMockFns.mockLockSubscriptionForSyncRetry).toHaveBeenCalledWith(
-      expect.anything(),
-      'sub-row-1'
-    )
-    expect(billingSubscriptionSyncMockFns.mockRecommitSubscriptionSync).toHaveBeenCalledWith(
-      expect.anything(),
-      'stripe.sync-cancel-at-period-end',
-      'sub-row-1'
-    )
     expect(result).toMatchObject({
       operationId: '67e55044-10b1-426f-9247-bb680e5fe0c8',
       status: 'pending',

apps/sim/lib/auth/auth.ts

@@ -104,8 +104,8 @@ import {
   handleSubscriptionDeleted,
 } from '@/lib/billing/webhooks/subscription'
 import {
-  commitCustomerRestoredSubscription,
   reconcileSubscriptionSyncFromStripe,
+  recordCustomerRestoreAfterHook,
 } from '@/lib/billing/webhooks/subscription-sync'
 import { handleSubscriptionUsageUpdate } from '@/lib/billing/webhooks/subscription-usage'
 import { env } from '@/lib/core/config/env'
@@ -1104,15 +1104,7 @@ export const auth = betterAuth({
       return
     }),
     after: createAuthMiddleware(async (ctx) => {
-      if (isBillingEnabled && ctx.path === '/subscription/restore') {
-        try {
-          await commitCustomerRestoredSubscription(ctx.context.returned)
-        } catch (error) {
-          logger.error('Failed to record a restored subscription as the committed value', {
-            error,
-          })
-        }
-      }
+      if (isBillingEnabled) await recordCustomerRestoreAfterHook(ctx)
 
       if (isBillingEnabled && ctx.path === '/subscription/upgrade') {
         const checkoutContext = ctx as typeof ctx & {

apps/sim/lib/billing/organizations/pause-pro-for-coverage.test.ts

@@ -1,8 +1,5 @@
 import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
-import {
-  billingSubscriptionSyncMock,
-  billingSubscriptionSyncMockFns,
-} from '@sim/testing/mocks/billing-subscription-sync.mock'
+import { billingSubscriptionSyncMock } from '@sim/testing/mocks/billing-subscription-sync.mock'
 import { outboxServiceMock } from '@sim/testing/mocks/outbox-service.mock'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
@@ -16,9 +13,6 @@ vi.mock('@/lib/billing/webhooks/subscription-sync', () => billingSubscriptionSyn
 
 import { pauseProSubscriptionForOrgCoverage } from '@/lib/billing/organizations/membership'
 
-const mockEnqueueCancelAtPeriodEndSync =
-  billingSubscriptionSyncMockFns.mockEnqueueCancelAtPeriodEndSync
-
 const ACTIVE_PERSONAL_PRO = {
   id: 'sub-personal',
   plan: 'pro_6000',
@@ -55,7 +49,7 @@ describe('pauseProSubscriptionForOrgCoverage', () => {
     resetDbChainMock()
   })
 
-  it('pauses the personal Pro and queues the Stripe sync when an entitled paid org covers the user', async () => {
+  it('pauses the personal Pro when an entitled paid org covers the user', async () => {
     queueWhereResponses([
       [{ organizationId: 'org-1' }],
       [{ plan: 'team_6000', referenceId: 'org-1' }],
@@ -73,12 +67,6 @@ describe('pauseProSubscriptionForOrgCoverage', () => {
       organizationId: 'org-1',
     })
     expect(dbChainMockFns.set).toHaveBeenCalledWith({ cancelAtPeriodEnd: true })
-    expect(mockEnqueueCancelAtPeriodEndSync).toHaveBeenCalledWith(expect.anything(), {
-      stripeSubscriptionId: 'stripe-sub-personal',
-      subscriptionId: 'sub-personal',
-      cancelAtPeriodEnd: true,
-      reason: 'covered-by-organization',
-    })
   })
 
   it('reports covered even when no entitled personal Pro row exists', async () => {
@@ -96,7 +84,6 @@ describe('pauseProSubscriptionForOrgCoverage', () => {
       organizationId: 'org-1',
     })
     expect(dbChainMockFns.update).not.toHaveBeenCalled()
-    expect(mockEnqueueCancelAtPeriodEndSync).not.toHaveBeenCalled()
   })
 
   it('reports covered without pausing again when the personal Pro is already pausing', async () => {
@@ -115,6 +102,5 @@ describe('pauseProSubscriptionForOrgCoverage', () => {
       organizationId: 'org-1',
     })
     expect(dbChainMockFns.update).not.toHaveBeenCalled()
-    expect(mockEnqueueCancelAtPeriodEndSync).not.toHaveBeenCalled()
   })
 })

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

@@ -1,10 +1,7 @@
 import { billingCoreMock, billingCoreMockFns } from '@sim/testing/mocks/billing-core.mock'
 import { billingOutboxHandlersMock } from '@sim/testing/mocks/billing-outbox-handlers.mock'
 import { billingPlanMock, billingPlanMockFns } from '@sim/testing/mocks/billing-plan.mock'
-import {
-  billingSubscriptionSyncMock,
-  billingSubscriptionSyncMockFns,
-} from '@sim/testing/mocks/billing-subscription-sync.mock'
+import { billingSubscriptionSyncMock } from '@sim/testing/mocks/billing-subscription-sync.mock'
 import { dbChainMockFns, resetDbChainMock } from '@sim/testing/mocks/database.mock'
 import {
   organizationMembershipMock,
@@ -55,8 +52,6 @@ const { mockAcquireOrganizationMutationLock } = organizationMembershipMockFns
 const mockGetOrganizationSubscription = billingCoreMockFns.mockGetOrganizationSubscription
 const mockGetHighestPriorityPersonalSubscription =
   billingPlanMockFns.mockGetHighestPriorityPersonalSubscription
-const { mockEnqueueSubscriptionSeatsSync, mockEnqueueCancelAtPeriodEndSync } =
-  billingSubscriptionSyncMockFns
 
 /** The subscription row as the activation re-reads it under its lock. */
 function testExecutor(onSubscriptionLock: () => void = () => {}) {
@@ -130,11 +125,6 @@ describe('ensureTeamOrganizationForAcceptance', () => {
       },
     })
     expect(updateCalls.value).toContainEqual(expect.objectContaining({ plan: 'team_6000' }))
-    // The Pro→Team price migration is durably enqueued at conversion time.
-    expect(mockEnqueueSubscriptionSeatsSync).toHaveBeenCalledWith(
-      executor,
-      expect.objectContaining({ subscriptionId: 'sub-pro', seats: 1 })
-    )
     expect(mockGetOrganizationSubscription).toHaveBeenCalledWith(
       'org-1',
       expect.objectContaining({ executor })
@@ -219,15 +209,8 @@ describe('ensureTeamOrganizationForAcceptance', () => {
       executor,
       expect.objectContaining({ plan: 'team_6000', referenceId: 'owner-1' })
     )
-    // The plan change enqueues the price seat-sync...
-    expect(mockEnqueueSubscriptionSeatsSync).toHaveBeenCalledWith(
-      executor,
-      expect.objectContaining({ subscriptionId: 'sub-pro', seats: 1 })
-    )
     expect(dbChainMockFns.transaction).not.toHaveBeenCalled()
     expect(lockOrder).toEqual(['organization', 'subscription'])
-    // ...but with no scheduled cancellation there is no cancel-sync event.
-    expect(mockEnqueueCancelAtPeriodEndSync).not.toHaveBeenCalled()
   })
 
   it('blocks personal Pro conversion when the reused organization has unresolved Enterprise', async () => {
@@ -255,8 +238,6 @@ describe('ensureTeamOrganizationForAcceptance', () => {
       })
     ).rejects.toThrow('Enterprise issuance is unfinished')
     expect(updateCalls.value).toHaveLength(0)
-    expect(mockEnqueueSubscriptionSeatsSync).not.toHaveBeenCalled()
-    expect(mockEnqueueCancelAtPeriodEndSync).not.toHaveBeenCalled()
   })
 
   it('provisions an org for a legacy personal-scoped Team subscription without a plan change', async () => {
@@ -283,9 +264,6 @@ describe('ensureTeamOrganizationForAcceptance', () => {
       expect.anything(),
       expect.objectContaining({ plan: 'team', referenceId: 'owner-1' })
     )
-    // No plan change and no scheduled cancellation: nothing to push to Stripe.
-    expect(mockEnqueueSubscriptionSeatsSync).not.toHaveBeenCalled()
-    expect(mockEnqueueCancelAtPeriodEndSync).not.toHaveBeenCalled()
   })
 
   it('returns upgrade-required (no downgrade) when no eligible Team tier exists', async () => {

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

@@ -8,10 +8,7 @@ import {
   setEnvFlags,
 } from '@sim/testing'
 import { billingOutboxHandlersMock } from '@sim/testing/mocks/billing-outbox-handlers.mock'
-import {
-  billingSubscriptionSyncMock,
-  billingSubscriptionSyncMockFns,
-} from '@sim/testing/mocks/billing-subscription-sync.mock'
+import { billingSubscriptionSyncMock } from '@sim/testing/mocks/billing-subscription-sync.mock'
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const { mockSyncSubscriptionUsageLimits } = vi.hoisted(() => ({
@@ -30,8 +27,6 @@ vi.mock('@sim/audit', () => auditMock)
 
 import { reconcileOrganizationSeats } from '@/lib/billing/organizations/seats'
 
-const enqueueMock = billingSubscriptionSyncMockFns.mockEnqueueSubscriptionSeatsSync
-
 const teamSub = {
   id: 'sub-1',
   plan: 'team_6000',
@@ -58,7 +53,7 @@ describe('reconcileOrganizationSeats', () => {
     resetDbChainMock()
   })
 
-  it('grows seats to the member count and enqueues a Stripe sync', async () => {
+  it('grows seats to the member count', async () => {
     queueReconcileReads([teamSub], [{ value: 2 }])
 
     const result = await reconcileOrganizationSeats({
@@ -74,11 +69,6 @@ describe('reconcileOrganizationSeats', () => {
       outboxEventId: 'subscription-seats-sync-event',
     })
     expect(dbChainMockFns.set).toHaveBeenCalledWith({ seats: 2 })
-    expect(enqueueMock).toHaveBeenCalledWith(expect.anything(), {
-      subscriptionId: 'sub-1',
-      seats: 2,
-      reason: 'member-accepted-invite',
-    })
     expect(mockSyncSubscriptionUsageLimits).toHaveBeenCalledWith(
       expect.objectContaining({ id: 'sub-1', referenceId: 'org-1', seats: 2 })
     )
@@ -94,7 +84,6 @@ describe('reconcileOrganizationSeats', () => {
 
     expect(result.changed).toBe(true)
     expect(dbChainMockFns.set).toHaveBeenCalledWith({ seats: 2 })
-    expect(enqueueMock).toHaveBeenCalledOnce()
   })
 
   it('still records the seat audit when the post-commit usage-limit sync fails', async () => {
@@ -129,7 +118,6 @@ describe('reconcileOrganizationSeats', () => {
     expect(result.changed).toBe(true)
     expect(result.seats).toBe(2)
     expect(dbChainMockFns.set).toHaveBeenCalledWith({ seats: 2 })
-    expect(enqueueMock).toHaveBeenCalled()
   })
 
   it('never drops below one seat', async () => {

apps/sim/lib/billing/webhooks/outbox-handlers.ts

@@ -14,6 +14,7 @@ import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
 import {
   type CancelAtPeriodEndSyncPayload,
   cancelAtPeriodEndSyncIdempotencyKey,
+  readRecordedSyncValue,
   type SubscriptionSeatsSyncPayload,
 } from '@/lib/billing/webhooks/subscription-sync'
 import type { OutboxHandler } from '@/lib/core/outbox/service'
@@ -100,20 +101,32 @@ async function getSubscriptionSeatSyncState(subscriptionId: string) {
   return row ?? null
 }
 
-async function readCancelAtPeriodEnd(subscriptionId: string): Promise<boolean | null> {
+/**
+ * The value this sync should push: the one recorded on its own event, re-read now, which every
+ * commit keeps current on each sync that can still run. The subscription row is not used for
+ * this, because the Stripe plugin can overwrite it with a stale webhook payload before the
+ * reconcile step restores it. An event enqueued before values were recorded falls back to the
+ * row. Null when the subscription no longer exists.
+ */
+async function readDesiredCancelAtPeriodEnd(
+  eventId: string,
+  subscriptionId: string
+): Promise<boolean | null> {
   const [row] = await db
     .select({ cancelAtPeriodEnd: subscriptionTable.cancelAtPeriodEnd })
     .from(subscriptionTable)
     .where(eq(subscriptionTable.id, subscriptionId))
     .limit(1)
-  return row ? Boolean(row.cancelAtPeriodEnd) : null
+  if (!row) return null
+  return (await readRecordedSyncValue(eventId))?.cancelAtPeriodEnd ?? Boolean(row.cancelAtPeriodEnd)
 }
 
 /**
- * Pushes the row's current value, never the payload's: racing events for one subscription each
- * converge on the last committed value. Stripe is read first and written only when it differs,
- * and the row is re-read after the write so a value committed while this event's request was in
- * flight is pushed too, even when an earlier event's request lands in Stripe after a newer one.
+ * Pushes the latest committed value (see `readDesiredCancelAtPeriodEnd`), never the claim-time
+ * payload: racing events for one subscription each converge on it. Stripe is read first and
+ * written only when it differs, and the value is re-read after the write so one committed while
+ * this event's request was in flight is pushed too, even when an earlier event's request lands
+ * in Stripe after a newer one.
  */
 const stripeSyncCancelAtPeriodEnd: OutboxHandler<CancelAtPeriodEndSyncPayload> = async (
   payload,
@@ -123,7 +136,7 @@ const stripeSyncCancelAtPeriodEnd: OutboxHandler<CancelAtPeriodEndSyncPayload> =
   const stripe = requireStripeClient()
 
   for (let attempt = 1; attempt <= MAX_SYNC_ATTEMPTS; attempt++) {
-    const desiredValue = await readCancelAtPeriodEnd(payload.subscriptionId)
+    const desiredValue = await readDesiredCancelAtPeriodEnd(ctx.eventId, payload.subscriptionId)
     if (desiredValue === null) {
       logger.warn('Subscription not found when syncing cancel_at_period_end', {
         eventId: ctx.eventId,
@@ -142,7 +155,7 @@ const stripeSyncCancelAtPeriodEnd: OutboxHandler<CancelAtPeriodEndSyncPayload> =
       )
     }
 
-    const latestValue = await readCancelAtPeriodEnd(payload.subscriptionId)
+    const latestValue = await readDesiredCancelAtPeriodEnd(ctx.eventId, payload.subscriptionId)
     if (latestValue !== desiredValue) {
       logger.info('cancel_at_period_end changed during Stripe sync; retrying latest value', {
         eventId: ctx.eventId,
@@ -189,6 +202,10 @@ const stripeCancelSubscriptionImmediately: OutboxHandler<
   })
 }
 
+/**
+ * Pushes the seat count recorded on its own event, re-read each pass (falling back to the row
+ * for an event enqueued before values were recorded), for the same reason as the cancel sync.
+ */
 const stripeSyncSubscriptionSeats: OutboxHandler<SubscriptionSeatsSyncPayload> = async (
   payload,
   ctx
@@ -231,7 +248,7 @@ const stripeSyncSubscriptionSeats: OutboxHandler<SubscriptionSeatsSyncPayload> =
       return
     }
 
-    const desiredSeats = row.seats || 1
+    const desiredSeats = (await readRecordedSyncValue(ctx.eventId))?.seats ?? (row.seats || 1)
     const stripeSubscription = await stripe.subscriptions.retrieve(row.stripeSubscriptionId)
 
     if (!hasPaidSubscriptionStatus(stripeSubscription.status)) {
@@ -281,7 +298,7 @@ const stripeSyncSubscriptionSeats: OutboxHandler<SubscriptionSeatsSyncPayload> =
     }
 
     const latest = await getSubscriptionSeatSyncState(payload.subscriptionId)
-    const latestSeats = latest?.seats || 1
+    const latestSeats = (await readRecordedSyncValue(ctx.eventId))?.seats ?? (latest?.seats || 1)
     if (latestSeats !== desiredSeats) {
       logger.info('Subscription seats changed during Stripe sync; retrying latest value', {
         eventId: ctx.eventId,