fix(billing): converge Stripe cancel/contact syncs and guard webhook echoes over unsynced DB changes

Author: waleedlatif1

apps/sim/app/api/v1/admin/subscriptions/[id]/route.ts

@@ -33,8 +33,7 @@ import {
 } from '@/lib/api/contracts/v1/admin'
 import { parseRequest } from '@/lib/api/server'
 import { requireStripeClient } from '@/lib/billing/stripe-client'
-import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
-import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
+import { enqueueCancelAtPeriodEndSync } from '@/lib/billing/webhooks/subscription-sync'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { withAdminAuthParams } from '@/app/api/v1/admin/middleware'
 import {
@@ -113,15 +112,17 @@ export const DELETE = withRouteHandler(
       }
 
       if (atPeriodEnd) {
+        const stripeSubscriptionId = existing.stripeSubscriptionId
         await db.transaction(async (tx) => {
           await tx
             .update(subscription)
             .set({ cancelAtPeriodEnd: true })
             .where(eq(subscription.id, subscriptionId))
 
-          await enqueueOutboxEvent(tx, OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END, {
-            stripeSubscriptionId: existing.stripeSubscriptionId,
+          await enqueueCancelAtPeriodEndSync(tx, {
+            stripeSubscriptionId,
             subscriptionId: existing.id,
+            cancelAtPeriodEnd: true,
             reason: reason ?? 'admin-cancel-at-period-end',
           })
         })

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

@@ -2,6 +2,10 @@ 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 { 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'
@@ -24,6 +28,7 @@ vi.mock('@/lib/billing/organizations/membership', () => organizationMembershipMo
 vi.mock('@/lib/billing/stripe-client', () => stripeClientMock)
 vi.mock('@/lib/billing/webhooks/outbox-handlers', () => billingOutboxHandlersMock)
 vi.mock('@/lib/core/outbox/service', () => outboxServiceMock)
+vi.mock('@/lib/billing/webhooks/subscription-sync', () => billingSubscriptionSyncMock)
 
 import {
   refundDashboardSubscriptionPayment,
@@ -119,6 +124,11 @@ describe('admin subscription cancellation', () => {
       expect.objectContaining({ status: 'pending', attempts: 0, lastError: null })
     )
     expect(dbChainMockFns.set).toHaveBeenCalledWith({ cancelAtPeriodEnd: true })
+    expect(billingSubscriptionSyncMockFns.mockRecommitCancelAtPeriodEndSync).toHaveBeenCalledWith(
+      expect.anything(),
+      'outbox-1',
+      true
+    )
     expect(result).toMatchObject({
       operationId: '67e55044-10b1-426f-9247-bb680e5fe0c8',
       status: 'pending',

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

@@ -8,6 +8,10 @@ import { acquireOrganizationMutationLock } from '@/lib/billing/organizations/mem
 import { requireStripeClient } from '@/lib/billing/stripe-client'
 import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/utils'
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
+import {
+  enqueueCancelAtPeriodEndSync,
+  recommitCancelAtPeriodEndSync,
+} from '@/lib/billing/webhooks/subscription-sync'
 import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
 
 const RECENT_INVOICE_LIMIT = 12
@@ -301,6 +305,7 @@ export async function requestDashboardSubscriptionCancellation({
           if (!restoredSubscription) {
             throw new Error('Cancellation subscription no longer exists')
           }
+          await recommitCancelAtPeriodEndSync(tx, existingOperation.id, true)
         }
         await tx
           .update(outboxEvent)
@@ -384,18 +389,15 @@ export async function requestDashboardSubscriptionCancellation({
         .set({ cancelAtPeriodEnd: true })
         .where(eq(subscription.id, subscriptionRow.id))
     }
-    const eventId = await enqueueOutboxEvent(
-      tx,
-      OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
-      {
-        operationId,
-        organizationId,
-        subscriptionId: subscriptionRow.id,
-        stripeSubscriptionId: subscriptionRow.stripeSubscriptionId,
-        reason: normalizedReason,
-        requestedBy: actor,
-      }
-    )
+    const eventId = await enqueueCancelAtPeriodEndSync(tx, {
+      operationId,
+      organizationId,
+      subscriptionId: subscriptionRow.id,
+      stripeSubscriptionId: subscriptionRow.stripeSubscriptionId,
+      cancelAtPeriodEnd: true,
+      reason: normalizedReason,
+      requestedBy: actor,
+    })
     return {
       operationId,
       outboxEventId: eventId,

apps/sim/lib/auth/auth.ts

@@ -103,6 +103,7 @@ import {
   handleSubscriptionCreated,
   handleSubscriptionDeleted,
 } from '@/lib/billing/webhooks/subscription'
+import { reconcileSubscriptionSyncFromStripe } from '@/lib/billing/webhooks/subscription-sync'
 import { handleSubscriptionUsageUpdate } from '@/lib/billing/webhooks/subscription-usage'
 import { env } from '@/lib/core/config/env'
 import {
@@ -1727,6 +1728,7 @@ export const auth = betterAuth({
                   case 'customer.subscription.created':
                   case 'customer.subscription.updated': {
                     await handleManualEnterpriseSubscription(event)
+                    await reconcileSubscriptionSyncFromStripe(event)
                     await handleSubscriptionUsageUpdate(event)
                     break
                   }

apps/sim/lib/billing/organizations/lock-order.test.ts

@@ -17,6 +17,7 @@ import {
   workspace,
 } from '@sim/db/schema'
 import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
+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'
 
@@ -42,6 +43,7 @@ import type { DbOrTx } from '@/lib/db/types'
 import { attachOwnedWorkspacesToOrganizationTx } from '@/lib/workspaces/organization-workspaces'
 
 vi.mock('@/lib/core/outbox/service', () => outboxServiceMock)
+vi.mock('@/lib/billing/webhooks/subscription-sync', () => billingSubscriptionSyncMock)
 
 /**
  * A superset row that satisfies every read in the join path: a paid org sub, a

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

@@ -48,6 +48,7 @@ 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 { isBillingEnabled } from '@/lib/core/config/env-flags'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
@@ -288,9 +289,10 @@ export async function restoreUserProSubscription(userId: string): Promise<Restor
       .set({ cancelAtPeriodEnd: false })
       .where(eq(subscriptionTable.id, personalPro.id))
 
-    await enqueueOutboxEvent(tx, OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END, {
+    await enqueueCancelAtPeriodEndSync(tx, {
       stripeSubscriptionId: personalPro.stripeSubscriptionId,
       subscriptionId: personalPro.id,
+      cancelAtPeriodEnd: false,
       reason: 'member-left-paid-org',
     })
 
@@ -411,9 +413,10 @@ export async function pauseProSubscriptionForOrgCoverage(
       .where(eq(subscriptionTable.id, personalPro.id))
 
     if (personalPro.stripeSubscriptionId) {
-      await enqueueOutboxEvent(tx, OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END, {
+      await enqueueCancelAtPeriodEndSync(tx, {
         stripeSubscriptionId: personalPro.stripeSubscriptionId,
         subscriptionId: personalPro.id,
+        cancelAtPeriodEnd: true,
         reason: 'covered-by-organization',
       })
     }
@@ -859,9 +862,10 @@ async function applyPaidOrgJoinBillingTx(
       .where(eq(subscriptionTable.id, personalPro.id))
 
     if (personalPro.stripeSubscriptionId) {
-      await enqueueOutboxEvent(tx, OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END, {
+      await enqueueCancelAtPeriodEndSync(tx, {
         stripeSubscriptionId: personalPro.stripeSubscriptionId,
         subscriptionId: personalPro.id,
+        cancelAtPeriodEnd: true,
         reason: 'joined-paid-org',
         ...(options.sourceOperationId ? { sourceOperationId: options.sourceOperationId } : {}),
       })

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

@@ -1,5 +1,9 @@
 import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
-import { outboxServiceMock, outboxServiceMockFns } from '@sim/testing/mocks/outbox-service.mock'
+import {
+  billingSubscriptionSyncMock,
+  billingSubscriptionSyncMockFns,
+} 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'
 
 vi.mock('@/lib/billing/storage/payer-transfer', () => ({
@@ -8,11 +12,12 @@ vi.mock('@/lib/billing/storage/payer-transfer', () => ({
   changeWorkspaceStoragePayersInTx: vi.fn(),
 }))
 vi.mock('@/lib/core/outbox/service', () => outboxServiceMock)
+vi.mock('@/lib/billing/webhooks/subscription-sync', () => billingSubscriptionSyncMock)
 
 import { pauseProSubscriptionForOrgCoverage } from '@/lib/billing/organizations/membership'
-import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
 
-const mockEnqueueOutboxEvent = outboxServiceMockFns.mockEnqueueOutboxEvent
+const mockEnqueueCancelAtPeriodEndSync =
+  billingSubscriptionSyncMockFns.mockEnqueueCancelAtPeriodEndSync
 
 const ACTIVE_PERSONAL_PRO = {
   id: 'sub-personal',
@@ -68,15 +73,12 @@ describe('pauseProSubscriptionForOrgCoverage', () => {
       organizationId: 'org-1',
     })
     expect(dbChainMockFns.set).toHaveBeenCalledWith({ cancelAtPeriodEnd: true })
-    expect(mockEnqueueOutboxEvent).toHaveBeenCalledWith(
-      expect.anything(),
-      OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
-      {
-        stripeSubscriptionId: 'stripe-sub-personal',
-        subscriptionId: 'sub-personal',
-        reason: 'covered-by-organization',
-      }
-    )
+    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 () => {
@@ -94,7 +96,7 @@ describe('pauseProSubscriptionForOrgCoverage', () => {
       organizationId: 'org-1',
     })
     expect(dbChainMockFns.update).not.toHaveBeenCalled()
-    expect(mockEnqueueOutboxEvent).not.toHaveBeenCalled()
+    expect(mockEnqueueCancelAtPeriodEndSync).not.toHaveBeenCalled()
   })
 
   it('reports covered without pausing again when the personal Pro is already pausing', async () => {
@@ -113,6 +115,6 @@ describe('pauseProSubscriptionForOrgCoverage', () => {
       organizationId: 'org-1',
     })
     expect(dbChainMockFns.update).not.toHaveBeenCalled()
-    expect(mockEnqueueOutboxEvent).not.toHaveBeenCalled()
+    expect(mockEnqueueCancelAtPeriodEndSync).not.toHaveBeenCalled()
   })
 })

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

@@ -1,12 +1,16 @@
 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 { dbChainMockFns, resetDbChainMock } from '@sim/testing/mocks/database.mock'
 import {
   organizationMembershipMock,
   organizationMembershipMockFns,
 } from '@sim/testing/mocks/organization-membership.mock'
-import { outboxServiceMock, outboxServiceMockFns } from '@sim/testing/mocks/outbox-service.mock'
+import { outboxServiceMock } from '@sim/testing/mocks/outbox-service.mock'
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const {
@@ -41,6 +45,8 @@ vi.mock('@/lib/billing/plans', () => ({
 
 vi.mock('@/lib/core/outbox/service', () => outboxServiceMock)
 
+vi.mock('@/lib/billing/webhooks/subscription-sync', () => billingSubscriptionSyncMock)
+
 vi.mock('@/lib/billing/webhooks/outbox-handlers', () => billingOutboxHandlersMock)
 
 import { ensureTeamOrganizationForAcceptance } from '@/lib/billing/organizations/provision-seat'
@@ -49,15 +55,21 @@ const { mockAcquireOrganizationMutationLock } = organizationMembershipMockFns
 const mockGetOrganizationSubscription = billingCoreMockFns.mockGetOrganizationSubscription
 const mockGetHighestPriorityPersonalSubscription =
   billingPlanMockFns.mockGetHighestPriorityPersonalSubscription
-const enqueueMock = outboxServiceMockFns.mockEnqueueOutboxEvent
+const { mockEnqueueSubscriptionSeatsSync, mockEnqueueCancelAtPeriodEndSync } =
+  billingSubscriptionSyncMockFns
 
 function testExecutor(onUpdate: () => void = () => {}) {
   return {
     update: () => ({
       set: (values: Record<string, unknown>) => {
         onUpdate()
         updateCalls.value.push(values)
-        return { where: () => Promise.resolve([]) }
+        return {
+          where: () =>
+            Object.assign(Promise.resolve([]), {
+              returning: () => Promise.resolve([{ seats: 1 }]),
+            }),
+        }
       },
     }),
   } as never
@@ -113,10 +125,9 @@ describe('ensureTeamOrganizationForAcceptance', () => {
     })
     expect(updateCalls.value).toContainEqual(expect.objectContaining({ plan: 'team_6000' }))
     // The Pro→Team price migration is durably enqueued at conversion time.
-    expect(enqueueMock).toHaveBeenCalledWith(
+    expect(mockEnqueueSubscriptionSeatsSync).toHaveBeenCalledWith(
       executor,
-      'stripe.sync-subscription-seats',
-      expect.objectContaining({ subscriptionId: 'sub-pro' })
+      expect.objectContaining({ subscriptionId: 'sub-pro', seats: 1 })
     )
     expect(mockGetOrganizationSubscription).toHaveBeenCalledWith(
       'org-1',
@@ -203,19 +214,14 @@ describe('ensureTeamOrganizationForAcceptance', () => {
       expect.objectContaining({ plan: 'team_6000', referenceId: 'owner-1' })
     )
     // The plan change enqueues the price seat-sync...
-    expect(enqueueMock).toHaveBeenCalledWith(
+    expect(mockEnqueueSubscriptionSeatsSync).toHaveBeenCalledWith(
       executor,
-      'stripe.sync-subscription-seats',
-      expect.objectContaining({ subscriptionId: 'sub-pro' })
+      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(enqueueMock).not.toHaveBeenCalledWith(
-      expect.anything(),
-      'stripe.sync-cancel-at-period-end',
-      expect.anything()
-    )
+    expect(mockEnqueueCancelAtPeriodEndSync).not.toHaveBeenCalled()
   })
 
   it('blocks personal Pro conversion when the reused organization has unresolved Enterprise', async () => {
@@ -243,7 +249,8 @@ describe('ensureTeamOrganizationForAcceptance', () => {
       })
     ).rejects.toThrow('Enterprise issuance is unfinished')
     expect(updateCalls.value).toHaveLength(0)
-    expect(enqueueMock).not.toHaveBeenCalled()
+    expect(mockEnqueueSubscriptionSeatsSync).not.toHaveBeenCalled()
+    expect(mockEnqueueCancelAtPeriodEndSync).not.toHaveBeenCalled()
   })
 
   it('provisions an org for a legacy personal-scoped Team subscription without a plan change', async () => {
@@ -271,7 +278,8 @@ describe('ensureTeamOrganizationForAcceptance', () => {
       expect.objectContaining({ plan: 'team', referenceId: 'owner-1' })
     )
     // No plan change and no scheduled cancellation: nothing to push to Stripe.
-    expect(enqueueMock).not.toHaveBeenCalled()
+    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/provision-seat.ts

@@ -16,8 +16,10 @@ import {
 } from '@/lib/billing/plan-helpers'
 import { getPlanByName } from '@/lib/billing/plans'
 import { hasUsableSubscriptionStatus } from '@/lib/billing/subscriptions/utils'
-import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
-import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
+import {
+  enqueueCancelAtPeriodEndSync,
+  enqueueSubscriptionSeatsSync,
+} from '@/lib/billing/webhooks/subscription-sync'
 import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 
 const logger = createLogger('ProvisionSeat')
@@ -248,22 +250,25 @@ async function activateTeamSubscription(
     Boolean(sub.cancelAtPeriodEnd) && Boolean(sub.stripeSubscriptionId)
 
   const apply = async (tx: DbOrTx) => {
-    await tx
+    const [activated] = await tx
       .update(subscriptionTable)
       .set({ plan: targetPlan, cancelAtPeriodEnd: false })
       .where(eq(subscriptionTable.id, sub.id))
+      .returning({ seats: subscriptionTable.seats })
 
     if (planChanged) {
-      await enqueueOutboxEvent(tx, OUTBOX_EVENT_TYPES.STRIPE_SYNC_SUBSCRIPTION_SEATS, {
+      await enqueueSubscriptionSeatsSync(tx, {
         subscriptionId: sub.id,
+        seats: activated?.seats ?? 1,
         reason: 'pro-to-team-conversion',
       })
     }
 
     if (shouldClearCancellation) {
-      await enqueueOutboxEvent(tx, OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END, {
+      await enqueueCancelAtPeriodEndSync(tx, {
         stripeSubscriptionId: sub.stripeSubscriptionId as string,
         subscriptionId: sub.id,
+        cancelAtPeriodEnd: false,
         reason: 'pro-to-team-conversion',
       })
     }