fix(billing,deployments): recover silently dropped post-event writes

Author: icecrasher321

apps/sim/app/api/cron/reconcile-billing-seats/route.ts

@@ -2,6 +2,7 @@ import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
+import { reconcileEntitlementLimitDrift } from '@/lib/billing/entitlement-drift'
 import { reconcileTeamSeatDrift } from '@/lib/billing/organizations/seat-drift'
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-handlers'
 import { findDeadLetteredEvents } from '@/lib/core/outbox/service'
@@ -18,11 +19,12 @@ const BILLING_SYNC_EVENT_TYPES = [
 ]
 
 /**
- * Periodic billing-seat reconciliation. Self-heals Team organizations whose
- * stored seat count drifted from their member count, and reports any
- * dead-lettered Stripe seat/cancel sync events so a member who has access but
- * whose seat charge never synced is surfaced for manual remediation rather than
- * silently under-billed.
+ * Periodic billing projection reconciliation. Self-heals Team organizations
+ * whose stored seat count drifted from their member count, restores paying
+ * users whose usage limit was left at the free-tier default by a failed
+ * subscription webhook, and reports any dead-lettered Stripe seat/cancel sync
+ * events so a member who has access but whose seat charge never synced is
+ * surfaced for manual remediation rather than silently under-billed.
  *
  * Scheduled in helm/sim/values.yaml under cronjobs.jobs.reconcileBillingSeats.
  */
@@ -36,6 +38,14 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
 
   try {
     const drift = await reconcileTeamSeatDrift()
+    const entitlementDrift = await reconcileEntitlementLimitDrift()
+
+    if (entitlementDrift.drifted > 0) {
+      logger.warn(
+        'Paying users were found at the free-tier usage limit — a subscription webhook failed to write their entitlement',
+        { requestId, ...entitlementDrift }
+      )
+    }
 
     const deadLettered = await findDeadLetteredEvents(BILLING_SYNC_EVENT_TYPES)
     if (deadLettered.length > 0) {
@@ -54,20 +64,22 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       )
     }
 
