fix(billing): gate enterprise usage emails on an atomic claim and key the cache by source

The execution logger still summed an enterprise organization's whole reporting
window on every execution completion, only to compute the baseline for the
edge-triggered credits email. An edge needs an exact baseline, so the cached
sum could not serve it.

For reporting windows the organization credits email is now level-triggered.
After the run's usage is recorded, the logger reads the window's usage
through readSoftGateUsageCost and hands it to
maybeSendReportingCreditsThresholdEmail. That function compares the level to
a 'credits' claim on organization.limitNotifications, the same dedup the
storage, tables and seats emails use: re-arm below 70%, claim 80 or 100 with
one conditional UPDATE, and send only when this call won the claim. Opted-out
admins are dropped before the claim, so they never burn it. The claim flow is
now one shared notifyThresholdLevel in limit-notifications. It reads the
notified threshold by primary key first, so a completion below the band with
nothing armed, or above a threshold already sent, writes nothing and never
resolves recipients. The email content, recipients, opt-outs and billing link
are shared with the edge path through sendPaidCreditsEmail.

A cached level can trail the ledger by up to the cache TTL, so a crossing is
emailed up to one TTL late, on the first completion that sees it. It is never
sent twice and never dropped while usage stays over the threshold. A new
reporting window re-arms on its first completion because usage starts again
from zero. Stripe and default organization periods and personal accounts keep
the exact, edge-triggered path unchanged.

The reporting-window cache is now reachable only through
readSoftGateUsageCost. The raw cached reader is no longer exported and accepts
only a reporting period, and the cache key includes the period's source, so
another kind of period with the same bounds can never share a reporting sum.

Author: waleedlatif1

apps/sim/lib/billing/core/limit-notifications.test.ts

@@ -32,7 +32,10 @@ vi.mock('@/components/emails', () => ({
 }))
 vi.mock('@sim/platform-authz/workspace', () => ({ isOrgAdminRole: isOrgAdminRoleMock }))
 
