fix(billing): stamp every sync a later Stripe read confirms; ignore a moved cancel_at

Author: waleedlatif1

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

@@ -519,6 +519,66 @@ describe('cancel_at_period_end sync', () => {
     expect(stripe.subscription(pro.stripeSubscriptionId).cancel_at_period_end).toBe(false)
   })
 
+  it('records a later Stripe read even when every sync already carries its value', 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 restoreRead = stripe.holdNextRequest('subscriptions.retrieve')
+    const reconcilingRestore = deliver(stripe.events.at(-1) as Stripe.Event)
+    await restoreRead.reached
+
+    stripe.updateOutsideSim(pro.stripeSubscriptionId, { cancel_at_period_end: true })
+    const cancelRead = stripe.holdNextRequest('subscriptions.retrieve')
+    const reconcilingCancel = deliver(stripe.events.at(-1) as Stripe.Event)
+    await cancelRead.reached
+
+    cancelRead.release()
+    await reconcilingCancel
+    restoreRead.release()
+    await reconcilingRestore
+
+    expect((await storedSubscription(pro.subscriptionId)).cancelAtPeriodEnd).toBe(true)
+    await makeDue(pauseSync)
+    await expect(processEvent(pauseSync)).resolves.toBe('completed')
+    expect(stripe.subscription(pro.stripeSubscriptionId).cancel_at_period_end).toBe(true)
+  })
+
+  it('keeps a pending value when an unrelated Stripe update moves the cancellation date', 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 expect(processEvent(pauseSync)).resolves.toBe('completed')
+    const periodEnd = Math.floor(Date.now() / 1000) + 30 * 24 * 60 * 60
+    stripe.updateOutsideSim(pro.stripeSubscriptionId, { cancel_at: periodEnd })
+    await deliver(stripe.events.at(-1) as Stripe.Event)
+
+    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
+    )
+
+    stripe.updateOutsideSim(pro.stripeSubscriptionId, { cancel_at: periodEnd + 335 * 24 * 60 * 60 })
+    const moved = stripe.events.at(-1) as Stripe.Event
+    expect(Object.keys(moved.data.previous_attributes ?? {})).toEqual(['cancel_at'])
+    await deliver(moved)
+
+    expect((await storedSubscription(pro.subscriptionId)).cancelAtPeriodEnd).toBe(false)
+    await expect(processEvent(restoreSync)).resolves.toBe('completed')
+    expect(stripe.subscription(pro.stripeSubscriptionId).cancel_at_period_end).toBe(false)
+  })
+
   it('lets a change made in Stripe while a sync is pending win over the pending value', async () => {
     const pro = await createProUserInPaidOrganization()
     await pauseProSubscriptionForOrgCoverage(pro.userId)

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

@@ -101,8 +101,9 @@ async function readDatabaseClock(executor: DbOrTx): Promise<number> {
  *
  * A Sim commit is stamped with the clock under that lock and rewrites every such event. A value
  * taken from Stripe passes `observedAt`, the clock read just before Stripe was read, so it orders
- * by when it was observed (a slower reconcile of an earlier Stripe read cannot outrank a later
- * one), and rewrites only the events that do not already carry it.
+ * by when it was observed: it rewrites (value and stamp) only the events last stamped before that
+ * observation, including ones already holding the value, so a slower reconcile of an earlier
+ * Stripe read cannot outrank a later one and never overwrites a newer commit.
  */
 async function commitIntent<T extends SyncIntentFields>(
   tx: DbOrTx,
@@ -120,7 +121,7 @@ async function commitIntent<T extends SyncIntentFields>(
     eventType,
     subscriptionSubject(subscriptionId),
     committed,
-    observedAt === undefined ? undefined : fields
+    observedAt === undefined ? undefined : 'committedAt'
   )
   return committed
 }
@@ -358,15 +359,18 @@ async function readSyncIntents(executor: DbOrTx, subscriptionId: string) {
 }
 
 /**
- * True when the event records a cancellation change made in Stripe, not by Sim's sync. A change
- * to `cancel_at` counts too: Better Auth's restore clears `cancel_at` when it is set, and Stripe
- * may then list only `cancel_at` among the previous attributes.
+ * True when the event records a cancellation change made in Stripe, not by Sim's sync. A
+ * `cancel_at` that was set or cleared counts too: Better Auth's restore clears `cancel_at` when it
+ * is set, and Stripe may then list only `cancel_at` among the previous attributes. A `cancel_at`
+ * that only moved (e.g. a billing-interval switch on a subscription already ending) is not a
+ * cancellation change.
  */
 function isCancellationChangedInStripe(event: Stripe.Event): boolean {
   const previousAttributes = toRecord(event.data.previous_attributes)
-  if (!('cancel_at_period_end' in previousAttributes) && !('cancel_at' in previousAttributes)) {
-    return false
-  }
+  const scheduledOrCleared =
+    'cancel_at' in previousAttributes &&
+    (previousAttributes.cancel_at == null) !== (toRecord(event.data.object).cancel_at == null)
+  if (!('cancel_at_period_end' in previousAttributes) && !scheduledOrCleared) return false
   const idempotencyKey = event.request?.idempotency_key
   const issuedBySimSync =
     idempotencyKey?.startsWith(CANCEL_AT_PERIOD_END_SYNC_KEY_PREFIX) ||
@@ -410,9 +414,10 @@ function cancelAtPeriodEndSource(
  * - `cancelAtPeriodEnd`: while a cancel sync is in flight, its committed value wins over
  *   snapshots and over echoes of Sim's own writes. A change made in Stripe itself (customer
  *   portal, dashboard, Better Auth's cancel/restore endpoints), recognised by a non-Sim request
- *   changing `cancel_at_period_end` or `cancel_at`, wins and is committed onto every
- *   sync that can still run, unless Sim committed a newer value after Stripe was read. With no
- *   sync in flight Stripe wins, read live so out-of-order delivery cannot regress it.
+ *   changing `cancel_at_period_end` or setting or clearing `cancel_at`, wins and is committed
+ *   onto every sync that can still run, unless Sim committed a newer value after Stripe was
+ *   read. With no sync in flight Stripe wins, read live so out-of-order delivery cannot regress
+ *   it.
  * - Precedence across the two systems is arrival order, not wall-clock order: a Stripe-side
  *   change whose webhook is processed after a Sim commit wins even if the customer made it
  *   earlier. Stripe's `event.created` is not compared with the database clock, because skew

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

@@ -461,16 +461,17 @@ export async function listInflightOutboxEvents(
 
 /**
  * Shallow-merges `patch` into the payload of every `pending`, `processing`, or `dead_letter`
- * event of the type for one subject, skipping events whose payload already contains
- * `unlessPayloadContains`. One UPDATE; nothing is read into memory. Callers serialize writers
- * for the subject with their domain lock.
+ * event of the type for one subject. With `onlyIfOlderThanPatch`, naming a numeric payload key
+ * that `patch` sets, an event whose own value for that key is already at least the patch's is
+ * left alone, so a stale writer never overwrites a newer one. One UPDATE; nothing is read into
+ * memory. Callers serialize writers for the subject with their domain lock.
  */
 export async function patchRetryableOutboxEvents(
   executor: Pick<typeof db, 'update'>,
   eventType: string,
   subject: OutboxPayloadSubject,
   patch: Record<string, unknown>,
-  unlessPayloadContains?: Record<string, unknown>
+  onlyIfOlderThanPatch?: string
 ): Promise<number> {
   const patched = await executor
     .update(outboxEvent)
@@ -480,8 +481,8 @@ export async function patchRetryableOutboxEvents(
     .where(
       and(
         eventsForSubject([eventType], subject, RETRYABLE_OUTBOX_STATUSES),
-        unlessPayloadContains
-          ? sql`not (${outboxEvent.payload}::jsonb @> ${JSON.stringify(unlessPayloadContains)}::jsonb)`
+        onlyIfOlderThanPatch
+          ? sql`coalesce((${outboxEvent.payload} ->> ${onlyIfOlderThanPatch})::numeric, -1) < ${String(patch[onlyIfOlderThanPatch])}::numeric`
           : undefined
       )
     )