-    logger.info('Billing seat reconciliation completed', {
+    logger.info('Billing reconciliation completed', {
       requestId,
       ...drift,
+      entitlementDrift,
       deadLetteredBillingSyncs: deadLettered.length,
     })
 
     return NextResponse.json({
       success: true,
       requestId,
       drift,
+      entitlementDrift,
       deadLetteredBillingSyncs: deadLettered.length,
     })
   } catch (error) {
-    logger.error('Billing seat reconciliation failed', {
+    logger.error('Billing reconciliation failed', {
       requestId,
       error: toError(error).message,
     })

apps/sim/lib/billing/entitlement-drift.test.ts

@@ -0,0 +1,88 @@
+/**
+ * @vitest-environment node
+ */
+import {
+  queueTableRows,
+  resetDbChainMock,
+  resetEnvFlagsMock,
+  schemaMock,
+  setEnvFlags,
+} from '@sim/testing'
+import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockSyncUsageLimitsFromSubscription } = vi.hoisted(() => ({
+  mockSyncUsageLimitsFromSubscription: vi.fn(),
+}))
+
+vi.mock('@/lib/billing/core/usage', () => ({
+  syncUsageLimitsFromSubscription: mockSyncUsageLimitsFromSubscription,
+}))
+
+import { reconcileEntitlementLimitDrift } from '@/lib/billing/entitlement-drift'
+
+afterAll(resetEnvFlagsMock)
+
+describe('reconcileEntitlementLimitDrift', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+    setEnvFlags({ isBillingEnabled: true })
+    mockSyncUsageLimitsFromSubscription.mockResolvedValue(undefined)
+  })
+
+  afterAll(resetDbChainMock)
+
+  it('heals a paying user whose limit never left the free-tier default', async () => {
+    queueTableRows(schemaMock.subscription, [{ userId: 'user-1' }])
+    queueTableRows(schemaMock.userStats, [{ currentUsageLimit: '30' }])
+
+    const result = await reconcileEntitlementLimitDrift()
+
+    expect(mockSyncUsageLimitsFromSubscription).toHaveBeenCalledWith('user-1')
+    expect(result).toEqual({ drifted: 1, healed: 1, unresolved: 0 })
+  })
+
+  /**
+   * The canonical sync resolves entitlement from its own read of
+   * `subscription`. If that read disagrees with the row this sweep matched it
+   * writes nothing — the exact silent no-op that made the original incident
+   * invisible — so the sweep must report it rather than count it as healed.
+   */
+  it('reports a user the sync declined to lift off the floor', async () => {
+    queueTableRows(schemaMock.subscription, [{ userId: 'user-1' }])
+    queueTableRows(schemaMock.userStats, [{ currentUsageLimit: '5' }])
+
+    const result = await reconcileEntitlementLimitDrift()
+
+    expect(result).toEqual({ drifted: 1, healed: 0, unresolved: 1 })
+  })
+
+  it('leaves an org-covered member alone once the sync clears their personal limit', async () => {
+    queueTableRows(schemaMock.subscription, [{ userId: 'user-1' }])
+    queueTableRows(schemaMock.userStats, [{ currentUsageLimit: null }])
+
+    const result = await reconcileEntitlementLimitDrift()
+
+    expect(result).toEqual({ drifted: 1, healed: 0, unresolved: 0 })
+  })
+
+  it('counts a failing sync as unresolved without aborting the sweep', async () => {
+    queueTableRows(schemaMock.subscription, [{ userId: 'user-1' }, { userId: 'user-2' }])
+    queueTableRows(schemaMock.userStats, [{ currentUsageLimit: '30' }])
+    mockSyncUsageLimitsFromSubscription.mockRejectedValueOnce(new Error('db unavailable'))
+
+    const result = await reconcileEntitlementLimitDrift()
+
+    expect(mockSyncUsageLimitsFromSubscription).toHaveBeenCalledTimes(2)
+    expect(result).toEqual({ drifted: 2, healed: 1, unresolved: 1 })
+  })
+
+  it('does nothing when billing is disabled', async () => {
+    setEnvFlags({ isBillingEnabled: false })
+
+    const result = await reconcileEntitlementLimitDrift()
+
+    expect(mockSyncUsageLimitsFromSubscription).not.toHaveBeenCalled()
+    expect(result).toEqual({ drifted: 0, healed: 0, unresolved: 0 })
+  })
+})

apps/sim/lib/billing/entitlement-drift.ts