-import { maybeSendLimitThresholdEmail } from '@/lib/billing/core/limit-notifications'
+import {
+  maybeSendLimitThresholdEmail,
+  notifyThresholdLevel,
+} from '@/lib/billing/core/limit-notifications'
 
 const baseUserParams = {
   category: 'storage' as const,
@@ -141,12 +144,41 @@ describe('maybeSendLimitThresholdEmail', () => {
   })
 
   it('re-arms but does not send when usage is fully cleared (zero usage)', async () => {
+    queueTableRows(schemaMock.userStats, [{ notifications: { storage: 100 } }])
     await maybeSendLimitThresholdEmail({ ...baseUserParams, currentUsage: 0, limit: 5 })
     expect(dbChainMockFns.update).toHaveBeenCalledTimes(1)
     expect(dbChainMockFns.returning).not.toHaveBeenCalled()
     expect(sendEmailSpy).not.toHaveBeenCalled()
   })
 
+  it('writes nothing below the band when the category is not armed', async () => {
+    await maybeSendLimitThresholdEmail({ ...baseUserParams, currentUsage: 0, limit: 5 })
+    expect(dbChainMockFns.update).not.toHaveBeenCalled()
+  })
+
+  it('skips recipients and the claim when the threshold was already notified', async () => {
+    queueTableRows(schemaMock.userStats, [{ notifications: { storage: 80 } }])
+    await maybeSendLimitThresholdEmail({ ...baseUserParams, currentUsage: 4.5, limit: 5 })
+    expect(getEmailPreferencesMock).not.toHaveBeenCalled()
+    expect(dbChainMockFns.update).not.toHaveBeenCalled()
+    expect(sendEmailSpy).not.toHaveBeenCalled()
+  })
+
+  it('keys claims per category, so a notified storage threshold never blocks credits', async () => {
+    queueTableRows(schemaMock.userStats, [{ notifications: { storage: 100 } }])
+    const send = vi.fn(() => Promise.resolve())
+    await notifyThresholdLevel({
+      scope: 'user',
+      stateId: 'u1',
+      category: 'credits',
+      percent: 100,
+      resolveRecipients: () => Promise.resolve([{ email: 'u1@example.com' }]),
+      send,
+    })
+    expect(dbChainMockFns.returning).toHaveBeenCalledTimes(1)
+    expect(send).toHaveBeenCalledWith(100, [{ email: 'u1@example.com' }])
+  })
+
   it('skips when the limit is non-positive', async () => {
     await maybeSendLimitThresholdEmail({ ...baseUserParams, currentUsage: 4, limit: 0 })
     expect(dbChainMockFns.update).not.toHaveBeenCalled()

apps/sim/lib/billing/core/limit-notifications.ts

@@ -2,6 +2,7 @@ import { db } from '@sim/db'
 import { member, organization, settings, user, userStats } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { isOrgAdminRole } from '@sim/platform-authz/workspace'
+import { toNumberOrNull } from '@sim/utils/coerce'
 import { and, eq, sql } from 'drizzle-orm'
 import type { HighestPrioritySubscription } from '@/lib/billing/core/plan'
 import { getHighestPrioritySubscription } from '@/lib/billing/core/subscription'
@@ -17,6 +18,18 @@ const logger = createLogger('LimitNotifications')
 /** Limit categories that send per-category threshold emails (credits has its own path). */
 export type LimitCategory = Extract<UpgradeReason, 'storage' | 'tables' | 'seats'>
 
+/**
+ * Every category whose threshold emails are deduplicated through `limitNotifications`: the limit
+ * categories above, plus credits for the level-triggered enterprise reporting-window email in
+ * `usage.ts`. Each category is its own key in the jsonb map, so their claims never interact.
+ */
+export type ThresholdCategory = LimitCategory | Extract<UpgradeReason, 'credits'>
+
+/** A threshold a notification is sent at: the 80% warning or the 100% limit reached. */
+export type NotifiedThreshold = 80 | 100
+
+type ThresholdScope = 'user' | 'organization'
+
 const WARN_THRESHOLD = 80
 const REACH_THRESHOLD = 100
 /** Usage must drop below this band before the same threshold can re-notify (hysteresis). */
@@ -26,7 +39,7 @@ const REARM_BELOW = 70
  * Resolve the threshold a given usage percent should be notified at:
  * 100 at/over the limit, 80 when approaching, 0 otherwise.
  */
-function thresholdFor(percent: number): 0 | 80 | 100 {
+function thresholdFor(percent: number): 0 | NotifiedThreshold {
   if (percent >= REACH_THRESHOLD) return REACH_THRESHOLD
   if (percent >= WARN_THRESHOLD) return WARN_THRESHOLD
   return 0
@@ -39,10 +52,10 @@ function thresholdFor(percent: number): 0 | 80 | 100 {
  * both claim, so the email is sent exactly once per crossing.
  */
 async function claimThreshold(
-  scope: 'user' | 'organization',
+  scope: ThresholdScope,
   id: string,
-  category: LimitCategory,
-  threshold: number
+  category: ThresholdCategory,
+  threshold: NotifiedThreshold
 ): Promise<boolean> {
   const setExpr = sql`jsonb_set(coalesce(${scope === 'user' ? userStats.limitNotifications : organization.limitNotifications}, '{}'::jsonb), ARRAY[${category}], to_jsonb(${threshold}::int))`
   const onlyIfLower =
@@ -68,9 +81,9 @@ async function claimThreshold(
 
 /** Re-arm a category (reset its stored threshold to 0) once usage falls back into the low band. */
 async function rearmThreshold(
-  scope: 'user' | 'organization',
+  scope: ThresholdScope,
   id: string,
-  category: LimitCategory
+  category: ThresholdCategory
 ): Promise<void> {
   const setExpr = sql`jsonb_set(coalesce(${scope === 'user' ? userStats.limitNotifications : organization.limitNotifications}, '{}'::jsonb), ARRAY[${category}], to_jsonb(0::int))`
   const onlyIfArmed =
@@ -91,7 +104,32 @@ async function rearmThreshold(
   }
 }
 
-interface LimitEmailRecipient {
+/**
+ * The threshold already notified for a category, read without taking a lock. It only lets a
+ * caller skip work — the re-arm and claim UPDATEs stay conditional, so a read that races a
+ * concurrent claim or re-arm can at worst cost one lost claim, which the next evaluation retries.
+ */
+async function readNotifiedThreshold(
+  scope: ThresholdScope,
+  id: string,
+  category: ThresholdCategory
+): Promise<number> {
+  const [row] =
+    scope === 'user'
+      ? await db
+          .select({ notifications: userStats.limitNotifications })
+          .from(userStats)
+          .where(eq(userStats.userId, id))
+          .limit(1)
+      : await db
+          .select({ notifications: organization.limitNotifications })
+          .from(organization)
+          .where(eq(organization.id, id))
+          .limit(1)
+  return toNumberOrNull(row?.notifications[category]) ?? 0
+}
+
+export interface LimitEmailRecipient {
   email: string
   name?: string
 }
@@ -108,8 +146,8 @@ async function isUnsubscribed(email: string): Promise<boolean> {
  * Returning an empty list means "nobody to notify" — the caller then skips the
  * claim so the dedup state isn't burned without an email going out.
  */
-async function resolveRecipients(
-  scope: 'user' | 'organization',
+export async function resolveLimitEmailRecipients(
+  scope: ThresholdScope,
   params: { userId?: string; userEmail?: string; userName?: string; organizationId?: string }
 ): Promise<LimitEmailRecipient[]> {
   if (scope === 'user') {
@@ -148,19 +186,62 @@ async function resolveRecipients(
   return recipients
 }
 
+/**
+ * Evaluate a threshold notification against the current usage level and send it at most once
+ * per crossing. The shared dedup flow behind every threshold email in this module and the
+ * credits email for enterprise reporting windows.
+ *
+ * The notified threshold is read first, so the steady state — usage below the band with nothing
+ * armed, or above a threshold already notified — costs one primary-key read and never writes.
+ * Below {@link REARM_BELOW} an armed category is re-armed; at or above a threshold not yet
+ * notified, eligible recipients are resolved with opt-outs applied BEFORE the atomic
+ * {@link claimThreshold}, so an opted-out recipient never burns the threshold (which would
+ * suppress a later email once notifications are re-enabled). `send` runs only when this call won
+ * the claim. Re-arm and claim are mutually exclusive per call, so the dedup stays a single atomic
+ * claim with no re-arm/claim interleaving race.
+ *
+ * Because the decision depends only on the current level and the persisted claim, a caller may
+ * pass a level that trails the truth: a crossing it misses is claimed on a later evaluation, and
+ * a crossing it sees twice is claimed once.
+ */
+export async function notifyThresholdLevel(params: {
+  scope: ThresholdScope
+  /** The user id for `user` scope, the organization id for `organization` scope. */
+  stateId: string
+  category: ThresholdCategory
+  /** Current usage as a percent of the limit. */
+  percent: number
+  /** Evaluate only the re-arm; never claim or send. */
+  rearmOnly?: boolean
+  resolveRecipients: () => Promise<LimitEmailRecipient[]>
+  send: (threshold: NotifiedThreshold, recipients: LimitEmailRecipient[]) => Promise<void>
+}): Promise<void> {
+  const { scope, stateId, category, percent } = params
+  const notified = await readNotifiedThreshold(scope, stateId, category)
+
+  if (percent < REARM_BELOW) {
+    if (notified > 0) await rearmThreshold(scope, stateId, category)
+    return
+  }
+
+  const desired = thresholdFor(percent)
+  if (params.rearmOnly || desired === 0 || notified >= desired) return
+
+  const recipients = await params.resolveRecipients()
+  if (recipients.length === 0) return
+
+  if (!(await claimThreshold(scope, stateId, category, desired))) return
+
+  await params.send(desired, recipients)
+}
+
 /**
  * Send a usage-limit threshold email (80% warning / 100% reached) for a
- * non-credit category, edge-triggered on the mutation that changed usage.
+ * non-credit category, evaluated on the mutation that changed usage.
  *
- * Flow: bail when billing is off or the limit is non-positive; re-arm the
- * persisted threshold when current usage is back in the low band; then (for
- * increases) resolve eligible recipients and atomically claim the threshold
- * before sending. Re-arm and claim are mutually exclusive per call — re-arm only
- * fires when `desired === 0` — so the dedup stays a single atomic
- * {@link claimThreshold} with no re-arm/claim interleaving race. Recipients are
- * resolved with opt-outs applied BEFORE the claim, so an opted-out recipient
- * never burns the threshold (which would suppress a later email once
- * notifications are re-enabled). Per-recipient send failures are isolated.
+ * Bails when billing is off or the limit is non-positive, then runs the shared
+ * {@link notifyThresholdLevel} flow on the category's persisted threshold.
+ * Per-recipient send failures are isolated.
  *
  * The highest threshold already emailed is persisted per category on
  * `user_stats` / `organization`; it re-arms once usage drops below
@@ -171,7 +252,7 @@ async function resolveRecipients(
  */
 export async function maybeSendLimitThresholdEmail(params: {
   category: LimitCategory
-  scope: 'user' | 'organization'
+  scope: ThresholdScope
   workspaceId: string
   currentUsage: number
   limit: number
@@ -195,64 +276,66 @@ export async function maybeSendLimitThresholdEmail(params: {
     if (params.limit <= 0) return
 
     const { category, scope } = params
-    const percent = Math.max(0, (params.currentUsage / params.limit) * 100)
-    const desired = thresholdFor(percent)
-
     const stateId = scope === 'user' ? params.userId : params.organizationId
     if (!stateId) return
 
-    if (percent < REARM_BELOW) {
-      await rearmThreshold(scope, stateId, category)
-    }
+    const percent = Math.max(0, (params.currentUsage / params.limit) * 100)
 
-    if (params.rearmOnly || desired === 0) return
-
-    const recipients = await resolveRecipients(scope, params)
-    if (recipients.length === 0) return
-
-    if (!(await claimThreshold(scope, stateId, category, desired))) return
-
-    const kind = desired === REACH_THRESHOLD ? 'reached' : 'warning'
-    const percentUsed = Math.min(100, Math.round(percent))
-    const upgradeLink = `${getBaseUrl()}${buildUpgradeHref(params.workspaceId, category)}`
-
-    const [{ getLimitEmailSubject, renderLimitThresholdEmail }, { sendEmail }] = await Promise.all([
-      import('@/components/emails'),
-      import('@/lib/messaging/email/mailer'),
-    ])
-
-    let sent = 0
-    for (const r of recipients) {
-      try {
-        const html = await renderLimitThresholdEmail({
-          kind,
-          reason: category,
-          userName: r.name,
-          usageLabel: params.usageLabel,
-          limitLabel: params.limitLabel,
-          percentUsed,
-          upgradeLink,
-        })
-
-        await sendEmail({
-          to: r.email,
-          subject: getLimitEmailSubject(category, kind),
-          html,
-          emailType: 'notifications',
-        })
-        sent++
-      } catch (sendError) {
-        logger.error('Failed to send limit email', {
-          category,
-          email: r.email,
-          error: sendError,
-        })
-      }
-    }
+    await notifyThresholdLevel({
+      scope,
+      stateId,
+      category,
+      percent,
+      rearmOnly: params.rearmOnly,
+      resolveRecipients: () => resolveLimitEmailRecipients(scope, params),
+      send: async (threshold, recipients) => {
+        const kind = threshold === REACH_THRESHOLD ? 'reached' : 'warning'
+        const percentUsed = Math.min(100, Math.round(percent))
+        const upgradeLink = `${getBaseUrl()}${buildUpgradeHref(params.workspaceId, category)}`
 
-    if (sent > 0) {
-      logger.info('Sent usage-limit threshold email', { category, scope, kind, percentUsed, sent })
-    }
+        const [{ getLimitEmailSubject, renderLimitThresholdEmail }, { sendEmail }] =
+          await Promise.all([import('@/components/emails'), import('@/lib/messaging/email/mailer')])
+
+        let sent = 0
+        for (const r of recipients) {
+          try {
+            const html = await renderLimitThresholdEmail({
+              kind,
+              reason: category,
+              userName: r.name,
+              usageLabel: params.usageLabel,
+              limitLabel: params.limitLabel,
+              percentUsed,
+              upgradeLink,
+            })
+
+            await sendEmail({
+              to: r.email,
+              subject: getLimitEmailSubject(category, kind),
+              html,
+              emailType: 'notifications',
+            })
+            sent++
+          } catch (sendError) {
+            logger.error('Failed to send limit email', {
+              category,
+              email: r.email,
+              error: sendError,
+            })
+          }
+        }
+
+        if (sent > 0) {
+          logger.info('Sent usage-limit threshold email', {
+            category,
+            scope,
+            kind,
+            percentUsed,
+            sent,
+          })
+        }
+      },
+    })
   } catch (error) {
     logger.error('Failed to send usage-limit threshold email', {
       category: params.category,