fix(billing): close remaining stale-intent paths (enterprise retry, legacy echoes, customer restore)

Author: waleedlatif1

apps/sim/lib/auth/auth.ts

@@ -103,7 +103,10 @@ import {
   handleSubscriptionCreated,
   handleSubscriptionDeleted,
 } from '@/lib/billing/webhooks/subscription'
-import { reconcileSubscriptionSyncFromStripe } from '@/lib/billing/webhooks/subscription-sync'
+import {
+  commitCustomerRestoredSubscription,
+  reconcileSubscriptionSyncFromStripe,
+} from '@/lib/billing/webhooks/subscription-sync'
 import { handleSubscriptionUsageUpdate } from '@/lib/billing/webhooks/subscription-usage'
 import { env } from '@/lib/core/config/env'
 import {
@@ -1101,6 +1104,16 @@ 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 && ctx.path === '/subscription/upgrade') {
         const checkoutContext = ctx as typeof ctx & {
           billingCheckoutAdmissionClaim?: CheckoutAdmissionClaim

apps/sim/lib/billing/enterprise-provisioning.ts

@@ -77,6 +77,7 @@ import { TERMINAL_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/util
 import { countPendingSeatInvitations } from '@/lib/billing/validation/seat-management'
 import { withEnterpriseReconciliationLease } from '@/lib/billing/webhooks/enterprise-reconciliation-lease'
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
+import { recommitSubscriptionSync } from '@/lib/billing/webhooks/subscription-sync'
 import { env } from '@/lib/core/config/env'
 import {
   continueOutboxHandler,
@@ -2135,6 +2136,13 @@ export async function retryEnterpriseFollowUpJob(
         processedAt: null,
       })
       .where(eq(outboxEvent.id, jobEventId))
+    if (detail.kind === 'personal_subscription_cancellation') {
+      await recommitSubscriptionSync(
+        tx,
+        OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
+        detail.subjectId
+      )
+    }
     return true
   })
 

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

@@ -53,7 +53,10 @@ import { isTeam } from '@/lib/billing/plan-helpers'
 import { syncSeatsFromStripeQuantity } from '@/lib/billing/validation/seat-management'
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
 import { billingOutboxHandlers } from '@/lib/billing/webhooks/outbox-handlers'
-import { reconcileSubscriptionSyncFromStripe } from '@/lib/billing/webhooks/subscription-sync'
+import {
+  commitCustomerRestoredSubscription,
+  reconcileSubscriptionSyncFromStripe,
+} from '@/lib/billing/webhooks/subscription-sync'
 import { enqueueOutboxEvent, processOutboxEventById } from '@/lib/core/outbox/service'
 import { POST as requeueOutboxEvent } from '@/app/api/v1/admin/outbox/[id]/requeue/route'
 
@@ -519,6 +522,89 @@ describe('cancel_at_period_end sync', () => {
 
     expect((await storedSubscription(pro.subscriptionId)).cancelAtPeriodEnd).toBe(true)
   })
+
+  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)
+    const pauseSync = await latestOutboxEventId(
+      OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
+      pro.subscriptionId
+    )
+    await deadLetter(pauseSync)
+    await leaveOrganization(pro.userId, pro.paidOrganization.organizationId)
+    await restoreUserProSubscription(pro.userId)
+    const restoreSync = await latestOutboxEventId(
+      OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
+      pro.subscriptionId
+    )
+    await expect(processEvent(restoreSync)).resolves.toBe('completed')
+
+    await testDatabase
+      .update(outboxEvent)
+      .set({
+        status: 'pending',
+        attempts: 0,
+        lastError: null,
+        availableAt: new Date(),
+        lockedAt: null,
+        processedAt: null,
+      })
+      .where(eq(outboxEvent.id, pauseSync))
+    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("treats a write from an older deploy's sync handler as Sim's own echo", async () => {
+    const pro = await createProUserInPaidOrganization()
+    await pauseProSubscriptionForOrgCoverage(pro.userId)
+    const pauseSync = await latestOutboxEventId(
+      OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
+      pro.subscriptionId
+    )
+    await leaveOrganization(pro.userId, pro.paidOrganization.organizationId)
+    await restoreUserProSubscription(pro.userId)
+
+    await stripe.client.subscriptions.update(
+      pro.stripeSubscriptionId,
+      { cancel_at_period_end: true },
+      { idempotencyKey: `outbox:${pauseSync}` }
+    )
+    await deliver(stripe.events.at(-1) as Stripe.Event)
+
+    expect((await storedSubscription(pro.subscriptionId)).cancelAtPeriodEnd).toBe(false)
+  })
+
+  it("keeps a customer's restore made while an earlier sync is retrying", 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')
+
+    const restored = await stripe.client.subscriptions.update(pro.stripeSubscriptionId, {
+      cancel_at_period_end: false,
+    })
+    const restoreEvent = stripe.events.at(-1) as Stripe.Event
+    await testDatabase
+      .update(subscription)
+      .set({ cancelAtPeriodEnd: false, cancelAt: null, canceledAt: null })
+      .where(eq(subscription.id, pro.subscriptionId))
+    await commitCustomerRestoredSubscription(restored)
+
+    await deliverUnrelatedUpdate(pro.stripeSubscriptionId)
+    await makeDue(pauseSync)
+    await expect(processEvent(pauseSync)).resolves.toBe('completed')
+    await deliver(restoreEvent)
+
+    expect((await storedSubscription(pro.subscriptionId)).cancelAtPeriodEnd).toBe(false)
+    expect(stripe.subscription(pro.stripeSubscriptionId).cancel_at_period_end).toBe(false)
+  })
 })
 
 describe('customer contact sync', () => {

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

@@ -10,6 +10,7 @@ import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
 import {
   enqueueOutboxEvent,
   listInflightOutboxEvents,
+  maxSettledOutboxPayloadNumber,
   patchInflightOutboxEvents,
 } from '@/lib/core/outbox/service'
 import type { DbOrTx } from '@/lib/db/types'
@@ -22,6 +23,8 @@ const logger = createLogger('BillingSubscriptionSync')
  * webhook is recognized as the echo of Sim's own sync rather than a change made in Stripe.
  */
 const CANCEL_AT_PERIOD_END_SYNC_KEY_PREFIX = 'outbox-sync-cancel-at-period-end:'
+/** The key prefix the cancel-sync handler used before the one above; old pods still send it mid-rollout. */
+const LEGACY_SYNC_KEY_PREFIX = 'outbox:'
 
 const CANCEL_SYNC = OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END
 const SEATS_SYNC = OUTBOX_EVENT_TYPES.STRIPE_SYNC_SUBSCRIPTION_SEATS
@@ -161,20 +164,62 @@ export async function recommitSubscriptionSync(
   )
 }
 
+/**
+ * Records a customer's restore through Better Auth's `/subscription/restore` as Sim's latest
+ * committed `cancelAtPeriodEnd`. That endpoint updates Stripe and then writes the row directly,
+ * so a cancel sync still in flight would otherwise carry the cancellation the customer just
+ * undid, and a webhook processed before the restore's own could restore it for that sync to
+ * push. Enqueuing re-pushes `false` even if such a sync already ran. Called from the endpoint's
+ * `after` hook with the Stripe subscription it returned.
+ */
+export async function commitCustomerRestoredSubscription(restored: unknown): Promise<void> {
+  const stripeSubscription = toRecord(restored)
+  const stripeSubscriptionId = stripeSubscription.id
+  if (
+    typeof stripeSubscriptionId !== 'string' ||
+    stripeSubscription.cancel_at_period_end !== false
+  ) {
+    return
+  }
+
+  await db.transaction(async (tx) => {
+    const [row] = await tx
+      .select({ id: subscription.id })
+      .from(subscription)
+      .where(eq(subscription.stripeSubscriptionId, stripeSubscriptionId))
+      .for('update')
+      .limit(1)
+    if (!row) return
+
+    await tx
+      .update(subscription)
+      .set({ cancelAtPeriodEnd: false })
+      .where(eq(subscription.id, row.id))
+    await enqueueCancelAtPeriodEndSync(tx, {
+      stripeSubscriptionId,
+      subscriptionId: row.id,
+      cancelAtPeriodEnd: false,
+      reason: 'customer-restored',
+    })
+  })
+}
+
 /** 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()}`
 }
 
 /**
- * What a sync type's in-flight events say about its field: nothing in flight, the latest
+ * What a sync type's in-flight events say about its field: no applicable value, the latest
  * committed value, or `legacy` when an event predates recorded values (enqueued by an older
- * deploy) so the committed value is unknown.
+ * deploy) so the committed value is unknown. An in-flight value older than one a settled event
+ * already carried was revived by a retry path that did not re-commit, and does not apply.
  */
 type InflightIntent<T> = { status: 'none' } | { status: 'legacy' } | { status: 'value'; value: T }
 
 function latestIntent<T>(
   events: { eventType: string; payload: unknown }[],
+  settledCommittedAt: Map<string, number>,
   eventType: SubscriptionSyncEventType,
   readValue: (payload: Record<string, unknown>) => T | undefined
 ): InflightIntent<T> {
@@ -188,20 +233,27 @@ function latestIntent<T>(
       latest = { committedAt: payload.committedAt, value }
     }
   }
-  return latest ? { status: 'value', value: latest.value } : { status: 'none' }
+  if (!latest || latest.committedAt < (settledCommittedAt.get(eventType) ?? 0)) {
+    return { status: 'none' }
+  }
+  return { status: 'value', value: latest.value }
 }
 
 async function readInflightIntents(executor: DbOrTx, subscriptionId: string) {
-  const events = await listInflightOutboxEvents(
+  const eventTypes = [CANCEL_SYNC, SEATS_SYNC]
+  const subject = subscriptionSubject(subscriptionId)
+  const events = await listInflightOutboxEvents(executor, eventTypes, subject)
+  const settledCommittedAt = await maxSettledOutboxPayloadNumber(
     executor,
-    [CANCEL_SYNC, SEATS_SYNC],
-    subscriptionSubject(subscriptionId)
+    eventTypes,
+    subject,
+    'committedAt'
   )
   return {
-    cancelAtPeriodEnd: latestIntent(events, CANCEL_SYNC, (payload) =>
+    cancelAtPeriodEnd: latestIntent(events, settledCommittedAt, CANCEL_SYNC, (payload) =>
       typeof payload.cancelAtPeriodEnd === 'boolean' ? payload.cancelAtPeriodEnd : undefined
     ),
-    seats: latestIntent(events, SEATS_SYNC, (payload) =>
+    seats: latestIntent(events, settledCommittedAt, SEATS_SYNC, (payload) =>
       typeof payload.seats === 'number' ? payload.seats : undefined
     ),
   }
@@ -212,7 +264,10 @@ function isCancellationChangedInStripe(event: Stripe.Event): boolean {
   const previousAttributes = toRecord(event.data.previous_attributes)
   if (!('cancel_at_period_end' in previousAttributes)) return false
   const idempotencyKey = event.request?.idempotency_key
-  return !idempotencyKey?.startsWith(CANCEL_AT_PERIOD_END_SYNC_KEY_PREFIX)
+  const issuedBySimSync =
+    idempotencyKey?.startsWith(CANCEL_AT_PERIOD_END_SYNC_KEY_PREFIX) ||
+    idempotencyKey?.startsWith(LEGACY_SYNC_KEY_PREFIX)
+  return !issuedBySimSync
 }
 
 type CancelAtPeriodEndSource =

apps/sim/lib/core/outbox/service.ts

@@ -4,7 +4,7 @@ import { createLogger } from '@sim/logger'
 import { describeError, toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { truncate } from '@sim/utils/string'
-import { and, asc, desc, eq, inArray, lte, sql } from 'drizzle-orm'
+import { and, asc, desc, eq, inArray, lte, notInArray, sql } from 'drizzle-orm'
 import {
   dueOutboxWorkQuery,
   isStuckProcessing,
@@ -440,6 +440,35 @@ export async function listInflightOutboxEvents(
   return limit === undefined ? query : query.limit(limit)
 }
 
+/**
+ * For each event type, the largest numeric `payloadKey` among the subject's settled
+ * (`completed` or `dead_letter`) events. Types with no such value are absent from the map.
+ */
+export async function maxSettledOutboxPayloadNumber(
+  executor: Pick<typeof db, 'select'>,
+  eventTypes: readonly string[],
+  subject: OutboxPayloadSubject,
+  payloadKey: string
+): Promise<Map<string, number>> {
+  const rows = await executor
+    .select({
+      eventType: outboxEvent.eventType,
+      value: sql<string | null>`max((${outboxEvent.payload} ->> ${payloadKey})::numeric)::text`,
+    })
+    .from(outboxEvent)
+    .where(
+      and(
+        inArray(outboxEvent.eventType, [...eventTypes]),
+        notInArray(outboxEvent.status, [...INFLIGHT_OUTBOX_STATUSES]),
+        sql`${outboxEvent.payload} ->> ${subject.payloadKey} = ${subject.payloadValue}`
+      )
+    )
+    .groupBy(outboxEvent.eventType)
+  return new Map(
+    rows.flatMap((row) => (row.value === null ? [] : [[row.eventType, Number(row.value)]]))
+  )
+}
+
 /**
  * Shallow-merges `patch` into the payload of every `pending` or `processing` event of the type
  * for one subject. Callers serialize writers for the subject with their domain lock.

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

@@ -28,6 +28,7 @@ export const billingSubscriptionSyncMockFns = {
     (eventId: string) => `outbox-sync-cancel-at-period-end:${eventId}:key`
   ),
   mockReconcileSubscriptionSyncFromStripe: vi.fn(async () => undefined),
+  mockCommitCustomerRestoredSubscription: vi.fn(async () => undefined),
 }
 
 /**
@@ -47,4 +48,6 @@ export const billingSubscriptionSyncMock = {
     billingSubscriptionSyncMockFns.mockCancelAtPeriodEndSyncIdempotencyKey,
   reconcileSubscriptionSyncFromStripe:
     billingSubscriptionSyncMockFns.mockReconcileSubscriptionSyncFromStripe,
+  commitCustomerRestoredSubscription:
+    billingSubscriptionSyncMockFns.mockCommitCustomerRestoredSubscription,
 }

packages/testing/src/mocks/outbox-service.mock.ts

@@ -70,6 +70,7 @@ export const outboxServiceMockFns = {
   mockFindDeadLetteredEvents: vi.fn(),
   mockListInflightOutboxEvents: vi.fn(),
   mockPatchInflightOutboxEvents: vi.fn(),
+  mockMaxSettledOutboxPayloadNumber: vi.fn(),
   mockHasInflightOutboxEvent: vi.fn(),
   mockHasDueOutboxWork: vi.fn(),
   mockProcessOutboxEvents: vi.fn(),
@@ -101,6 +102,7 @@ export const outboxServiceMock = {
   findDeadLetteredEvents: outboxServiceMockFns.mockFindDeadLetteredEvents,
   listInflightOutboxEvents: outboxServiceMockFns.mockListInflightOutboxEvents,
   patchInflightOutboxEvents: outboxServiceMockFns.mockPatchInflightOutboxEvents,
+  maxSettledOutboxPayloadNumber: outboxServiceMockFns.mockMaxSettledOutboxPayloadNumber,
   hasInflightOutboxEvent: outboxServiceMockFns.mockHasInflightOutboxEvent,
   hasDueOutboxWork: outboxServiceMockFns.mockHasDueOutboxWork,
   processOutboxEvents: outboxServiceMockFns.mockProcessOutboxEvents,