fix(billing): record each committed sync value on every in-flight event so no stale intent can resurface

Author: waleedlatif1

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

@@ -2,6 +2,7 @@ import { db } from '@sim/db'
 import { outboxEvent } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
+import { toRecord } from '@sim/utils/object'
 import { and, eq, sql } from 'drizzle-orm'
 import { NextResponse } from 'next/server'
 import { adminV1RequeueOutboxEventContract } from '@/lib/api/contracts/v1/admin'
@@ -11,6 +12,10 @@ import {
   ENTERPRISE_METADATA_SYNC_EVENT_TYPE,
   ENTERPRISE_PROVISION_EVENT_TYPE,
 } from '@/lib/billing/enterprise-outbox-events'
+import {
+  isSubscriptionSyncEventType,
+  recommitSubscriptionSync,
+} from '@/lib/billing/webhooks/subscription-sync'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { withAdminAuthParams } from '@/app/api/v1/admin/middleware'
 
@@ -28,7 +33,9 @@ const invalidOutboxEventResponse = (message: string) =>
  * will retry it. Resets `attempts`, `lastError`, and `availableAt` so
  * the next poll picks it up. Only dead-lettered events can be
  * requeued — completed/pending/processing rows are rejected to avoid
- * operator errors.
+ * operator errors. A Stripe subscription sync is re-committed with the
+ * subscription's current value, so the retry never revives the value it
+ * failed with.
  */
 export const POST = withRouteHandler(
   withAdminAuthParams<{ id: string }>(async (request, context) => {
@@ -61,23 +68,34 @@ export const POST = withRouteHandler(
       const deliveryRevision = metadataIntent?.success
         ? metadataIntent.data.deliveryRevision + 1
         : null
-      const result = await db
-        .update(outboxEvent)
-        .set({
-          status: 'pending',
-          attempts: 0,
-          lastError: null,
-          availableAt: new Date(),
-          lockedAt: null,
-          processedAt: null,
-          ...(deliveryRevision === null
-            ? {}
-            : {
-                payload: sql`(${outboxEvent.payload}::jsonb || ${JSON.stringify({ deliveryRevision })}::jsonb)::json`,
-              }),
-        })
-        .where(and(eq(outboxEvent.id, id), eq(outboxEvent.status, 'dead_letter')))
-        .returning({ id: outboxEvent.id, eventType: outboxEvent.eventType })
+      const result = await db.transaction(async (tx) => {
+        const requeued = await tx
+          .update(outboxEvent)
+          .set({
+            status: 'pending',
+            attempts: 0,
+            lastError: null,
+            availableAt: new Date(),
+            lockedAt: null,
+            processedAt: null,
+            ...(deliveryRevision === null
+              ? {}
+              : {
+                  payload: sql`(${outboxEvent.payload}::jsonb || ${JSON.stringify({ deliveryRevision })}::jsonb)::json`,
+                }),
+          })
+          .where(and(eq(outboxEvent.id, id), eq(outboxEvent.status, 'dead_letter')))
+          .returning({ id: outboxEvent.id, eventType: outboxEvent.eventType })
+        const subscriptionId = toRecord(existing?.payload).subscriptionId
+        if (
+          requeued.length > 0 &&
+          isSubscriptionSyncEventType(requeued[0].eventType) &&
+          typeof subscriptionId === 'string'
+        ) {
+          await recommitSubscriptionSync(tx, requeued[0].eventType, subscriptionId)
+        }
+        return requeued
+      })
 
       if (result.length === 0) {
         return NextResponse.json(

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

@@ -124,10 +124,10 @@ describe('admin subscription cancellation', () => {
       expect.objectContaining({ status: 'pending', attempts: 0, lastError: null })
     )
     expect(dbChainMockFns.set).toHaveBeenCalledWith({ cancelAtPeriodEnd: true })
-    expect(billingSubscriptionSyncMockFns.mockRecommitCancelAtPeriodEndSync).toHaveBeenCalledWith(
+    expect(billingSubscriptionSyncMockFns.mockRecommitSubscriptionSync).toHaveBeenCalledWith(
       expect.anything(),
-      'outbox-1',
-      true
+      'stripe.sync-cancel-at-period-end',
+      'sub-row-1'
     )
     expect(result).toMatchObject({
       operationId: '67e55044-10b1-426f-9247-bb680e5fe0c8',

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

@@ -10,7 +10,7 @@ import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/util
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
 import {
   enqueueCancelAtPeriodEndSync,
-  recommitCancelAtPeriodEndSync,
+  recommitSubscriptionSync,
 } from '@/lib/billing/webhooks/subscription-sync'
 import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
 
@@ -305,7 +305,6 @@ export async function requestDashboardSubscriptionCancellation({
           if (!restoredSubscription) {
             throw new Error('Cancellation subscription no longer exists')
           }
-          await recommitCancelAtPeriodEndSync(tx, existingOperation.id, true)
         }
         await tx
           .update(outboxEvent)
@@ -320,6 +319,13 @@ export async function requestDashboardSubscriptionCancellation({
           .where(
             and(eq(outboxEvent.id, existingOperation.id), eq(outboxEvent.status, 'dead_letter'))
           )
+        if (existingOperation.eventType === OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END) {
+          await recommitSubscriptionSync(
+            tx,
+            OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
+            existingOperation.subscriptionId
+          )
+        }
         return {
           operationId,
           outboxEventId: existingOperation.id,

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

@@ -1,18 +1,5 @@
 export const OUTBOX_EVENT_TYPES = {
-  /**
-   * Sync a subscription's `cancel_at_period_end` flag from our DB to
-   * Stripe. Enqueue through `enqueueCancelAtPeriodEndSync` in the same
-   * transaction as every DB change to `cancelAtPeriodEnd`.
-   *
-   * Guarantee: once every in-flight event for a subscription completes,
-   * Stripe holds the last value committed to the DB. Each handler pushes
-   * the row's current value (not its payload's) and re-reads the row
-   * after its Stripe write, retrying while the value moved, so racing
-   * events converge even when an earlier request lands in Stripe last.
-   * While an event is in flight, `reconcileSubscriptionSyncFromStripe`
-   * keeps webhook echoes and stale snapshots from overwriting the
-   * committed value; a change made in Stripe itself wins over it.
-   */
+  /** Sync `cancelAtPeriodEnd` from our DB to Stripe; enqueue via `enqueueCancelAtPeriodEndSync`. */
   STRIPE_SYNC_CANCEL_AT_PERIOD_END: 'stripe.sync-cancel-at-period-end',
   /** Cancel in Stripe; the verified deletion webhook remains the only DB entitlement authority. */
   STRIPE_CANCEL_SUBSCRIPTION_IMMEDIATELY: 'stripe.cancel-subscription-immediately',

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

@@ -20,6 +20,17 @@ import type { OutboxHandler } from '@/lib/core/outbox/service'
 
 const logger = createLogger('BillingOutboxHandlers')
 
+/**
+ * Passes a DB→Stripe sync handler makes before throwing for an outbox retry: each pass re-reads
+ * the row after its Stripe write and goes again when the value moved in the meantime.
+ */
+const MAX_SYNC_ATTEMPTS = 2
+
+/** A fresh key per Stripe write: the SDK reuses it across its own network retries of that call. */
+function customerContactSyncIdempotencyKey(eventId: string): string {
+  return `outbox:${eventId}:${generateShortId()}`
+}
+
 interface StripeCancelSubscriptionImmediatelyPayload {
   stripeSubscriptionId: string
   subscriptionId: string
@@ -110,9 +121,8 @@ const stripeSyncCancelAtPeriodEnd: OutboxHandler<CancelAtPeriodEndSyncPayload> =
 ) => {
   await recordAdminCancellationAudit({ ...payload, timing: 'period_end' })
   const stripe = requireStripeClient()
-  const maxSyncAttempts = 2
 
-  for (let attempt = 1; attempt <= maxSyncAttempts; attempt++) {
+  for (let attempt = 1; attempt <= MAX_SYNC_ATTEMPTS; attempt++) {
     const desiredValue = await readCancelAtPeriodEnd(payload.subscriptionId)
     if (desiredValue === null) {
       logger.warn('Subscription not found when syncing cancel_at_period_end', {
@@ -184,9 +194,8 @@ const stripeSyncSubscriptionSeats: OutboxHandler<SubscriptionSeatsSyncPayload> =
   ctx
 ) => {
   const stripe = requireStripeClient()
-  const maxSyncAttempts = 2
 
-  for (let attempt = 1; attempt <= maxSyncAttempts; attempt++) {
+  for (let attempt = 1; attempt <= MAX_SYNC_ATTEMPTS; attempt++) {
     const row = await getSubscriptionSeatSyncState(payload.subscriptionId)
     if (!row) {
       logger.warn('Subscription not found when syncing seats', {
@@ -450,9 +459,8 @@ const stripeSyncCustomerContact: OutboxHandler<StripeSyncCustomerContactPayload>
   ctx
 ) => {
   const stripe = requireStripeClient()
-  const maxSyncAttempts = 2
 
-  for (let attempt = 1; attempt <= maxSyncAttempts; attempt++) {
+  for (let attempt = 1; attempt <= MAX_SYNC_ATTEMPTS; attempt++) {
     const contact = await readCustomerContact(payload.subscriptionId)
     if (contact.status === 'skipped') {
       logger.warn(contact.reason, {
@@ -476,7 +484,7 @@ const stripeSyncCustomerContact: OutboxHandler<StripeSyncCustomerContactPayload>
           email: contact.email,
           ...(contact.name ? { name: contact.name } : {}),
         },
-        { idempotencyKey: `outbox:${ctx.eventId}:${generateShortId()}` }
+        { idempotencyKey: customerContactSyncIdempotencyKey(ctx.eventId) }
       )
     }