@@ -0,0 +1,136 @@
+import { db } from '@sim/db'
+import { subscription, userStats } from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { toError } from '@sim/utils/errors'
+import { and, eq, inArray, isNotNull, like, or, sql } from 'drizzle-orm'
+import { syncUsageLimitsFromSubscription } from '@/lib/billing/core/usage'
+import { ENTITLED_SUBSCRIPTION_STATUSES, getFreeTierLimit } from '@/lib/billing/subscriptions/utils'
+import { isBillingEnabled } from '@/lib/core/config/env-flags'
+
+const logger = createLogger('EntitlementDriftSweep')
+
+/** Max users healed per sweep run so a mass-drift event can't run away. */
+const MAX_RECONCILES_PER_RUN = 100
+
+export interface EntitlementDriftSweepResult {
+  /** Paying users found sitting at or below their free-tier floor. */
+  drifted: number
+  /** How many of those the canonical sync lifted off the floor this run. */
+  healed: number
+  /** Drifted users the sync declined to lift — always a bug, never expected. */
+  unresolved: number
+}
+
+/**
+ * Periodic backstop for the personal usage-limit projection.
+ *
+ * `user_stats.current_usage_limit` is written inline by the Stripe
+ * subscription webhook (`onSubscriptionComplete` → `syncSubscriptionUsageLimits`),
+ * behind five other statements that can each throw. There is no outbox event
+ * and no retry: if any of them fails, a customer who paid for Pro keeps the
+ * $5 free-tier limit indefinitely, with a perfectly correct `subscription`
+ * row next to it. That is exactly what a `jsonb_typeof`-on-a-`json`-column
+ * regression in the shared idempotency service did between 2026-08-05 and
+ * 2026-08-10, and nothing surfaced it — it took manual forensics weeks later.
+ *
+ * The detector is deliberately narrow: a Pro-plan user whose limit is at or
+ * below their free-plus-prepaid floor cannot be a legitimate state, because
+ * every paid tier's minimum is strictly above the free tier and
+ * `canEditUsageLimit` only ever lets a user raise their own limit. Partial
+ * drift (stuck at a lower *paid* tier after an upgrade) is not covered here —
+ * catching that would mean restating the per-tier dollar math in SQL, and
+ * this file must not become a second source of truth for it.
+ *
+ * Healing routes through `syncUsageLimitsFromSubscription`, which re-derives
+ * everything canonically — org-over-personal priority included, so a member
+ * covered by a paid org is correctly left alone rather than given a personal
+ * limit. That makes a false-positive candidate harmless, which is why the
+ * cheap filter is safe. The entitled branch of that sync raises with
+ * `greatest(...)`, so a limit a user raised themselves is never lowered.
+ */
+export async function reconcileEntitlementLimitDrift(): Promise<EntitlementDriftSweepResult> {
+  if (!isBillingEnabled) {
+    return { drifted: 0, healed: 0, unresolved: 0 }
+  }
+
+  const freeTierFloor = getFreeTierLimit().toString()
+
+  /**
+   * Joining `user_stats` on the subscription's `reference_id` also does the
+   * personal-vs-org scoping for free: an org-referenced subscription's
+   * `reference_id` is an organization id and matches no `user_stats` row.
+   */
+  const driftedRows = await db
+    .select({ userId: subscription.referenceId })
+    .from(subscription)
+    .innerJoin(userStats, eq(userStats.userId, subscription.referenceId))
+    .where(
+      and(
+        inArray(subscription.status, ENTITLED_SUBSCRIPTION_STATUSES),
+        or(eq(subscription.plan, 'pro'), like(subscription.plan, 'pro\\_%')),
+        isNotNull(userStats.currentUsageLimit),
+        sql`${userStats.currentUsageLimit} <= ${freeTierFloor}::numeric + coalesce(${userStats.creditBalance}, 0)`
+      )
+    )
+    .orderBy(sql`random()`)
+
+  const batch = driftedRows.slice(0, MAX_RECONCILES_PER_RUN)
+  if (driftedRows.length > batch.length) {
+    logger.warn('Entitlement drift exceeded the per-run cap; remainder deferred to later runs', {
+      drifted: driftedRows.length,
+      cap: MAX_RECONCILES_PER_RUN,
+      deferred: driftedRows.length - batch.length,
+    })
+  }
+
+  let healed = 0
+  let unresolved = 0
+
+  for (const row of batch) {
+    try {
+      await syncUsageLimitsFromSubscription(row.userId)
+
+      const [after] = await db
+        .select({ currentUsageLimit: userStats.currentUsageLimit })
+        .from(userStats)
+        .where(eq(userStats.userId, row.userId))
+        .limit(1)
+
+      /**
+       * A null limit means the sync resolved this user as org-scoped and
+       * cleared their personal limit — correct, not drift.
+       */
+      if (after?.currentUsageLimit == null) continue
+
+      if (Number(after.currentUsageLimit) > Number(freeTierFloor)) {
+        healed++
+        logger.info('Healed a paying user stuck at the free-tier limit', {
+          userId: row.userId,
+          currentUsageLimit: after.currentUsageLimit,
+        })
+        continue
+      }
+
+      unresolved++
+      /**
+       * The canonical sync ran and still left a paying user on the floor. It
+       * resolves entitlement from its own read of `subscription`, so this
+       * means that read disagrees with the row this sweep just matched — the
+       * sync is refusing, not the webhook. Loud because it is unreachable by
+       * design.
+       */
+      logger.error('Usage-limit sync left a paying user at the free-tier limit', {
+        userId: row.userId,
+        currentUsageLimit: after.currentUsageLimit,
+      })
+    } catch (error) {
+      unresolved++
+      logger.error('Failed to reconcile entitlement drift for user', {
+        userId: row.userId,
+        error: toError(error).message,
+      })
+    }
+  }
+
+  return { drifted: driftedRows.length, healed, unresolved }
+}