From ea79a8b7d59e7018d13ad9d51f2721f6d791ef92 Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Sat, 26 Sep 2026 10:14:39 -0700
Subject: [PATCH 1/4] improvement(billing): share the enterprise reporting
 usage read across processes

---
 .../lib/billing/core/limit-notifications.ts   |  23 +++-
 .../core/reporting-usage-cache.integration.ts | 121 ++++++++++++++++++
 .../lib/billing/core/reporting-usage-cache.ts |  99 ++++++++++++--
 .../core/usage-threshold-email.integration.ts | 108 ++++++++++++++++
 apps/sim/lib/billing/core/usage.test.ts       |  25 +---
 apps/sim/lib/billing/core/usage.ts            |  55 ++++----
 apps/sim/lib/logs/execution/logger.ts         |  56 ++++----
 packages/db/schema.ts                         |   7 +-
 8 files changed, 406 insertions(+), 88 deletions(-)
 create mode 100644 apps/sim/lib/billing/core/reporting-usage-cache.integration.ts
 create mode 100644 apps/sim/lib/billing/core/usage-threshold-email.integration.ts

diff --git a/apps/sim/lib/billing/core/limit-notifications.ts b/apps/sim/lib/billing/core/limit-notifications.ts
index 698bcf60239..f09b782d449 100644
--- a/apps/sim/lib/billing/core/limit-notifications.ts
+++ b/apps/sim/lib/billing/core/limit-notifications.ts
@@ -17,6 +17,9 @@ 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 emailed threshold is persisted, including credits. */
+type ClaimCategory = LimitCategory | Extract<UpgradeReason, 'credits'>
+
 const WARN_THRESHOLD = 80
 const REACH_THRESHOLD = 100
 /** Usage must drop below this band before the same threshold can re-notify (hysteresis). */
@@ -41,7 +44,7 @@ function thresholdFor(percent: number): 0 | 80 | 100 {
 async function claimThreshold(
   scope: 'user' | 'organization',
   id: string,
-  category: LimitCategory,
+  category: ClaimCategory,
   threshold: number
 ): Promise<boolean> {
   const setExpr = sql`jsonb_set(coalesce(${scope === 'user' ? userStats.limitNotifications : organization.limitNotifications}, '{}'::jsonb), ARRAY[${category}], to_jsonb(${threshold}::int))`
@@ -66,6 +69,24 @@ async function claimThreshold(
   return claimed.length > 0
 }
 
+const DAY_MS = 24 * 60 * 60 * 1000
+
+/**
+ * Claim a credits threshold (80 or 100) once per billing period, returning whether THIS call won
+ * it. The stored value is the period's start day followed by the threshold, so it only grows: a
+ * later period outranks every claim of an earlier one and re-arms both thresholds with no reset
+ * write, while within a period a claim of 100 also retires 80, and never the reverse.
+ */
+export function claimCreditsThreshold(
+  scope: 'user' | 'organization',
+  id: string,
+  periodStart: Date,
+  threshold: 80 | 100
+): Promise<boolean> {
+  const periodDay = Math.floor(periodStart.getTime() / DAY_MS)
+  return claimThreshold(scope, id, 'credits', periodDay * 1000 + threshold)
+}
+
 /** Re-arm a category (reset its stored threshold to 0) once usage falls back into the low band. */
 async function rearmThreshold(
   scope: 'user' | 'organization',
diff --git a/apps/sim/lib/billing/core/reporting-usage-cache.integration.ts b/apps/sim/lib/billing/core/reporting-usage-cache.integration.ts
new file mode 100644
index 00000000000..2ba801a18d5
--- /dev/null
+++ b/apps/sim/lib/billing/core/reporting-usage-cache.integration.ts
@@ -0,0 +1,121 @@
+/**
+ * The shared reporting-usage read against a real ledger in a disposable PostgreSQL schema and a
+ * real Redis. Skipped without `TEST_REDIS_URL`. Each test uses a fresh payer, so the in-process
+ * cache is always cold and every read models a new process.
+ */
+import type { db } from '@sim/db'
+import * as schema from '@sim/db/schema'
+import { readTestDatabaseUrl, readTestRedisUrl } from '@sim/db/testing/test-infrastructure'
+import { redisConfigMock, redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock'
+import { generateId } from '@sim/utils/id'
+import { drizzle } from 'drizzle-orm/postgres-js'
+import Redis from 'ioredis'
+import postgres from 'postgres'
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { transaction } = vi.hoisted(() => ({ transaction: vi.fn() }))
+const databaseUrl = readTestDatabaseUrl()
+const redisUrl = readTestRedisUrl()
+
+vi.mock('@sim/db', () => ({ db: { transaction }, dbReplica: {} }))
+vi.mock('@/lib/core/config/redis', () => redisConfigMock)
+
+import { readSoftGateUsageCost } from '@/lib/billing/core/reporting-usage-cache'
+import type { BillingEntity, UsageQueryPeriod } from '@/lib/billing/core/usage-log'
+
+const schemaName = `reporting_usage_${generateId().replaceAll('-', '')}`
+const connection = postgres(databaseUrl, {
+  max: 2,
+  prepare: false,
+  connection: { search_path: schemaName },
+  onnotice: () => undefined,
+})
+const database = drizzle(connection, { schema }) as typeof db
+
+const REPORTING: UsageQueryPeriod = {
+  start: new Date('2026-01-01T00:00:00.000Z'),
+  end: new Date('2027-01-01T00:00:00.000Z'),
+  source: 'reporting',
+}
+
+function sharedKey(payer: BillingEntity): string {
+  return `usage:reporting:v1:${payer.type}:${payer.id}:reporting:${REPORTING.start.toISOString()}:${REPORTING.end.toISOString()}`
+}
+
+describe.runIf(Boolean(redisUrl))('shared reporting usage read', () => {
+  let redis: Redis
+  let payer: BillingEntity
+
+  beforeAll(async () => {
+    redis = new Redis(redisUrl!, { lazyConnect: true, maxRetriesPerRequest: 0 })
+    await redis.connect()
+    await connection.unsafe(`CREATE SCHEMA "${schemaName}"`)
+    await connection.unsafe(`CREATE TABLE usage_log (
+      id text PRIMARY KEY, cost numeric NOT NULL, billing_entity_type text,
+      billing_entity_id text, billing_period_start timestamp, billing_period_end timestamp,
+      created_at timestamp NOT NULL
+    )`)
+    transaction.mockImplementation((callback) => database.transaction(callback))
+  })
+
+  beforeEach(async () => {
+    transaction.mockClear()
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(redis)
+    payer = { type: 'organization', id: generateId() }
+    await connection`INSERT INTO usage_log (id, cost, billing_entity_type, billing_entity_id, created_at)
+      VALUES (${generateId()}, 4.25, 'organization', ${payer.id}, '2026-03-01'),
+        (${generateId()}, 1.5, 'organization', ${payer.id}, '2026-06-01'),
+        (${generateId()}, 99, 'organization', ${payer.id}, '2025-12-31')`
+  })
+
+  afterEach(async () => {
+    await redis.del(sharedKey(payer))
+  })
+
+  afterAll(async () => {
+    await redis?.quit()
+    await connection.unsafe(`DROP SCHEMA IF EXISTS "${schemaName}" CASCADE`)
+    await connection.end()
+  })
+
+  it('serves a sum another process stored without summing the ledger', async () => {
+    await redis.set(sharedKey(payer), '12.5', 'PX', 30_000)
+
+    await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(12.5)
+    expect(transaction).not.toHaveBeenCalled()
+  })
+
+  it('sums the ledger exactly on a miss and stores the sum for other processes', async () => {
+    await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
+    expect(transaction).toHaveBeenCalledTimes(1)
+
+    await vi.waitFor(async () => expect(await redis.get(sharedKey(payer))).toBe('5.75'))
+    const ttl = await redis.pttl(sharedKey(payer))
+    expect(ttl).toBeGreaterThan(25_000)
+    expect(ttl).toBeLessThanOrEqual(35_000)
+  })
+
+  it('treats an unreadable stored sum as a miss and overwrites it', async () => {
+    await redis.set(sharedKey(payer), 'not-a-number', 'PX', 30_000)
+
+    await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
+    await vi.waitFor(async () => expect(await redis.get(sharedKey(payer))).toBe('5.75'))
+  })
+
+  it('sums the ledger promptly when Redis is unreachable', async () => {
+    const unreachable = new Redis('redis://127.0.0.1:1', {
+      enableOfflineQueue: true,
+      maxRetriesPerRequest: null,
+      retryStrategy: () => 1_000,
+    })
+    unreachable.on('error', () => undefined)
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(unreachable)
+    try {
+      const startedAt = Date.now()
+      await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
+      expect(Date.now() - startedAt).toBeLessThan(2_000)
+    } finally {
+      unreachable.disconnect()
+    }
+  })
+})
diff --git a/apps/sim/lib/billing/core/reporting-usage-cache.ts b/apps/sim/lib/billing/core/reporting-usage-cache.ts
index 2278207c117..c2b62655696 100644
--- a/apps/sim/lib/billing/core/reporting-usage-cache.ts
+++ b/apps/sim/lib/billing/core/reporting-usage-cache.ts
@@ -1,12 +1,19 @@
 import { db } from '@sim/db'
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
+import { randomInt } from '@sim/utils/random'
 import { LRUCache } from 'lru-cache'
 import {
   type BillingEntity,
   getBillingPeriodUsageCost,
   type UsageQueryPeriod,
 } from '@/lib/billing/core/usage-log'
+import { getRedisClient } from '@/lib/core/config/redis'
+import { withinDeadline } from '@/lib/core/utils/deadline'
 import type { DbClient } from '@/lib/db/types'
 
+const logger = createLogger('ReportingUsageCache')
+
 /**
  * How long a reporting-window usage sum is served before it is summed again.
  *
@@ -18,23 +25,97 @@ import type { DbClient } from '@/lib/db/types'
  * an admission gate lets a payer run on for at most this long past their limit, and a sum at or
  * above the limit is a refusal the true sum would also give. Thirty seconds keeps that overrun
  * small against a year-long allowance while turning a per-event scan into one per window.
+ *
+ * A sum is held both in Redis, shared by every process, and in each process that reads it, so a
+ * served sum can be up to twice this old (plus the Redis expiry's jitter).
  */
 export const REPORTING_USAGE_CACHE_TTL_MS = 30_000
 
+/** Redis expiry is jittered by up to this much, so payers summed together do not expire together. */
+const SHARED_TTL_JITTER_MS = 5_000
+
+/**
+ * How long a read waits on Redis before summing the ledger instead. The shared client queues
+ * commands while disconnected and has long timeouts, so without this a Redis outage would stall
+ * every gate behind it rather than cost one sum.
+ */
+const SHARED_READ_TIMEOUT_MS = 250
+
+/** Bump when a stored sum's meaning changes; old entries are then ignored. */
+const SHARED_KEY_VERSION = 'v1'
+
 /** A usage window known to be an enterprise reporting window — the only kind this cache serves. */
 type ReportingQueryPeriod = UsageQueryPeriod & { source: 'reporting' }
 
+function sharedReportingUsageKey(key: string): string {
+  return `usage:reporting:${SHARED_KEY_VERSION}:${key}`
+}
+
+/**
+ * A sum another process stored, or `undefined` when there is none to use. Redis being absent,
+ * slow, or failing, and a value that is not a non-negative number, are all misses: the caller
+ * sums the ledger, so the cache can cost a read its latency but never its answer.
+ */
+async function readSharedReportingUsageCost(key: string): Promise<number | undefined> {
+  const redis = getRedisClient()
+  if (!redis) return undefined
+  try {
+    const stored = await withinDeadline(
+      () => redis.get(sharedReportingUsageKey(key)),
+      Date.now() + SHARED_READ_TIMEOUT_MS
+    )
+    if (stored === null) return undefined
+    const cost = Number(stored)
+    if (stored.trim() !== '' && Number.isFinite(cost) && cost >= 0) return cost
+    logger.warn('Discarding unreadable shared reporting usage', { key })
+  } catch (error) {
+    logger.warn('Shared reporting usage read failed; summing the ledger', {
+      error: getErrorMessage(error),
+    })
+  }
+  return undefined
+}
+
+/** Fire-and-forget: a read never waits on, or fails because of, the shared write. */
+function writeSharedReportingUsageCost(key: string, cost: number): void {
+  const redis = getRedisClient()
+  if (!redis) return
+  const ttlMs = REPORTING_USAGE_CACHE_TTL_MS + randomInt(0, SHARED_TTL_JITTER_MS)
+  redis.set(sharedReportingUsageKey(key), String(cost), 'PX', ttlMs).catch((error: unknown) => {
+    logger.warn('Shared reporting usage write failed', { error: getErrorMessage(error) })
+  })
+}
+
+/**
+ * The sum from Redis when another process stored one, else the ledger's exact sum, stored for
+ * the others. Trigger.dev runs each task in a fresh process, so the in-process cache alone is
+ * always cold there; the shared value is what spares those runs the scan.
+ */
+async function sumReportingUsageCost(
+  key: string,
+  entity: BillingEntity,
+  period: ReportingQueryPeriod
+): Promise<number> {
+  const shared = await readSharedReportingUsageCost(key)
+  if (shared !== undefined) return shared
+  const cost = await getBillingPeriodUsageCost(entity, period)
+  writeSharedReportingUsageCost(key, cost)
+  return cost
+}
+
 /**
- * Sums shared across callers, one per payer and window. Every key is an enterprise payer's
- * current window, a few dozen bytes each, so the ceiling sits far above any process's working
- * set and only backstops memory; an eviction inside the TTL costs one extra sum.
+ * Sums held by this process, one per payer and window, in front of the shared Redis value. Every
+ * key is an enterprise payer's current window, a few dozen bytes each, so the ceiling sits far
+ * above any process's working set and only backstops memory; an eviction inside the TTL costs
+ * one extra read.
  *
- * `fetchMethod` coalesces concurrent misses onto one sum. A rejected sum is evicted rather than
+ * `fetchMethod` coalesces concurrent misses onto one read. A rejected sum is evicted rather than
  * stored (`noDeleteOnFetchRejection` and `allowStaleOnFetchRejection` stay off), so every caller
  * of that read sees the error it would have seen uncached and the next call sums again. There is
  * no settle deadline: the sum runs under the ledger's own `statement_timeout`, so the database
- * ends a slow one. There is deliberately no invalidator either — usage is written by execution
- * workers in other processes, so the TTL is the real bound.
+ * ends a slow one. There is deliberately no invalidator or lock either — usage is written by
+ * execution workers in other processes, so the TTL is the real bound, and concurrent misses in
+ * different processes each sum once.
  */
 const reportingUsageCache = new LRUCache<
   string,
@@ -43,8 +124,8 @@ const reportingUsageCache = new LRUCache<
 >({
   max: 1_000,
   ttl: REPORTING_USAGE_CACHE_TTL_MS,
-  fetchMethod: (_key, _stale, { context }) =>
-    getBillingPeriodUsageCost(context.entity, context.period),
+  fetchMethod: (key, _stale, { context }) =>
+    sumReportingUsageCost(key, context.entity, context.period),
 })
 
 /**
@@ -72,7 +153,7 @@ async function readCachedReportingUsageCost(
 /**
  * Period usage for a soft reader: an admission check, a display, or a level-triggered
  * notification that tolerates the cache's bounded under-count. Enterprise reporting windows are
- * served from the shared cache for up to {@link REPORTING_USAGE_CACHE_TTL_MS}, since their
+ * served from the shared cache for up to twice {@link REPORTING_USAGE_CACHE_TTL_MS}, since their
  * year-long sum is the expensive one; every other period is summed exactly, as before. A read on
  * a caller's own executor (a transaction or a replica) keeps its own snapshot and is never shared.
  * Never use it for invoicing, cycle close, an edge-triggered decision, or a read that must see its
diff --git a/apps/sim/lib/billing/core/usage-threshold-email.integration.ts b/apps/sim/lib/billing/core/usage-threshold-email.integration.ts
new file mode 100644
index 00000000000..7a26f066a5b
--- /dev/null
+++ b/apps/sim/lib/billing/core/usage-threshold-email.integration.ts
@@ -0,0 +1,108 @@
+/**
+ * The level-triggered usage threshold email against real claim state in a disposable PostgreSQL
+ * schema. Only delivery is stubbed: the mailer is the external boundary, and a sent email is the
+ * outcome under test.
+ */
+import type { db } from '@sim/db'
+import * as schema from '@sim/db/schema'
+import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
+import { emailMailerMock, emailMailerMockFns } from '@sim/testing/mocks/email-mailer.mock'
+import { emailTemplatesMock } from '@sim/testing/mocks/email-templates.mock'
+import { envFlagsMock, resetEnvFlagsMock, setEnvFlags } from '@sim/testing/mocks/env-flags.mock'
+import { generateId } from '@sim/utils/id'
+import { drizzle } from 'drizzle-orm/postgres-js'
+import postgres from 'postgres'
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { select, update } = vi.hoisted(() => ({ select: vi.fn(), update: vi.fn() }))
+const databaseUrl = readTestDatabaseUrl()
+
+vi.mock('@sim/db', () => ({ db: { select, update }, dbReplica: { select } }))
+vi.mock('@/lib/core/config/env-flags', () => envFlagsMock)
+vi.mock('@/components/emails', () => emailTemplatesMock)
+vi.mock('@/lib/messaging/email/mailer', () => emailMailerMock)
+vi.mock('@/lib/messaging/email/unsubscribe', () => ({ getEmailPreferences: async () => null }))
+
+import { maybeSendUsageThresholdEmail } from '@/lib/billing/core/usage'
+
+const { mockSendEmail } = emailMailerMockFns
+
+const schemaName = `usage_threshold_${generateId().replaceAll('-', '')}`
+const connection = postgres(databaseUrl, {
+  max: 4,
+  prepare: false,
+  connection: { search_path: schemaName },
+  onnotice: () => undefined,
+})
+const database = drizzle(connection, { schema }) as typeof db
+
+const SEPTEMBER = new Date('2026-09-01T00:00:00.000Z')
+const OCTOBER = new Date('2026-10-01T00:00:00.000Z')
+
+let organizationId: string
+
+function notify(currentUsage: number, periodStart = SEPTEMBER) {
+  return maybeSendUsageThresholdEmail({
+    scope: 'organization',
+    organizationId,
+    planName: 'Enterprise',
+    periodStart,
+    workspaceId: 'workspace',
+    currentUsage,
+    limit: 100,
+  })
+}
+
+beforeAll(async () => {
+  await connection.unsafe(`CREATE SCHEMA "${schemaName}"`)
+  await connection.unsafe(`
+    CREATE TABLE organization (id text PRIMARY KEY, limit_notifications jsonb);
+    CREATE TABLE "user" (id text PRIMARY KEY, email text, name text);
+    CREATE TABLE member (id text PRIMARY KEY, organization_id text, user_id text, role text);
+    CREATE TABLE settings (id text PRIMARY KEY, user_id text, billing_usage_notifications_enabled boolean);
+    INSERT INTO "user" VALUES ('admin', 'admin@example.com', 'Admin');
+  `)
+  select.mockImplementation((fields) => database.select(fields))
+  update.mockImplementation((table) => database.update(table))
+  setEnvFlags({ isBillingEnabled: true })
+})
+
+beforeEach(async () => {
+  mockSendEmail.mockClear()
+  organizationId = generateId()
+  await connection`INSERT INTO organization (id) VALUES (${organizationId})`
+  await connection`INSERT INTO member VALUES (${generateId()}, ${organizationId}, 'admin', 'owner')`
+})
+
+afterAll(async () => {
+  resetEnvFlagsMock()
+  await connection.unsafe(`DROP SCHEMA IF EXISTS "${schemaName}" CASCADE`)
+  await connection.end()
+})
+
+describe('usage threshold email', () => {
+  it('warns once per period however many completions find usage above 80%', async () => {
+    await Promise.all([notify(85), notify(85), notify(86)])
+    await notify(90)
+
+    expect(mockSendEmail).toHaveBeenCalledTimes(1)
+  })
+
+  it('still sends the reached email after the warning, but never the warning after it', async () => {
+    await notify(85)
+    await notify(100)
+    await notify(100)
+    await notify(85)
+
+    expect(mockSendEmail).toHaveBeenCalledTimes(2)
+  })
+
+  it('re-arms both thresholds in the next billing period', async () => {
+    await notify(100)
+    await notify(85, OCTOBER)
+    await notify(100, OCTOBER)
+    await notify(85)
+
+    expect(mockSendEmail).toHaveBeenCalledTimes(3)
+  })
+})
diff --git a/apps/sim/lib/billing/core/usage.test.ts b/apps/sim/lib/billing/core/usage.test.ts
index 8e6c862f637..ca214235db6 100644
--- a/apps/sim/lib/billing/core/usage.test.ts
+++ b/apps/sim/lib/billing/core/usage.test.ts
@@ -315,6 +315,7 @@ describe('maybeSendUsageThresholdEmail', () => {
     userEmail: 'user-1@example.com',
     userName: 'Ada',
     workspaceId: 'ws-1',
+    periodStart: new Date('2026-09-01T00:00:00.000Z'),
     limit: 20,
   }
 
@@ -323,6 +324,7 @@ describe('maybeSendUsageThresholdEmail', () => {
     setEnvFlags({ isBillingEnabled: true })
     mockGetEmailPreferences.mockResolvedValue(null)
     mockIsOrgAdminRole.mockReturnValue(true)
+    dbChainMockFns.returning.mockResolvedValue([{ id: 'claimed' }])
   })
 
   afterAll(() => {
@@ -332,15 +334,14 @@ describe('maybeSendUsageThresholdEmail', () => {
   it('emails a paid personal account at 100% with the raise-your-limit template', async () => {
     await maybeSendUsageThresholdEmail({
       ...paidUser,
-      percentBefore: 90,
-      percentAfter: 100,
-      currentUsageAfter: 20,
+      currentUsage: 20,
     })
 
     expect(mockRenderUsageLimitReached).toHaveBeenCalledWith(
       expect.objectContaining({ scope: 'user', planName: 'Pro' })
     )
     expect(mockRenderCreditsExhausted).not.toHaveBeenCalled()
+    expect(mockRenderUsageThreshold).not.toHaveBeenCalled()
     expect(mockGetLimitEmailSubject).toHaveBeenCalledWith('credits', 'reached')
     expect(mockSendEmail).toHaveBeenCalledTimes(1)
   })
@@ -357,9 +358,8 @@ describe('maybeSendUsageThresholdEmail', () => {
       planName: 'Team',
       organizationId: 'org-1',
       workspaceId: 'ws-1',
-      percentBefore: 95,
-      percentAfter: 100,
-      currentUsageAfter: 500,
+      periodStart: new Date('2026-09-01T00:00:00.000Z'),
+      currentUsage: 500,
       limit: 500,
     })
 
@@ -371,17 +371,4 @@ describe('maybeSendUsageThresholdEmail', () => {
       expect.objectContaining({ scope: 'organization' })
     )
   })
-
-  it('sends only the reached email when one execution crosses 80 and 100 together', async () => {
-    await maybeSendUsageThresholdEmail({
-      ...paidUser,
-      percentBefore: 70,
-      percentAfter: 100,
-      currentUsageAfter: 20,
-    })
-
-    expect(mockRenderUsageThreshold).not.toHaveBeenCalled()
-    expect(mockRenderUsageLimitReached).toHaveBeenCalledTimes(1)
-    expect(mockSendEmail).toHaveBeenCalledTimes(1)
-  })
 })
diff --git a/apps/sim/lib/billing/core/usage.ts b/apps/sim/lib/billing/core/usage.ts
index dbb35478073..c6454beca86 100644
--- a/apps/sim/lib/billing/core/usage.ts
+++ b/apps/sim/lib/billing/core/usage.ts
@@ -6,6 +6,7 @@ import { generateId } from '@sim/utils/id'
 import { and, eq, isNull, sql } from 'drizzle-orm'
 import { getEffectiveBillingStatus } from '@/lib/billing/core/access'
 import { defaultBillingPeriod } from '@/lib/billing/core/billing-period'
+import { claimCreditsThreshold } from '@/lib/billing/core/limit-notifications'
 import {
   getHighestPriorityPersonalSubscription,
   getHighestPrioritySubscription,
@@ -686,7 +687,11 @@ export async function getEffectiveCurrentPeriodCost(
 }
 
 /**
- * Send usage threshold notification when crossing from <80% to ≥80%.
+ * Send the usage threshold notification for the level usage is at: the 80% warning or the 100%
+ * limit-reached email, each at most once per billing period.
+ * - Level-triggered: any caller at or above a threshold may send it, so it tolerates a usage read
+ *   that lags the ledger, and the per-period claim ({@link claimCreditsThreshold}) is what keeps
+ *   repeated or concurrent callers to one email.
  * - Skips when billing is disabled.
  * - Respects user-level notifications toggle and unsubscribe preferences.
  * - For organization plans, emails owners/admins who have notifications enabled.
@@ -694,20 +699,27 @@ export async function getEffectiveCurrentPeriodCost(
 export async function maybeSendUsageThresholdEmail(params: {
   scope: 'user' | 'organization'
   planName: string
-  percentBefore: number
-  percentAfter: number
+  /** Start of the billing period the usage belongs to; each period re-arms both thresholds. */
+  periodStart: Date
   userId?: string
   userEmail?: string
   userName?: string
   organizationId?: string
   /** Workspace the usage occurred in, used to build a live upgrade/billing link. */
   workspaceId?: string
-  currentUsageAfter: number
+  currentUsage: number
   limit: number
 }): Promise<void> {
   try {
     if (!isBillingEnabled) return
-    if (params.limit <= 0 || params.currentUsageAfter <= 0) return
+    if (params.limit <= 0 || params.currentUsage <= 0) return
+
+    const percentUsed = (params.currentUsage / params.limit) * 100
+    const threshold = percentUsed >= 100 ? 100 : percentUsed >= 80 ? 80 : undefined
+    if (threshold === undefined) return
+    const stateId = params.scope === 'user' ? params.userId : params.organizationId
+    if (!stateId) return
+    if (!(await claimCreditsThreshold(params.scope, stateId, params.periodStart, threshold))) return
 
     const baseUrl = getBaseUrl()
     const isFreeUser = params.planName === 'Free'
@@ -726,14 +738,6 @@ export async function maybeSendUsageThresholdEmail(params: {
         ? `${baseUrl}/workspace/${params.workspaceId}/settings/billing`
         : `${baseUrl}/account/settings/billing`
 
-    // Check for 80% threshold crossing — used for paid users (budget warning) and free users (upgrade nudge)
-    const crosses80 = params.percentBefore < 80 && params.percentAfter >= 80
-    // Check for 100% threshold — every plan and scope (usage limit reached)
-    const crosses100 = params.percentBefore < 100 && params.percentAfter >= 100
-
-    // Skip if no thresholds crossed
-    if (!crosses80 && !crosses100) return
-
     /**
      * Delivers to the account's notification recipients: the payer for personal
      * scope, every org admin/owner for organization scope. Honors the per-user
@@ -773,8 +777,7 @@ export async function maybeSendUsageThresholdEmail(params: {
       }
     }
 
-    // !crosses100: one "reached" email, not a "nearing" and a "reached" in the same moment
-    if (crosses80 && !isFreeUser && !crosses100) {
+    if (threshold === 80 && !isFreeUser) {
       const ctaLink = billingSettingsLink
       await deliverToScope(async (email, name) => {
         const prefs = await getEmailPreferences(email)
@@ -784,8 +787,8 @@ export async function maybeSendUsageThresholdEmail(params: {
         const html = await renderUsageThresholdEmail({
           userName: name,
           planName: params.planName,
-          percentUsed: Math.min(100, Math.round(params.percentAfter)),
-          currentUsage: params.currentUsageAfter,
+          percentUsed: Math.round(percentUsed),
+          currentUsage: params.currentUsage,
           limit: params.limit,
           ctaLink,
         })
@@ -799,8 +802,7 @@ export async function maybeSendUsageThresholdEmail(params: {
       })
     }
 
-    // For 80% threshold email (free users only — skip if they also crossed 100% in same call)
-    if (crosses80 && isFreeUser && !crosses100) {
+    if (threshold === 80 && isFreeUser) {
       const upgradeLink = upgradeCreditsLink
       await deliverToScope(async (email, name) => {
         const prefs = await getEmailPreferences(email)
@@ -809,8 +811,8 @@ export async function maybeSendUsageThresholdEmail(params: {
         const { renderFreeTierUpgradeEmail, getEmailSubject, sendEmail } = await loadEmailDelivery()
         const html = await renderFreeTierUpgradeEmail({
           userName: name,
-          percentUsed: Math.min(100, Math.round(params.percentAfter)),
-          currentUsage: params.currentUsageAfter,
+          percentUsed: Math.round(percentUsed),
+          currentUsage: params.currentUsage,
           limit: params.limit,
           upgradeLink,
         })
@@ -824,15 +826,14 @@ export async function maybeSendUsageThresholdEmail(params: {
 
         logger.info('Free tier upgrade email sent', {
           email,
-          percentUsed: Math.round(params.percentAfter),
-          currentUsage: params.currentUsageAfter,
+          percentUsed: Math.round(percentUsed),
+          currentUsage: params.currentUsage,
           limit: params.limit,
         })
       })
     }
 
-    // Paid and org accounts get raise-your-limit copy — upgrading is not their remedy
-    if (crosses100) {
+    if (threshold === 100) {
       const useFreeCopy = isFreeUser && params.scope === 'user'
 
       await deliverToScope(async (email, name) => {
@@ -856,7 +857,7 @@ export async function maybeSendUsageThresholdEmail(params: {
               userName: name,
               planName: params.planName,
               scope: params.scope,
-              currentUsage: params.currentUsageAfter,
+              currentUsage: params.currentUsage,
               limit: params.limit,
               ctaLink: billingSettingsLink,
             })
@@ -874,7 +875,7 @@ export async function maybeSendUsageThresholdEmail(params: {
           email,
           scope: params.scope,
           planName: params.planName,
-          currentUsage: params.currentUsageAfter,
+          currentUsage: params.currentUsage,
           limit: params.limit,
         })
       })
diff --git a/apps/sim/lib/logs/execution/logger.ts b/apps/sim/lib/logs/execution/logger.ts
index a63334e1ecd..36e1d9db0a6 100644
--- a/apps/sim/lib/logs/execution/logger.ts
+++ b/apps/sim/lib/logs/execution/logger.ts
@@ -16,6 +16,7 @@ import {
   type BillingAttributionSnapshot,
   toBillingContext,
 } from '@/lib/billing/core/billing-attribution'
+import { readSoftGateUsageCost } from '@/lib/billing/core/reporting-usage-cache'
 import {
   getHighestPriorityPersonalSubscription,
   getHighestPrioritySubscription,
@@ -1291,9 +1292,12 @@ export class ExecutionLogger implements IExecutionLoggerService {
             )[0]
           : undefined
 
-      // Resolve the billing context + the pre-increment usage snapshot for the
-      // threshold email BEFORE recording, so currentUsageAfter = before +
-      // costDelta doesn't double-count this boundary's own increment.
+      /**
+       * The billing context and pre-increment usage for the threshold email are read BEFORE
+       * recording, so usage after = before + costDelta doesn't double-count this boundary's own
+       * increment. The organization read is the soft one: the email is level-triggered and
+       * claimed once per period, so a lagging sum only delays it.
+       */
       type EmailContext =
         | {
             scope: 'user'
@@ -1301,12 +1305,14 @@ export class ExecutionLogger implements IExecutionLoggerService {
             userEmail: string
             userName: string | null
             planName: string
+            periodStart: Date
             before: Awaited<ReturnType<typeof checkResolvedUsageStatus>>
           }
         | {
             scope: 'organization'
             organizationId: string
             planName: string
+            periodStart: Date
             orgLimit: number
             orgUsageBefore: number
           }
@@ -1326,19 +1332,22 @@ export class ExecutionLogger implements IExecutionLoggerService {
           payerSubscription.plan,
           payerSubscription.seats
         )
-        const { getBillingPeriodUsageCost } = await import('@/lib/billing/core/usage-log')
-        const orgLedger = await getBillingPeriodUsageCost(
-          billingAttribution.billingEntity,
-          exactBillingContext.billingPeriod
-        )
         emailContext = {
           scope: 'organization',
           organizationId,
           planName: getDisplayPlanName(payerSubscription.plan),
+          periodStart: exactBillingContext.billingPeriod.start,
           orgLimit,
-          orgUsageBefore: orgLedger,
+          orgUsageBefore: await readSoftGateUsageCost(
+            billingAttribution.billingEntity,
+            exactBillingContext.billingPeriod
+          ),
         }
-      } else if (billingAttribution?.billingEntity.type === 'user' && usr?.email) {
+      } else if (
+        billingAttribution?.billingEntity.type === 'user' &&
+        exactBillingContext &&
+        usr?.email
+      ) {
         const sub = await getHighestPriorityPersonalSubscription(usr.id)
         const { getDisplayPlanName } = await import('@/lib/billing/plan-helpers')
         emailContext = {
@@ -1347,6 +1356,7 @@ export class ExecutionLogger implements IExecutionLoggerService {
           userEmail: usr.email,
           userName: usr.name,
           planName: getDisplayPlanName(sub?.plan),
+          periodStart: exactBillingContext.billingPeriod.start,
           before: await checkResolvedUsageStatus(usr.id, sub),
         }
       }
@@ -1366,40 +1376,26 @@ export class ExecutionLogger implements IExecutionLoggerService {
 
       // Best-effort usage-threshold email.
       if (emailContext?.scope === 'user') {
-        const limit = emailContext.before.limit
-        const percentBefore = emailContext.before.percentUsed
-        const percentAfter =
-          limit > 0 ? Math.min(100, percentBefore + (costDelta / limit) * 100) : percentBefore
-        const currentUsageAfter = emailContext.before.currentUsage + costDelta
-
         await maybeSendUsageThresholdEmail({
           scope: 'user',
           userId: emailContext.userId,
           userEmail: emailContext.userEmail,
           userName: emailContext.userName || undefined,
           planName: emailContext.planName,
+          periodStart: emailContext.periodStart,
           workspaceId: updatedLog.workspaceId,
-          percentBefore,
-          percentAfter,
-          currentUsageAfter,
-          limit,
+          currentUsage: emailContext.before.currentUsage + costDelta,
+          limit: emailContext.before.limit,
         })
       } else if (emailContext?.scope === 'organization') {
-        const { orgLimit, orgUsageBefore } = emailContext
-        const percentBefore = orgLimit > 0 ? Math.min(100, (orgUsageBefore / orgLimit) * 100) : 0
-        const percentAfter =
-          orgLimit > 0 ? Math.min(100, percentBefore + (costDelta / orgLimit) * 100) : percentBefore
-        const currentUsageAfter = orgUsageBefore + costDelta
-
         await maybeSendUsageThresholdEmail({
           scope: 'organization',
           organizationId: emailContext.organizationId,
           planName: emailContext.planName,
+          periodStart: emailContext.periodStart,
           workspaceId: updatedLog.workspaceId,
-          percentBefore,
-          percentAfter,
-          currentUsageAfter,
-          limit: orgLimit,
+          currentUsage: emailContext.orgUsageBefore + costDelta,
+          limit: emailContext.orgLimit,
         })
       }
     } catch (e) {
diff --git a/packages/db/schema.ts b/packages/db/schema.ts
index 330068231c5..0a5df7328aa 100644
--- a/packages/db/schema.ts
+++ b/packages/db/schema.ts
@@ -1357,7 +1357,9 @@ export const userStats = pgTable('user_stats', {
    * Highest usage-limit threshold already emailed per category (e.g.
    * `{ storage: 80, tables: 100 }`). Prevents re-spamming the same warning;
    * re-arms when usage drops back below the re-arm band. Keyed by limit
-   * category ('storage' | 'tables'); seats live on `organization`.
+   * category ('storage' | 'tables'); seats live on `organization`. `credits`
+   * holds the billing period's start day times 1000 plus the threshold, so each
+   * period re-arms it without a reset (see `claimCreditsThreshold`).
    *
    * Dedup granularity is per billing account per category — intentionally NOT
    * per table, so a user hitting the row limit on several tables gets one
@@ -1719,7 +1721,8 @@ export const organization = pgTable('organization', {
   /**
    * Highest usage-limit threshold already emailed per category for this org
    * (e.g. `{ seats: 80, storage: 100 }`). Mirrors `user_stats.limitNotifications`
-   * for org-scoped (pooled) limits. Re-arms when usage drops below the re-arm band.
+   * for org-scoped (pooled) limits. Re-arms when usage drops below the re-arm band;
+   * `credits` instead re-arms each billing period (see `claimCreditsThreshold`).
    */
   limitNotifications: jsonb('limit_notifications')
     .$type<Record<string, number>>()

From 655432c54b7f8ff4ebd3e5b3f15df6d08ae19bcb Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Sat, 26 Sep 2026 10:38:50 -0700
Subject: [PATCH 2/4] improvement(billing): key the usage email claim on period
 and limit, and harden the shared read

---
 .../lib/billing/core/limit-notifications.ts   |  86 +++++++++-----
 .../core/reporting-usage-cache.integration.ts |  23 +++-
 .../lib/billing/core/reporting-usage-cache.ts |  43 ++++---
 .../core/usage-threshold-email.integration.ts |  55 +++++++--
 apps/sim/lib/billing/core/usage.test.ts       |   6 +-
 apps/sim/lib/billing/core/usage.ts            | 111 ++++++------------
 apps/sim/lib/logs/execution/logger.ts         |   8 +-
 packages/db/schema.ts                         |   8 +-
 8 files changed, 203 insertions(+), 137 deletions(-)

diff --git a/apps/sim/lib/billing/core/limit-notifications.ts b/apps/sim/lib/billing/core/limit-notifications.ts
index f09b782d449..eb6d48ba358 100644
--- a/apps/sim/lib/billing/core/limit-notifications.ts
+++ b/apps/sim/lib/billing/core/limit-notifications.ts
@@ -2,7 +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 { and, eq, sql } from 'drizzle-orm'
+import { and, eq, type SQL, sql } from 'drizzle-orm'
 import type { HighestPrioritySubscription } from '@/lib/billing/core/plan'
 import { getHighestPrioritySubscription } from '@/lib/billing/core/subscription'
 import type { BillingEntity } from '@/lib/billing/core/usage-log'
@@ -17,9 +17,6 @@ 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 emailed threshold is persisted, including credits. */
-type ClaimCategory = LimitCategory | Extract<UpgradeReason, 'credits'>
-
 const WARN_THRESHOLD = 80
 const REACH_THRESHOLD = 100
 /** Usage must drop below this band before the same threshold can re-notify (hysteresis). */
@@ -35,6 +32,32 @@ function thresholdFor(percent: number): 0 | 80 | 100 {
   return 0
 }
 
+/**
+ * Replace the account's `limitNotifications` with `next` when `condition` holds, returning
+ * whether the row was updated.
+ */
+async function writeLimitNotifications(
+  scope: 'user' | 'organization',
+  id: string,
+  next: SQL,
+  condition: SQL
+): Promise<boolean> {
+  const written =
+    scope === 'user'
+      ? await db
+          .update(userStats)
+          .set({ limitNotifications: next })
+          .where(and(eq(userStats.userId, id), condition))
+          .returning({ id: userStats.userId })
+      : await db
+          .update(organization)
+          .set({ limitNotifications: next })
+          .where(and(eq(organization.id, id), condition))
+          .returning({ id: organization.id })
+
+  return written.length > 0
+}
+
 /**
  * Atomically claim a threshold for a category: advance the stored value to
  * `threshold` only if it is currently lower, returning whether THIS call won the
@@ -44,7 +67,7 @@ function thresholdFor(percent: number): 0 | 80 | 100 {
 async function claimThreshold(
   scope: 'user' | 'organization',
   id: string,
-  category: ClaimCategory,
+  category: LimitCategory,
   threshold: number
 ): Promise<boolean> {
   const setExpr = sql`jsonb_set(coalesce(${scope === 'user' ? userStats.limitNotifications : organization.limitNotifications}, '{}'::jsonb), ARRAY[${category}], to_jsonb(${threshold}::int))`
@@ -53,38 +76,37 @@ async function claimThreshold(
       ? sql`coalesce((${userStats.limitNotifications} ->> ${category})::int, 0) < ${threshold}`
       : sql`coalesce((${organization.limitNotifications} ->> ${category})::int, 0) < ${threshold}`
 
-  const claimed =
-    scope === 'user'
-      ? await db
-          .update(userStats)
-          .set({ limitNotifications: setExpr })
-          .where(and(eq(userStats.userId, id), onlyIfLower))
-          .returning({ id: userStats.userId })
-      : await db
-          .update(organization)
-          .set({ limitNotifications: setExpr })
-          .where(and(eq(organization.id, id), onlyIfLower))
-          .returning({ id: organization.id })
-
-  return claimed.length > 0
+  return writeLimitNotifications(scope, id, setExpr, onlyIfLower)
 }
 
 const DAY_MS = 24 * 60 * 60 * 1000
 
 /**
- * Claim a credits threshold (80 or 100) once per billing period, returning whether THIS call won
- * it. The stored value is the period's start day followed by the threshold, so it only grows: a
- * later period outranks every claim of an earlier one and re-arms both thresholds with no reset
- * write, while within a period a claim of 100 also retires 80, and never the reverse.
+ * Claim a credits threshold (80 or 100), returning whether THIS call won it. The claim is keyed
+ * on the billing period and the limit: `credits` holds the highest threshold emailed while
+ * `creditsPeriod` (the period's start day) and `creditsLimit` (the limit in cents) still match,
+ * so a new period or a changed limit — in either direction — re-arms both thresholds with no
+ * reset write, while within one a claim of 100 also retires 80, and never the reverse.
  */
-export function claimCreditsThreshold(
-  scope: 'user' | 'organization',
-  id: string,
-  periodStart: Date,
+export function claimCreditsThreshold(params: {
+  scope: 'user' | 'organization'
+  id: string
+  periodStart: Date
+  limit: number
   threshold: 80 | 100
-): Promise<boolean> {
-  const periodDay = Math.floor(periodStart.getTime() / DAY_MS)
-  return claimThreshold(scope, id, 'credits', periodDay * 1000 + threshold)
+}): Promise<boolean> {
+  const { scope, id, threshold } = params
+  const periodDay = Math.floor(params.periodStart.getTime() / DAY_MS)
+  const limitCents = Math.round(params.limit * 100)
+  const column = scope === 'user' ? userStats.limitNotifications : organization.limitNotifications
+  const next = sql`coalesce(${column}, '{}'::jsonb) || jsonb_build_object('credits', ${threshold}::int, 'creditsPeriod', ${periodDay}::bigint, 'creditsLimit', ${limitCents}::bigint)`
+  const unclaimed = sql`not (
+    (${column} ->> 'creditsPeriod')::bigint is not distinct from ${periodDay}::bigint
+    and (${column} ->> 'creditsLimit')::bigint is not distinct from ${limitCents}::bigint
+    and coalesce((${column} ->> 'credits')::int, 0) >= ${threshold}::int
+  )`
+
+  return writeLimitNotifications(scope, id, next, unclaimed)
 }
 
 /** Re-arm a category (reset its stored threshold to 0) once usage falls back into the low band. */
@@ -129,7 +151,7 @@ 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(
+export async function resolveLimitEmailRecipients(
   scope: 'user' | 'organization',
   params: { userId?: string; userEmail?: string; userName?: string; organizationId?: string }
 ): Promise<LimitEmailRecipient[]> {
@@ -228,7 +250,7 @@ export async function maybeSendLimitThresholdEmail(params: {
 
     if (params.rearmOnly || desired === 0) return
 
-    const recipients = await resolveRecipients(scope, params)
+    const recipients = await resolveLimitEmailRecipients(scope, params)
     if (recipients.length === 0) return
 
     if (!(await claimThreshold(scope, stateId, category, desired))) return
diff --git a/apps/sim/lib/billing/core/reporting-usage-cache.integration.ts b/apps/sim/lib/billing/core/reporting-usage-cache.integration.ts
index 2ba801a18d5..a67ac000e80 100644
--- a/apps/sim/lib/billing/core/reporting-usage-cache.integration.ts
+++ b/apps/sim/lib/billing/core/reporting-usage-cache.integration.ts
@@ -95,11 +95,30 @@ describe.runIf(Boolean(redisUrl))('shared reporting usage read', () => {
     expect(ttl).toBeLessThanOrEqual(35_000)
   })
 
-  it('treats an unreadable stored sum as a miss and overwrites it', async () => {
+  it('treats an unreadable stored sum as a miss', async () => {
     await redis.set(sharedKey(payer), 'not-a-number', 'PX', 30_000)
 
     await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
-    await vi.waitFor(async () => expect(await redis.get(sharedKey(payer))).toBe('5.75'))
+    expect(transaction).toHaveBeenCalledTimes(1)
+  })
+
+  it('never lets a slower, older sum replace one stored while it ran', async () => {
+    transaction.mockImplementationOnce(async (callback) => {
+      const result = await database.transaction(callback)
+      await redis.set(sharedKey(payer), '9.99', 'PX', 30_000)
+      return result
+    })
+
+    await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
+    expect(await redis.get(sharedKey(payer))).toBe('9.99')
+  })
+
+  it('sums the ledger when the Redis client cannot be built', async () => {
+    redisConfigMockFns.mockGetRedisClient.mockImplementation(() => {
+      throw new Error('Invalid Redis configuration')
+    })
+
+    await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
   })
 
   it('sums the ledger promptly when Redis is unreachable', async () => {
diff --git a/apps/sim/lib/billing/core/reporting-usage-cache.ts b/apps/sim/lib/billing/core/reporting-usage-cache.ts
index c2b62655696..571f9fbb799 100644
--- a/apps/sim/lib/billing/core/reporting-usage-cache.ts
+++ b/apps/sim/lib/billing/core/reporting-usage-cache.ts
@@ -26,8 +26,10 @@ const logger = createLogger('ReportingUsageCache')
  * above the limit is a refusal the true sum would also give. Thirty seconds keeps that overrun
  * small against a year-long allowance while turning a per-event scan into one per window.
  *
- * A sum is held both in Redis, shared by every process, and in each process that reads it, so a
- * served sum can be up to twice this old (plus the Redis expiry's jitter).
+ * A sum is held both in Redis, shared by every process, and in each process that reads it. It
+ * reflects the ledger as of the moment its sum began, so a served sum can omit usage written over
+ * the sum's own duration, plus up to this TTL and its jitter in Redis, plus up to this TTL again
+ * in the reading process.
  */
 export const REPORTING_USAGE_CACHE_TTL_MS = 30_000
 
@@ -57,9 +59,9 @@ function sharedReportingUsageKey(key: string): string {
  * sums the ledger, so the cache can cost a read its latency but never its answer.
  */
 async function readSharedReportingUsageCost(key: string): Promise<number | undefined> {
-  const redis = getRedisClient()
-  if (!redis) return undefined
   try {
+    const redis = getRedisClient()
+    if (!redis) return undefined
     const stored = await withinDeadline(
       () => redis.get(sharedReportingUsageKey(key)),
       Date.now() + SHARED_READ_TIMEOUT_MS
@@ -76,14 +78,26 @@ async function readSharedReportingUsageCost(key: string): Promise<number | undef
   return undefined
 }
 
-/** Fire-and-forget: a read never waits on, or fails because of, the shared write. */
+function warnSharedWriteFailed(error: unknown): void {
+  logger.warn('Shared reporting usage write failed', { error: getErrorMessage(error) })
+}
+
+/**
+ * Fire-and-forget: a read never waits on, or fails because of, the shared write. The write only
+ * lands when no sum is stored (`NX`), so a slow, older sum can never replace a fresher one or
+ * extend its expiry.
+ */
 function writeSharedReportingUsageCost(key: string, cost: number): void {
-  const redis = getRedisClient()
-  if (!redis) return
-  const ttlMs = REPORTING_USAGE_CACHE_TTL_MS + randomInt(0, SHARED_TTL_JITTER_MS)
-  redis.set(sharedReportingUsageKey(key), String(cost), 'PX', ttlMs).catch((error: unknown) => {
-    logger.warn('Shared reporting usage write failed', { error: getErrorMessage(error) })
-  })
+  try {
+    const redis = getRedisClient()
+    if (!redis) return
+    const ttlMs = REPORTING_USAGE_CACHE_TTL_MS + randomInt(0, SHARED_TTL_JITTER_MS)
+    redis
+      .set(sharedReportingUsageKey(key), String(cost), 'PX', ttlMs, 'NX')
+      .catch(warnSharedWriteFailed)
+  } catch (error) {
+    warnSharedWriteFailed(error)
+  }
 }
 
 /**
@@ -153,9 +167,10 @@ async function readCachedReportingUsageCost(
 /**
  * Period usage for a soft reader: an admission check, a display, or a level-triggered
  * notification that tolerates the cache's bounded under-count. Enterprise reporting windows are
- * served from the shared cache for up to twice {@link REPORTING_USAGE_CACHE_TTL_MS}, since their
- * year-long sum is the expensive one; every other period is summed exactly, as before. A read on
- * a caller's own executor (a transaction or a replica) keeps its own snapshot and is never shared.
+ * served from the shared cache, within the lag {@link REPORTING_USAGE_CACHE_TTL_MS} describes,
+ * since their year-long sum is the expensive one; every other period is summed exactly, as
+ * before. A read on a caller's own executor (a transaction or a replica) keeps its own snapshot
+ * and is never shared.
  * Never use it for invoicing, cycle close, an edge-triggered decision, or a read that must see its
  * own write.
  */
diff --git a/apps/sim/lib/billing/core/usage-threshold-email.integration.ts b/apps/sim/lib/billing/core/usage-threshold-email.integration.ts
index 7a26f066a5b..6b47ed63553 100644
--- a/apps/sim/lib/billing/core/usage-threshold-email.integration.ts
+++ b/apps/sim/lib/billing/core/usage-threshold-email.integration.ts
@@ -40,19 +40,30 @@ const SEPTEMBER = new Date('2026-09-01T00:00:00.000Z')
 const OCTOBER = new Date('2026-10-01T00:00:00.000Z')
 
 let organizationId: string
+let adminId: string
 
-function notify(currentUsage: number, periodStart = SEPTEMBER) {
+/** One completion that leaves the organization at `usage`, having recorded `costDelta` of it. */
+function notify(
+  usage: number,
+  { periodStart = SEPTEMBER, limit = 100, costDelta = 1 } = {}
+): Promise<void> {
   return maybeSendUsageThresholdEmail({
     scope: 'organization',
     organizationId,
     planName: 'Enterprise',
     periodStart,
     workspaceId: 'workspace',
-    currentUsage,
-    limit: 100,
+    usageBefore: usage - costDelta,
+    costDelta,
+    limit,
   })
 }
 
+async function setNotificationsEnabled(enabled: boolean): Promise<void> {
+  await connection`INSERT INTO settings VALUES (${adminId}, ${adminId}, ${enabled})
+    ON CONFLICT (id) DO UPDATE SET billing_usage_notifications_enabled = ${enabled}`
+}
+
 beforeAll(async () => {
   await connection.unsafe(`CREATE SCHEMA "${schemaName}"`)
   await connection.unsafe(`
@@ -60,7 +71,6 @@ beforeAll(async () => {
     CREATE TABLE "user" (id text PRIMARY KEY, email text, name text);
     CREATE TABLE member (id text PRIMARY KEY, organization_id text, user_id text, role text);
     CREATE TABLE settings (id text PRIMARY KEY, user_id text, billing_usage_notifications_enabled boolean);
-    INSERT INTO "user" VALUES ('admin', 'admin@example.com', 'Admin');
   `)
   select.mockImplementation((fields) => database.select(fields))
   update.mockImplementation((table) => database.update(table))
@@ -70,8 +80,10 @@ beforeAll(async () => {
 beforeEach(async () => {
   mockSendEmail.mockClear()
   organizationId = generateId()
+  adminId = generateId()
   await connection`INSERT INTO organization (id) VALUES (${organizationId})`
-  await connection`INSERT INTO member VALUES (${generateId()}, ${organizationId}, 'admin', 'owner')`
+  await connection`INSERT INTO "user" VALUES (${adminId}, ${`${adminId}@example.com`}, 'Admin')`
+  await connection`INSERT INTO member VALUES (${generateId()}, ${organizationId}, ${adminId}, 'owner')`
 })
 
 afterAll(async () => {
@@ -97,12 +109,37 @@ describe('usage threshold email', () => {
     expect(mockSendEmail).toHaveBeenCalledTimes(2)
   })
 
-  it('re-arms both thresholds in the next billing period', async () => {
+  it('re-arms both thresholds whenever the billing period changes, even to an earlier one', async () => {
     await notify(100)
-    await notify(85, OCTOBER)
-    await notify(100, OCTOBER)
+    await notify(85, { periodStart: OCTOBER })
+    await notify(100, { periodStart: OCTOBER })
     await notify(85)
 
-    expect(mockSendEmail).toHaveBeenCalledTimes(3)
+    expect(mockSendEmail).toHaveBeenCalledTimes(4)
+  })
+
+  it('warns again at a raised limit after the old one was reached', async () => {
+    await notify(100)
+    await notify(100, { limit: 125 })
+    await notify(110, { limit: 125 })
+
+    expect(mockSendEmail).toHaveBeenCalledTimes(2)
+  })
+
+  it('keeps the claim for a later completion when nobody can be notified', async () => {
+    await setNotificationsEnabled(false)
+    await notify(90)
+    await setNotificationsEnabled(true)
+    await notify(90)
+
+    expect(mockSendEmail).toHaveBeenCalledTimes(1)
+  })
+
+  it('keeps the claim when a completion recorded no cost', async () => {
+    await notify(90, { costDelta: 0 })
+    expect(mockSendEmail).not.toHaveBeenCalled()
+
+    await notify(90)
+    expect(mockSendEmail).toHaveBeenCalledTimes(1)
   })
 })
diff --git a/apps/sim/lib/billing/core/usage.test.ts b/apps/sim/lib/billing/core/usage.test.ts
index ca214235db6..c0842c536b0 100644
--- a/apps/sim/lib/billing/core/usage.test.ts
+++ b/apps/sim/lib/billing/core/usage.test.ts
@@ -334,7 +334,8 @@ describe('maybeSendUsageThresholdEmail', () => {
   it('emails a paid personal account at 100% with the raise-your-limit template', async () => {
     await maybeSendUsageThresholdEmail({
       ...paidUser,
-      currentUsage: 20,
+      usageBefore: 19,
+      costDelta: 1,
     })
 
     expect(mockRenderUsageLimitReached).toHaveBeenCalledWith(
@@ -359,7 +360,8 @@ describe('maybeSendUsageThresholdEmail', () => {
       organizationId: 'org-1',
       workspaceId: 'ws-1',
       periodStart: new Date('2026-09-01T00:00:00.000Z'),
-      currentUsage: 500,
+      usageBefore: 499,
+      costDelta: 1,
       limit: 500,
     })
 
diff --git a/apps/sim/lib/billing/core/usage.ts b/apps/sim/lib/billing/core/usage.ts
index c6454beca86..bde888d1dc5 100644
--- a/apps/sim/lib/billing/core/usage.ts
+++ b/apps/sim/lib/billing/core/usage.ts
@@ -1,12 +1,14 @@
 import { db } from '@sim/db'
-import { member, organization, settings, user, userStats } from '@sim/db/schema'
+import { member, organization, userStats } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { isOrgAdminRole } from '@sim/platform-authz/workspace'
 import { generateId } from '@sim/utils/id'
 import { and, eq, isNull, sql } from 'drizzle-orm'
 import { getEffectiveBillingStatus } from '@/lib/billing/core/access'
 import { defaultBillingPeriod } from '@/lib/billing/core/billing-period'
-import { claimCreditsThreshold } from '@/lib/billing/core/limit-notifications'
+import {
+  claimCreditsThreshold,
+  resolveLimitEmailRecipients,
+} from '@/lib/billing/core/limit-notifications'
 import {
   getHighestPriorityPersonalSubscription,
   getHighestPrioritySubscription,
@@ -40,7 +42,6 @@ import { Decimal, toDecimal, toNumber } from '@/lib/billing/utils/decimal'
 import { isBillingEnabled } from '@/lib/core/config/env-flags'
 import { getBaseUrl } from '@/lib/core/utils/urls'
 import type { DbClient } from '@/lib/db/types'
-import { getEmailPreferences } from '@/lib/messaging/email/unsubscribe'
 import { APP_ENTRY_PATH } from '@/lib/navigation/paths'
 
 const logger = createLogger('UsageManagement')
@@ -688,18 +689,19 @@ export async function getEffectiveCurrentPeriodCost(
 
 /**
  * Send the usage threshold notification for the level usage is at: the 80% warning or the 100%
- * limit-reached email, each at most once per billing period.
+ * limit-reached email, each at most once per billing period and limit.
  * - Level-triggered: any caller at or above a threshold may send it, so it tolerates a usage read
- *   that lags the ledger, and the per-period claim ({@link claimCreditsThreshold}) is what keeps
- *   repeated or concurrent callers to one email.
+ *   that lags the ledger, and the claim ({@link claimCreditsThreshold}) is what keeps repeated or
+ *   concurrent callers to one email. A new period or a changed limit re-arms both thresholds.
  * - Skips when billing is disabled.
- * - Respects user-level notifications toggle and unsubscribe preferences.
+ * - Respects user-level notifications toggle and unsubscribe preferences, resolved before the
+ *   claim so an account with nobody to notify never consumes it.
  * - For organization plans, emails owners/admins who have notifications enabled.
  */
 export async function maybeSendUsageThresholdEmail(params: {
   scope: 'user' | 'organization'
   planName: string
-  /** Start of the billing period the usage belongs to; each period re-arms both thresholds. */
+  /** Start of the billing period the usage belongs to. */
   periodStart: Date
   userId?: string
   userEmail?: string
@@ -707,19 +709,32 @@ export async function maybeSendUsageThresholdEmail(params: {
   organizationId?: string
   /** Workspace the usage occurred in, used to build a live upgrade/billing link. */
   workspaceId?: string
-  currentUsage: number
+  /** Usage before this completion was recorded. */
+  usageBefore: number
+  /** Cost this completion recorded; one that recorded nothing cannot move usage. */
+  costDelta: number
   limit: number
 }): Promise<void> {
   try {
     if (!isBillingEnabled) return
-    if (params.limit <= 0 || params.currentUsage <= 0) return
+    if (params.limit <= 0 || params.costDelta <= 0) return
 
-    const percentUsed = (params.currentUsage / params.limit) * 100
+    const currentUsage = params.usageBefore + params.costDelta
+    const percentUsed = (currentUsage / params.limit) * 100
     const threshold = percentUsed >= 100 ? 100 : percentUsed >= 80 ? 80 : undefined
     if (threshold === undefined) return
     const stateId = params.scope === 'user' ? params.userId : params.organizationId
     if (!stateId) return
-    if (!(await claimCreditsThreshold(params.scope, stateId, params.periodStart, threshold))) return
+    const recipients = await resolveLimitEmailRecipients(params.scope, params)
+    if (recipients.length === 0) return
+    const claimed = await claimCreditsThreshold({
+      scope: params.scope,
+      id: stateId,
+      periodStart: params.periodStart,
+      limit: params.limit,
+      threshold,
+    })
+    if (!claimed) return
 
     const baseUrl = getBaseUrl()
     const isFreeUser = params.planName === 'Free'
@@ -738,57 +753,15 @@ export async function maybeSendUsageThresholdEmail(params: {
         ? `${baseUrl}/workspace/${params.workspaceId}/settings/billing`
         : `${baseUrl}/account/settings/billing`
 
-    /**
-     * Delivers to the account's notification recipients: the payer for personal
-     * scope, every org admin/owner for organization scope. Honors the per-user
-     * billing-notification toggle in both.
-     */
-    const deliverToScope = async (send: (email: string, name?: string) => Promise<void>) => {
-      if (params.scope === 'user' && params.userId && params.userEmail) {
-        const rows = await db
-          .select({ enabled: settings.billingUsageNotificationsEnabled })
-          .from(settings)
-          .where(eq(settings.userId, params.userId))
-          .limit(1)
-        if (rows.length > 0 && rows[0].enabled === false) return
-        await send(params.userEmail, params.userName)
-        return
-      }
-
-      if (params.scope === 'organization' && params.organizationId) {
-        const admins = await db
-          .select({
-            email: user.email,
-            name: user.name,
-            enabled: settings.billingUsageNotificationsEnabled,
-            role: member.role,
-          })
-          .from(member)
-          .innerJoin(user, eq(member.userId, user.id))
-          .leftJoin(settings, eq(settings.userId, member.userId))
-          .where(eq(member.organizationId, params.organizationId))
-
-        for (const a of admins) {
-          if (!isOrgAdminRole(a.role)) continue
-          if (a.enabled === false) continue
-          if (!a.email) continue
-          await send(a.email, a.name || undefined)
-        }
-      }
-    }
-
     if (threshold === 80 && !isFreeUser) {
       const ctaLink = billingSettingsLink
-      await deliverToScope(async (email, name) => {
-        const prefs = await getEmailPreferences(email)
-        if (prefs?.unsubscribeAll || prefs?.unsubscribeNotifications) return
-
+      for (const { email, name } of recipients) {
         const { renderUsageThresholdEmail, getEmailSubject, sendEmail } = await loadEmailDelivery()
         const html = await renderUsageThresholdEmail({
           userName: name,
           planName: params.planName,
           percentUsed: Math.round(percentUsed),
-          currentUsage: params.currentUsage,
+          currentUsage,
           limit: params.limit,
           ctaLink,
         })
@@ -799,20 +772,17 @@ export async function maybeSendUsageThresholdEmail(params: {
           html,
           emailType: 'notifications',
         })
-      })
+      }
     }
 
     if (threshold === 80 && isFreeUser) {
       const upgradeLink = upgradeCreditsLink
-      await deliverToScope(async (email, name) => {
-        const prefs = await getEmailPreferences(email)
-        if (prefs?.unsubscribeAll || prefs?.unsubscribeNotifications) return
-
+      for (const { email, name } of recipients) {
         const { renderFreeTierUpgradeEmail, getEmailSubject, sendEmail } = await loadEmailDelivery()
         const html = await renderFreeTierUpgradeEmail({
           userName: name,
           percentUsed: Math.round(percentUsed),
-          currentUsage: params.currentUsage,
+          currentUsage,
           limit: params.limit,
           upgradeLink,
         })
@@ -827,19 +797,16 @@ export async function maybeSendUsageThresholdEmail(params: {
         logger.info('Free tier upgrade email sent', {
           email,
           percentUsed: Math.round(percentUsed),
-          currentUsage: params.currentUsage,
+          currentUsage,
           limit: params.limit,
         })
-      })
+      }
     }
 
     if (threshold === 100) {
       const useFreeCopy = isFreeUser && params.scope === 'user'
 
-      await deliverToScope(async (email, name) => {
-        const prefs = await getEmailPreferences(email)
-        if (prefs?.unsubscribeAll || prefs?.unsubscribeNotifications) return
-
+      for (const { email, name } of recipients) {
         const {
           renderCreditsExhaustedEmail,
           renderUsageLimitReachedEmail,
@@ -857,7 +824,7 @@ export async function maybeSendUsageThresholdEmail(params: {
               userName: name,
               planName: params.planName,
               scope: params.scope,
-              currentUsage: params.currentUsage,
+              currentUsage,
               limit: params.limit,
               ctaLink: billingSettingsLink,
             })
@@ -875,10 +842,10 @@ export async function maybeSendUsageThresholdEmail(params: {
           email,
           scope: params.scope,
           planName: params.planName,
-          currentUsage: params.currentUsage,
+          currentUsage,
           limit: params.limit,
         })
-      })
+      }
     }
   } catch (error) {
     logger.error('Failed to send usage threshold email', {
diff --git a/apps/sim/lib/logs/execution/logger.ts b/apps/sim/lib/logs/execution/logger.ts
index 36e1d9db0a6..150a3096915 100644
--- a/apps/sim/lib/logs/execution/logger.ts
+++ b/apps/sim/lib/logs/execution/logger.ts
@@ -1357,7 +1357,7 @@ export class ExecutionLogger implements IExecutionLoggerService {
           userName: usr.name,
           planName: getDisplayPlanName(sub?.plan),
           periodStart: exactBillingContext.billingPeriod.start,
-          before: await checkResolvedUsageStatus(usr.id, sub),
+          before: await checkResolvedUsageStatus(usr.id, sub, exactBillingContext),
         }
       }
 
@@ -1384,7 +1384,8 @@ export class ExecutionLogger implements IExecutionLoggerService {
           planName: emailContext.planName,
           periodStart: emailContext.periodStart,
           workspaceId: updatedLog.workspaceId,
-          currentUsage: emailContext.before.currentUsage + costDelta,
+          usageBefore: emailContext.before.currentUsage,
+          costDelta,
           limit: emailContext.before.limit,
         })
       } else if (emailContext?.scope === 'organization') {
@@ -1394,7 +1395,8 @@ export class ExecutionLogger implements IExecutionLoggerService {
           planName: emailContext.planName,
           periodStart: emailContext.periodStart,
           workspaceId: updatedLog.workspaceId,
-          currentUsage: emailContext.orgUsageBefore + costDelta,
+          usageBefore: emailContext.orgUsageBefore,
+          costDelta,
           limit: emailContext.orgLimit,
         })
       }
diff --git a/packages/db/schema.ts b/packages/db/schema.ts
index 0a5df7328aa..3661e870eeb 100644
--- a/packages/db/schema.ts
+++ b/packages/db/schema.ts
@@ -1358,8 +1358,9 @@ export const userStats = pgTable('user_stats', {
    * `{ storage: 80, tables: 100 }`). Prevents re-spamming the same warning;
    * re-arms when usage drops back below the re-arm band. Keyed by limit
    * category ('storage' | 'tables'); seats live on `organization`. `credits`
-   * holds the billing period's start day times 1000 plus the threshold, so each
-   * period re-arms it without a reset (see `claimCreditsThreshold`).
+   * instead holds the threshold emailed for the billing period and limit in
+   * `creditsPeriod` (start day) and `creditsLimit` (cents), so a new period or a
+   * changed limit re-arms it without a reset (see `claimCreditsThreshold`).
    *
    * Dedup granularity is per billing account per category — intentionally NOT
    * per table, so a user hitting the row limit on several tables gets one
@@ -1722,7 +1723,8 @@ export const organization = pgTable('organization', {
    * Highest usage-limit threshold already emailed per category for this org
    * (e.g. `{ seats: 80, storage: 100 }`). Mirrors `user_stats.limitNotifications`
    * for org-scoped (pooled) limits. Re-arms when usage drops below the re-arm band;
-   * `credits` instead re-arms each billing period (see `claimCreditsThreshold`).
+   * `credits` instead re-arms with a new billing period or a changed limit (see
+   * `claimCreditsThreshold`).
    */
   limitNotifications: jsonb('limit_notifications')
     .$type<Record<string, number>>()

From 5b4bcf926efdf9c829776a09d5065c34b3e30240 Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Sat, 26 Sep 2026 10:53:17 -0700
Subject: [PATCH 3/4] improvement(billing): skip the shared usage read while
 Redis is not ready

---
 .../core/reporting-usage-cache.integration.ts | 48 ++++++++++----
 .../lib/billing/core/reporting-usage-cache.ts | 26 +++++---
 .../core/usage-threshold-email.integration.ts | 63 +++++++++++++++----
 apps/sim/lib/billing/core/usage.test.ts       | 29 ++++-----
 4 files changed, 122 insertions(+), 44 deletions(-)

diff --git a/apps/sim/lib/billing/core/reporting-usage-cache.integration.ts b/apps/sim/lib/billing/core/reporting-usage-cache.integration.ts
index a67ac000e80..3efd5100633 100644
--- a/apps/sim/lib/billing/core/reporting-usage-cache.integration.ts
+++ b/apps/sim/lib/billing/core/reporting-usage-cache.integration.ts
@@ -3,6 +3,8 @@
  * real Redis. Skipped without `TEST_REDIS_URL`. Each test uses a fresh payer, so the in-process
  * cache is always cold and every read models a new process.
  */
+
+import { type AddressInfo, createServer, type Socket } from 'node:net'
 import type { db } from '@sim/db'
 import * as schema from '@sim/db/schema'
 import { readTestDatabaseUrl, readTestRedisUrl } from '@sim/db/testing/test-infrastructure'
@@ -78,16 +80,14 @@ describe.runIf(Boolean(redisUrl))('shared reporting usage read', () => {
     await connection.end()
   })
 
-  it('serves a sum another process stored without summing the ledger', async () => {
+  it('serves a sum another process stored instead of the ledger', async () => {
     await redis.set(sharedKey(payer), '12.5', 'PX', 30_000)
 
     await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(12.5)
-    expect(transaction).not.toHaveBeenCalled()
   })
 
   it('sums the ledger exactly on a miss and stores the sum for other processes', async () => {
     await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
-    expect(transaction).toHaveBeenCalledTimes(1)
 
     await vi.waitFor(async () => expect(await redis.get(sharedKey(payer))).toBe('5.75'))
     const ttl = await redis.pttl(sharedKey(payer))
@@ -99,7 +99,6 @@ describe.runIf(Boolean(redisUrl))('shared reporting usage read', () => {
     await redis.set(sharedKey(payer), 'not-a-number', 'PX', 30_000)
 
     await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
-    expect(transaction).toHaveBeenCalledTimes(1)
   })
 
   it('never lets a slower, older sum replace one stored while it ran', async () => {
@@ -121,20 +120,45 @@ describe.runIf(Boolean(redisUrl))('shared reporting usage read', () => {
     await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
   })
 
-  it('sums the ledger promptly when Redis is unreachable', async () => {
-    const unreachable = new Redis('redis://127.0.0.1:1', {
-      enableOfflineQueue: true,
-      maxRetriesPerRequest: null,
-      retryStrategy: () => 1_000,
+  it('sums the ledger without issuing or queuing a command while Redis is not ready', async () => {
+    const notReady = new Redis(redisUrl!, { lazyConnect: true, maxRetriesPerRequest: 0 })
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(notReady)
+    try {
+      await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
+      expect(notReady.status).toBe('wait')
+
+      await notReady.connect()
+      await notReady.ping()
+      expect(await redis.get(sharedKey(payer))).toBeNull()
+    } finally {
+      notReady.disconnect()
+    }
+  })
+
+  it('sums the ledger promptly when a connected Redis stops answering', async () => {
+    const sockets = new Set<Socket>()
+    /** Completes the client's handshake, then never answers a read. */
+    const silent = createServer((socket) => {
+      sockets.add(socket)
+      socket.on('data', (data) => {
+        const text = data.toString()
+        if (/\bGET\b/i.test(text)) return
+        socket.write('+OK\r\n'.repeat(text.match(/^\*\d+\r\n/gm)?.length ?? 0))
+      })
     })
-    unreachable.on('error', () => undefined)
-    redisConfigMockFns.mockGetRedisClient.mockReturnValue(unreachable)
+    await new Promise<void>((resolve) => silent.listen(0, '127.0.0.1', resolve))
+    const { port } = silent.address() as AddressInfo
+    const hung = new Redis({ host: '127.0.0.1', port, lazyConnect: true, enableReadyCheck: false })
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(hung)
     try {
+      await hung.connect()
       const startedAt = Date.now()
       await expect(readSoftGateUsageCost(payer, REPORTING)).resolves.toBe(5.75)
       expect(Date.now() - startedAt).toBeLessThan(2_000)
     } finally {
-      unreachable.disconnect()
+      hung.disconnect()
+      for (const socket of sockets) socket.destroy()
+      silent.close()
     }
   })
 })
diff --git a/apps/sim/lib/billing/core/reporting-usage-cache.ts b/apps/sim/lib/billing/core/reporting-usage-cache.ts
index 571f9fbb799..440b69ca2dc 100644
--- a/apps/sim/lib/billing/core/reporting-usage-cache.ts
+++ b/apps/sim/lib/billing/core/reporting-usage-cache.ts
@@ -2,6 +2,7 @@ import { db } from '@sim/db'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import { randomInt } from '@sim/utils/random'
+import type Redis from 'ioredis'
 import { LRUCache } from 'lru-cache'
 import {
   type BillingEntity,
@@ -37,9 +38,8 @@ export const REPORTING_USAGE_CACHE_TTL_MS = 30_000
 const SHARED_TTL_JITTER_MS = 5_000
 
 /**
- * How long a read waits on Redis before summing the ledger instead. The shared client queues
- * commands while disconnected and has long timeouts, so without this a Redis outage would stall
- * every gate behind it rather than cost one sum.
+ * How long a read waits on a connected Redis before summing the ledger instead, so a socket that
+ * has silently stopped answering costs one sum rather than the shared client's long timeouts.
  */
 const SHARED_READ_TIMEOUT_MS = 250
 
@@ -53,14 +53,26 @@ function sharedReportingUsageKey(key: string): string {
   return `usage:reporting:${SHARED_KEY_VERSION}:${key}`
 }
 
+/**
+ * The shared client, only while its connection is ready. A disconnected client parks commands in
+ * its offline queue and replays them on reconnect, which would land a stale sum with a fresh
+ * expiry, and `NX` would then keep newer sums out; so while it is not ready, no command is issued
+ * at all and the ledger answers instead. Trigger.dev's `init` hook warms the connection, so it is
+ * ready by the time a task reads.
+ */
+function readyRedisClient(): Redis | null {
+  const redis = getRedisClient()
+  return redis?.status === 'ready' ? redis : null
+}
+
 /**
  * A sum another process stored, or `undefined` when there is none to use. Redis being absent,
- * slow, or failing, and a value that is not a non-negative number, are all misses: the caller
- * sums the ledger, so the cache can cost a read its latency but never its answer.
+ * not ready, slow, or failing, and a value that is not a non-negative number, are all misses:
+ * the caller sums the ledger, so the cache can cost a read its latency but never its answer.
  */
 async function readSharedReportingUsageCost(key: string): Promise<number | undefined> {
   try {
-    const redis = getRedisClient()
+    const redis = readyRedisClient()
     if (!redis) return undefined
     const stored = await withinDeadline(
       () => redis.get(sharedReportingUsageKey(key)),
@@ -89,7 +101,7 @@ function warnSharedWriteFailed(error: unknown): void {
  */
 function writeSharedReportingUsageCost(key: string, cost: number): void {
   try {
-    const redis = getRedisClient()
+    const redis = readyRedisClient()
     if (!redis) return
     const ttlMs = REPORTING_USAGE_CACHE_TTL_MS + randomInt(0, SHARED_TTL_JITTER_MS)
     redis
diff --git a/apps/sim/lib/billing/core/usage-threshold-email.integration.ts b/apps/sim/lib/billing/core/usage-threshold-email.integration.ts
index 6b47ed63553..7de3c0aa830 100644
--- a/apps/sim/lib/billing/core/usage-threshold-email.integration.ts
+++ b/apps/sim/lib/billing/core/usage-threshold-email.integration.ts
@@ -1,13 +1,13 @@
 /**
  * The level-triggered usage threshold email against real claim state in a disposable PostgreSQL
- * schema. Only delivery is stubbed: the mailer is the external boundary, and a sent email is the
- * outcome under test.
+ * schema. Only delivery is stubbed: the mailer is the external boundary, so the outcomes under
+ * test are the messages handed to it and the claim state left in the database.
  */
 import type { db } from '@sim/db'
 import * as schema from '@sim/db/schema'
 import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
 import { emailMailerMock, emailMailerMockFns } from '@sim/testing/mocks/email-mailer.mock'
-import { emailTemplatesMock } from '@sim/testing/mocks/email-templates.mock'
+import { emailTemplatesMock, emailTemplatesMockFns } from '@sim/testing/mocks/email-templates.mock'
 import { envFlagsMock, resetEnvFlagsMock, setEnvFlags } from '@sim/testing/mocks/env-flags.mock'
 import { generateId } from '@sim/utils/id'
 import { drizzle } from 'drizzle-orm/postgres-js'
@@ -59,6 +59,40 @@ function notify(
   })
 }
 
+/** Every message handed to the mailer so far, as who received which email. */
+function delivered(): { to: string; subject: string }[] {
+  return mockSendEmail.mock.calls.map(([message]) => ({ to: message.to, subject: message.subject }))
+}
+
+function warning(): { to: string; subject: string } {
+  return {
+    to: `${adminId}@example.com`,
+    subject: emailTemplatesMockFns.mockGetEmailSubject('usage-threshold'),
+  }
+}
+
+function reached(): { to: string; subject: string } {
+  return {
+    to: `${adminId}@example.com`,
+    subject: emailTemplatesMockFns.mockGetLimitEmailSubject('credits', 'reached'),
+  }
+}
+
+/** The organization's persisted threshold claims. */
+async function claims(): Promise<Record<string, number>> {
+  const [row] = await connection<{ limit_notifications: Record<string, number> | null }[]>`
+    SELECT limit_notifications FROM organization WHERE id = ${organizationId}`
+  return row.limit_notifications ?? {}
+}
+
+function claimOf(threshold: 80 | 100, periodStart = SEPTEMBER, limitCents = 10_000) {
+  return {
+    credits: threshold,
+    creditsPeriod: Math.floor(periodStart.getTime() / 86_400_000),
+    creditsLimit: limitCents,
+  }
+}
+
 async function setNotificationsEnabled(enabled: boolean): Promise<void> {
   await connection`INSERT INTO settings VALUES (${adminId}, ${adminId}, ${enabled})
     ON CONFLICT (id) DO UPDATE SET billing_usage_notifications_enabled = ${enabled}`
@@ -97,7 +131,8 @@ describe('usage threshold email', () => {
     await Promise.all([notify(85), notify(85), notify(86)])
     await notify(90)
 
-    expect(mockSendEmail).toHaveBeenCalledTimes(1)
+    expect(delivered()).toEqual([warning()])
+    expect(await claims()).toEqual(claimOf(80))
   })
 
   it('still sends the reached email after the warning, but never the warning after it', async () => {
@@ -106,7 +141,8 @@ describe('usage threshold email', () => {
     await notify(100)
     await notify(85)
 
-    expect(mockSendEmail).toHaveBeenCalledTimes(2)
+    expect(delivered()).toEqual([warning(), reached()])
+    expect(await claims()).toEqual(claimOf(100))
   })
 
   it('re-arms both thresholds whenever the billing period changes, even to an earlier one', async () => {
@@ -115,7 +151,8 @@ describe('usage threshold email', () => {
     await notify(100, { periodStart: OCTOBER })
     await notify(85)
 
-    expect(mockSendEmail).toHaveBeenCalledTimes(4)
+    expect(delivered()).toEqual([reached(), warning(), reached(), warning()])
+    expect(await claims()).toEqual(claimOf(80))
   })
 
   it('warns again at a raised limit after the old one was reached', async () => {
@@ -123,23 +160,27 @@ describe('usage threshold email', () => {
     await notify(100, { limit: 125 })
     await notify(110, { limit: 125 })
 
-    expect(mockSendEmail).toHaveBeenCalledTimes(2)
+    expect(delivered()).toEqual([reached(), warning()])
+    expect(await claims()).toEqual(claimOf(80, SEPTEMBER, 12_500))
   })
 
   it('keeps the claim for a later completion when nobody can be notified', async () => {
     await setNotificationsEnabled(false)
     await notify(90)
+    expect(delivered()).toEqual([])
+    expect(await claims()).toEqual({})
+
     await setNotificationsEnabled(true)
     await notify(90)
-
-    expect(mockSendEmail).toHaveBeenCalledTimes(1)
+    expect(delivered()).toEqual([warning()])
   })
 
   it('keeps the claim when a completion recorded no cost', async () => {
     await notify(90, { costDelta: 0 })
-    expect(mockSendEmail).not.toHaveBeenCalled()
+    expect(delivered()).toEqual([])
+    expect(await claims()).toEqual({})
 
     await notify(90)
-    expect(mockSendEmail).toHaveBeenCalledTimes(1)
+    expect(delivered()).toEqual([warning()])
   })
 })
diff --git a/apps/sim/lib/billing/core/usage.test.ts b/apps/sim/lib/billing/core/usage.test.ts
index c0842c536b0..6bd5d99c56e 100644
--- a/apps/sim/lib/billing/core/usage.test.ts
+++ b/apps/sim/lib/billing/core/usage.test.ts
@@ -331,6 +331,15 @@ describe('maybeSendUsageThresholdEmail', () => {
     resetEnvFlagsMock()
   })
 
+  /** Who received which email, identified by its subject and rendered template. */
+  function sentMessages() {
+    return mockSendEmail.mock.calls.map(([message]) => ({
+      to: message.to,
+      subject: message.subject,
+      html: message.html,
+    }))
+  }
+
   it('emails a paid personal account at 100% with the raise-your-limit template', async () => {
     await maybeSendUsageThresholdEmail({
       ...paidUser,
@@ -338,13 +347,9 @@ describe('maybeSendUsageThresholdEmail', () => {
       costDelta: 1,
     })
 
-    expect(mockRenderUsageLimitReached).toHaveBeenCalledWith(
-      expect.objectContaining({ scope: 'user', planName: 'Pro' })
-    )
-    expect(mockRenderCreditsExhausted).not.toHaveBeenCalled()
-    expect(mockRenderUsageThreshold).not.toHaveBeenCalled()
-    expect(mockGetLimitEmailSubject).toHaveBeenCalledWith('credits', 'reached')
-    expect(mockSendEmail).toHaveBeenCalledTimes(1)
+    expect(sentMessages()).toEqual([
+      { to: 'user-1@example.com', subject: 'Limit subject', html: '<html>reached</html>' },
+    ])
   })
 
   it('fans out to org admins at 100% and skips non-admin members', async () => {
@@ -365,12 +370,8 @@ describe('maybeSendUsageThresholdEmail', () => {
       limit: 500,
     })
 
-    expect(mockSendEmail).toHaveBeenCalledTimes(1)
-    expect(mockSendEmail).toHaveBeenCalledWith(
-      expect.objectContaining({ to: 'admin@example.com', emailType: 'notifications' })
-    )
-    expect(mockRenderUsageLimitReached).toHaveBeenCalledWith(
-      expect.objectContaining({ scope: 'organization' })
-    )
+    expect(sentMessages()).toEqual([
+      { to: 'admin@example.com', subject: 'Limit subject', html: '<html>reached</html>' },
+    ])
   })
 })

From 3f3716bbeb714c6b572a9d807433275409591516 Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Sat, 26 Sep 2026 13:16:50 -0700
Subject: [PATCH 4/4] improvement(billing): skip claimed usage thresholds
 before resolving recipients

---
 .../billing/core/limit-notifications.test.ts  | 13 ++--
 .../lib/billing/core/limit-notifications.ts   | 74 +++++++++++++------
 .../core/usage-threshold-email.integration.ts | 17 ++++-
 apps/sim/lib/billing/core/usage.test.ts       | 15 ++--
 apps/sim/lib/billing/core/usage.ts            | 15 ++--
 apps/sim/lib/messaging/email/mailer.test.ts   | 20 ++---
 .../src/mocks/email-unsubscribe.mock.ts       | 41 ++++++++++
 packages/testing/src/mocks/index.ts           |  4 +
 8 files changed, 149 insertions(+), 50 deletions(-)
 create mode 100644 packages/testing/src/mocks/email-unsubscribe.mock.ts

diff --git a/apps/sim/lib/billing/core/limit-notifications.test.ts b/apps/sim/lib/billing/core/limit-notifications.test.ts
index 9ac89cba167..e69ec58861d 100644
--- a/apps/sim/lib/billing/core/limit-notifications.test.ts
+++ b/apps/sim/lib/billing/core/limit-notifications.test.ts
@@ -1,25 +1,24 @@
 import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing/mocks/database.mock'
 import { emailMailerMock, emailMailerMockFns } from '@sim/testing/mocks/email-mailer.mock'
 import { emailTemplatesMock, emailTemplatesMockFns } from '@sim/testing/mocks/email-templates.mock'
+import {
+  emailUnsubscribeMock,
+  emailUnsubscribeMockFns,
+} from '@sim/testing/mocks/email-unsubscribe.mock'
 import { resetEnvFlagsMock, setEnvFlags } from '@sim/testing/mocks/env-flags.mock'
 import { schemaMock } from '@sim/testing/mocks/schema.mock'
 import { resetUrlsMock, urlsMockFns } from '@sim/testing/mocks/urls.mock'
 import { workspaceAuthzMock } from '@sim/testing/mocks/workspace-authz.mock'
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { getEmailPreferencesMock } = vi.hoisted(() => ({
-  getEmailPreferencesMock: vi.fn(() => Promise.resolve(null as unknown)),
-}))
-
 vi.mock('@/lib/messaging/email/mailer', () => emailMailerMock)
-vi.mock('@/lib/messaging/email/unsubscribe', () => ({
-  getEmailPreferences: getEmailPreferencesMock,
-}))
+vi.mock('@/lib/messaging/email/unsubscribe', () => emailUnsubscribeMock)
 vi.mock('@/components/emails', () => emailTemplatesMock)
 vi.mock('@sim/platform-authz/workspace', () => workspaceAuthzMock)
 
 import { maybeSendLimitThresholdEmail } from '@/lib/billing/core/limit-notifications'
 
+const getEmailPreferencesMock = emailUnsubscribeMockFns.mockGetEmailPreferences
 const sendEmailSpy = emailMailerMockFns.mockSendEmail
 const renderMock = emailTemplatesMockFns.mockRenderLimitThresholdEmail
 const subjectMock = emailTemplatesMockFns.mockGetLimitEmailSubject
diff --git a/apps/sim/lib/billing/core/limit-notifications.ts b/apps/sim/lib/billing/core/limit-notifications.ts
index eb6d48ba358..30c2238aee2 100644
--- a/apps/sim/lib/billing/core/limit-notifications.ts
+++ b/apps/sim/lib/billing/core/limit-notifications.ts
@@ -81,32 +81,64 @@ async function claimThreshold(
 
 const DAY_MS = 24 * 60 * 60 * 1000
 
-/**
- * Claim a credits threshold (80 or 100), returning whether THIS call won it. The claim is keyed
- * on the billing period and the limit: `credits` holds the highest threshold emailed while
- * `creditsPeriod` (the period's start day) and `creditsLimit` (the limit in cents) still match,
- * so a new period or a changed limit — in either direction — re-arms both thresholds with no
- * reset write, while within one a claim of 100 also retires 80, and never the reverse.
- */
-export function claimCreditsThreshold(params: {
+/** One account's credits threshold, keyed on its billing period and limit. */
+export interface CreditsThresholdClaim {
   scope: 'user' | 'organization'
   id: string
   periodStart: Date
   limit: number
   threshold: 80 | 100
-}): Promise<boolean> {
-  const { scope, id, threshold } = params
-  const periodDay = Math.floor(params.periodStart.getTime() / DAY_MS)
-  const limitCents = Math.round(params.limit * 100)
-  const column = scope === 'user' ? userStats.limitNotifications : organization.limitNotifications
-  const next = sql`coalesce(${column}, '{}'::jsonb) || jsonb_build_object('credits', ${threshold}::int, 'creditsPeriod', ${periodDay}::bigint, 'creditsLimit', ${limitCents}::bigint)`
-  const unclaimed = sql`not (
-    (${column} ->> 'creditsPeriod')::bigint is not distinct from ${periodDay}::bigint
-    and (${column} ->> 'creditsLimit')::bigint is not distinct from ${limitCents}::bigint
-    and coalesce((${column} ->> 'credits')::int, 0) >= ${threshold}::int
-  )`
-
-  return writeLimitNotifications(scope, id, next, unclaimed)
+}
+
+/**
+ * The stored claims for a credits threshold, and the condition under which it is still unclaimed:
+ * `credits` holds the highest threshold emailed while `creditsPeriod` (the period's start day) and
+ * `creditsLimit` (the limit in cents) still match, so a new period or a changed limit — in either
+ * direction — re-arms both thresholds with no reset write, while within one a claim of 100 also
+ * retires 80, and never the reverse.
+ */
+function creditsThresholdSql(claim: CreditsThresholdClaim) {
+  const periodDay = Math.floor(claim.periodStart.getTime() / DAY_MS)
+  const limitCents = Math.round(claim.limit * 100)
+  const column =
+    claim.scope === 'user' ? userStats.limitNotifications : organization.limitNotifications
+  return {
+    next: sql`coalesce(${column}, '{}'::jsonb) || jsonb_build_object('credits', ${claim.threshold}::int, 'creditsPeriod', ${periodDay}::bigint, 'creditsLimit', ${limitCents}::bigint)`,
+    unclaimed: sql<boolean>`not (
+      (${column} ->> 'creditsPeriod')::bigint is not distinct from ${periodDay}::bigint
+      and (${column} ->> 'creditsLimit')::bigint is not distinct from ${limitCents}::bigint
+      and coalesce((${column} ->> 'credits')::int, 0) >= ${claim.threshold}::int
+    )`,
+  }
+}
+
+/**
+ * Whether a credits threshold is still unclaimed, read with one indexed lookup of the account
+ * row. A cheap pre-check only: callers run it on every completion above a threshold, so once
+ * the threshold is claimed they stop there instead of resolving recipients. The atomic
+ * {@link claimCreditsThreshold} remains the real dedup.
+ */
+export async function isCreditsThresholdUnclaimed(claim: CreditsThresholdClaim): Promise<boolean> {
+  const { unclaimed } = creditsThresholdSql(claim)
+  const [row] =
+    claim.scope === 'user'
+      ? await db
+          .select({ unclaimed })
+          .from(userStats)
+          .where(eq(userStats.userId, claim.id))
+          .limit(1)
+      : await db
+          .select({ unclaimed })
+          .from(organization)
+          .where(eq(organization.id, claim.id))
+          .limit(1)
+  return row?.unclaimed === true
+}
+
+/** Claim a credits threshold, returning whether THIS call won it. */
+export function claimCreditsThreshold(claim: CreditsThresholdClaim): Promise<boolean> {
+  const { next, unclaimed } = creditsThresholdSql(claim)
+  return writeLimitNotifications(claim.scope, claim.id, next, unclaimed)
 }
 
 /** Re-arm a category (reset its stored threshold to 0) once usage falls back into the low band. */
diff --git a/apps/sim/lib/billing/core/usage-threshold-email.integration.ts b/apps/sim/lib/billing/core/usage-threshold-email.integration.ts
index 7de3c0aa830..c78c90a4c3c 100644
--- a/apps/sim/lib/billing/core/usage-threshold-email.integration.ts
+++ b/apps/sim/lib/billing/core/usage-threshold-email.integration.ts
@@ -8,6 +8,7 @@ import * as schema from '@sim/db/schema'
 import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
 import { emailMailerMock, emailMailerMockFns } from '@sim/testing/mocks/email-mailer.mock'
 import { emailTemplatesMock, emailTemplatesMockFns } from '@sim/testing/mocks/email-templates.mock'
+import { emailUnsubscribeMock } from '@sim/testing/mocks/email-unsubscribe.mock'
 import { envFlagsMock, resetEnvFlagsMock, setEnvFlags } from '@sim/testing/mocks/env-flags.mock'
 import { generateId } from '@sim/utils/id'
 import { drizzle } from 'drizzle-orm/postgres-js'
@@ -21,18 +22,21 @@ vi.mock('@sim/db', () => ({ db: { select, update }, dbReplica: { select } }))
 vi.mock('@/lib/core/config/env-flags', () => envFlagsMock)
 vi.mock('@/components/emails', () => emailTemplatesMock)
 vi.mock('@/lib/messaging/email/mailer', () => emailMailerMock)
-vi.mock('@/lib/messaging/email/unsubscribe', () => ({ getEmailPreferences: async () => null }))
+vi.mock('@/lib/messaging/email/unsubscribe', () => emailUnsubscribeMock)
 
 import { maybeSendUsageThresholdEmail } from '@/lib/billing/core/usage'
 
 const { mockSendEmail } = emailMailerMockFns
 
 const schemaName = `usage_threshold_${generateId().replaceAll('-', '')}`
+/** Every statement sent to the database, as the driver issued it. */
+const statements: string[] = []
 const connection = postgres(databaseUrl, {
   max: 4,
   prepare: false,
   connection: { search_path: schemaName },
   onnotice: () => undefined,
+  debug: (_connection, query) => statements.push(query),
 })
 const database = drizzle(connection, { schema }) as typeof db
 
@@ -164,6 +168,17 @@ describe('usage threshold email', () => {
     expect(await claims()).toEqual(claimOf(80, SEPTEMBER, 12_500))
   })
 
+  it('stops at one account read once the threshold is claimed', async () => {
+    await notify(90)
+    statements.length = 0
+
+    await notify(95)
+
+    expect(statements).toHaveLength(1)
+    expect(statements[0]).toMatch(/^select [\s\S]* from "organization" where/i)
+    expect(delivered()).toEqual([warning()])
+  })
+
   it('keeps the claim for a later completion when nobody can be notified', async () => {
     await setNotificationsEnabled(false)
     await notify(90)
diff --git a/apps/sim/lib/billing/core/usage.test.ts b/apps/sim/lib/billing/core/usage.test.ts
index 6bd5d99c56e..13ccd8a0572 100644
--- a/apps/sim/lib/billing/core/usage.test.ts
+++ b/apps/sim/lib/billing/core/usage.test.ts
@@ -16,6 +16,10 @@ import { billingUsageLogMock } from '@sim/testing/mocks/billing-usage-log.mock'
 import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing/mocks/database.mock'
 import { emailMailerMock, emailMailerMockFns } from '@sim/testing/mocks/email-mailer.mock'
 import { emailTemplatesMock, emailTemplatesMockFns } from '@sim/testing/mocks/email-templates.mock'
+import {
+  emailUnsubscribeMock,
+  emailUnsubscribeMockFns,
+} from '@sim/testing/mocks/email-unsubscribe.mock'
 import { resetEnvFlagsMock, setEnvFlags } from '@sim/testing/mocks/env-flags.mock'
 import { schemaMock } from '@sim/testing/mocks/schema.mock'
 import { workspaceAuthzMock, workspaceAuthzMockFns } from '@sim/testing/mocks/workspace-authz.mock'
@@ -25,10 +29,6 @@ afterAll(() => {
   resetDbChainMock()
 })
 
-const { mockGetEmailPreferences } = vi.hoisted(() => ({
-  mockGetEmailPreferences: vi.fn(() => Promise.resolve(null as unknown)),
-}))
-
 vi.mock('@/lib/billing/subscriptions/utils', () => billingSubscriptionUtilsMock)
 
 vi.mock('@/lib/billing/core/plan', () => billingPlanMock)
@@ -45,9 +45,7 @@ vi.mock('@/components/emails', () => emailTemplatesMock)
 
 vi.mock('@/lib/messaging/email/mailer', () => emailMailerMock)
 
-vi.mock('@/lib/messaging/email/unsubscribe', () => ({
-  getEmailPreferences: mockGetEmailPreferences,
-}))
+vi.mock('@/lib/messaging/email/unsubscribe', () => emailUnsubscribeMock)
 
 vi.mock('@sim/platform-authz/workspace', () => workspaceAuthzMock)
 
@@ -57,6 +55,7 @@ import {
   syncUsageLimitsFromSubscription,
 } from '@/lib/billing/core/usage'
 
+const { mockGetEmailPreferences } = emailUnsubscribeMockFns
 const mockIsOrgAdminRole = workspaceAuthzMockFns.mockIsOrgAdminRole
 const {
   mockGetFreeTierLimit,
@@ -341,6 +340,7 @@ describe('maybeSendUsageThresholdEmail', () => {
   }
 
   it('emails a paid personal account at 100% with the raise-your-limit template', async () => {
+    queueTableRows(schemaMock.userStats, [{ unclaimed: true }])
     await maybeSendUsageThresholdEmail({
       ...paidUser,
       usageBefore: 19,
@@ -354,6 +354,7 @@ describe('maybeSendUsageThresholdEmail', () => {
 
   it('fans out to org admins at 100% and skips non-admin members', async () => {
     mockIsOrgAdminRole.mockImplementation((role: unknown) => role === 'admin')
+    queueTableRows(schemaMock.organization, [{ unclaimed: true }])
     queueTableRows(schemaMock.member, [
       { email: 'admin@example.com', name: 'Admin', enabled: null, role: 'admin' },
       { email: 'member@example.com', name: 'Member', enabled: null, role: 'member' },
diff --git a/apps/sim/lib/billing/core/usage.ts b/apps/sim/lib/billing/core/usage.ts
index bde888d1dc5..8052798a855 100644
--- a/apps/sim/lib/billing/core/usage.ts
+++ b/apps/sim/lib/billing/core/usage.ts
@@ -6,7 +6,9 @@ import { and, eq, isNull, sql } from 'drizzle-orm'
 import { getEffectiveBillingStatus } from '@/lib/billing/core/access'
 import { defaultBillingPeriod } from '@/lib/billing/core/billing-period'
 import {
+  type CreditsThresholdClaim,
   claimCreditsThreshold,
+  isCreditsThresholdUnclaimed,
   resolveLimitEmailRecipients,
 } from '@/lib/billing/core/limit-notifications'
 import {
@@ -694,6 +696,8 @@ export async function getEffectiveCurrentPeriodCost(
  *   that lags the ledger, and the claim ({@link claimCreditsThreshold}) is what keeps repeated or
  *   concurrent callers to one email. A new period or a changed limit re-arms both thresholds.
  * - Skips when billing is disabled.
+ * - Returns after one indexed read once the threshold is claimed, so the completions that follow
+ *   in the period cost no recipient lookup or write.
  * - Respects user-level notifications toggle and unsubscribe preferences, resolved before the
  *   claim so an account with nobody to notify never consumes it.
  * - For organization plans, emails owners/admins who have notifications enabled.
@@ -725,16 +729,17 @@ export async function maybeSendUsageThresholdEmail(params: {
     if (threshold === undefined) return
     const stateId = params.scope === 'user' ? params.userId : params.organizationId
     if (!stateId) return
-    const recipients = await resolveLimitEmailRecipients(params.scope, params)
-    if (recipients.length === 0) return
-    const claimed = await claimCreditsThreshold({
+    const claim: CreditsThresholdClaim = {
       scope: params.scope,
       id: stateId,
       periodStart: params.periodStart,
       limit: params.limit,
       threshold,
-    })
-    if (!claimed) return
+    }
+    if (!(await isCreditsThresholdUnclaimed(claim))) return
+    const recipients = await resolveLimitEmailRecipients(params.scope, params)
+    if (recipients.length === 0) return
+    if (!(await claimCreditsThreshold(claim))) return
 
     const baseUrl = getBaseUrl()
     const isFreeUser = params.planName === 'Free'
diff --git a/apps/sim/lib/messaging/email/mailer.test.ts b/apps/sim/lib/messaging/email/mailer.test.ts
index f29f44c915b..f9263cf3da6 100644
--- a/apps/sim/lib/messaging/email/mailer.test.ts
+++ b/apps/sim/lib/messaging/email/mailer.test.ts
@@ -1,3 +1,7 @@
+import {
+  emailUnsubscribeMock,
+  emailUnsubscribeMockFns,
+} from '@sim/testing/mocks/email-unsubscribe.mock'
 import { resetEnvMock } from '@sim/testing/mocks/env.mock'
 import { resetUrlsMock, urlsMockFns } from '@sim/testing/mocks/urls.mock'
 import { afterAll, beforeEach, describe, expect, it, type Mock, vi } from 'vitest'
@@ -43,10 +47,7 @@ vi.mock('@azure/communication-email', () => {
   }
 })
 
-vi.mock('@/lib/messaging/email/unsubscribe', () => ({
-  isUnsubscribed: vi.fn(),
-  generateUnsubscribeToken: vi.fn(),
-}))
+vi.mock('@/lib/messaging/email/unsubscribe', () => emailUnsubscribeMock)
 
 vi.mock('@/lib/auth/access-control', () => ({
   getAccessControlConfig: vi.fn().mockResolvedValue({
@@ -68,7 +69,8 @@ vi.mock('@/lib/messaging/email/utils', () => ({
 
 import { isEmailBlockedByAccessControl } from '@/lib/auth/access-control'
 import { sendEmail } from './mailer'
-import { generateUnsubscribeToken, isUnsubscribed } from './unsubscribe'
+
+const { mockGenerateUnsubscribeToken, mockIsUnsubscribed } = emailUnsubscribeMockFns
 
 urlsMockFns.mockGetEmailDomain.mockReturnValue('sim.ai')
 urlsMockFns.mockGetBaseUrl.mockReturnValue('https://test.sim.ai')
@@ -87,8 +89,8 @@ describe('mailer', () => {
 
   beforeEach(() => {
     ;(isEmailBlockedByAccessControl as Mock).mockReturnValue(false)
-    ;(isUnsubscribed as Mock).mockResolvedValue(false)
-    ;(generateUnsubscribeToken as Mock).mockReturnValue('mock-token-123')
+    mockIsUnsubscribed.mockResolvedValue(false)
+    mockGenerateUnsubscribeToken.mockReturnValue('mock-token-123')
 
     mockSend.mockResolvedValue({
       data: { id: 'test-email-id' },
@@ -112,7 +114,7 @@ describe('mailer', () => {
 
   describe('sendEmail', () => {
     it('should skip sending if user has unsubscribed', async () => {
-      ;(isUnsubscribed as Mock).mockResolvedValue(true)
+      mockIsUnsubscribed.mockResolvedValue(true)
 
       const result = await sendEmail({
         ...testEmailOptions,
@@ -164,7 +166,7 @@ describe('mailer', () => {
       expect(result.message).toBe('Email skipped (recipient on access-control ban list)')
       expect(result.data).toEqual({ id: 'skipped-banned' })
       expect(mockSend).not.toHaveBeenCalled()
-      expect(isUnsubscribed).not.toHaveBeenCalled()
+      expect(mockIsUnsubscribed).not.toHaveBeenCalled()
     })
 
     it('should drop only the banned recipients from a multi-recipient send', async () => {
diff --git a/packages/testing/src/mocks/email-unsubscribe.mock.ts b/packages/testing/src/mocks/email-unsubscribe.mock.ts
new file mode 100644
index 00000000000..4a1852a0dc7
--- /dev/null
+++ b/packages/testing/src/mocks/email-unsubscribe.mock.ts
@@ -0,0 +1,41 @@
+import { vi } from 'vitest'
+
+/**
+ * Controllable mock functions for `@/lib/messaging/email/unsubscribe`. `getEmailPreferences`
+ * resolves `null` (no stored preferences, so nobody is unsubscribed); every other function is a
+ * bare `vi.fn()`.
+ *
+ * @example
+ * ```ts
+ * import { emailUnsubscribeMockFns } from '@sim/testing/mocks/email-unsubscribe.mock'
+ *
+ * emailUnsubscribeMockFns.mockGetEmailPreferences.mockResolvedValue({ unsubscribeAll: true })
+ * ```
+ */
+export const emailUnsubscribeMockFns = {
+  mockGenerateUnsubscribeToken: vi.fn(),
+  mockVerifyUnsubscribeToken: vi.fn(),
+  mockIsTransactionalEmail: vi.fn(),
+  mockGetEmailPreferences: vi.fn(async (): Promise<unknown> => null),
+  mockUpdateEmailPreferences: vi.fn(),
+  mockIsUnsubscribed: vi.fn(),
+  mockUnsubscribeFromAll: vi.fn(),
+}
+
+/**
+ * Static mock module for `@/lib/messaging/email/unsubscribe`.
+ *
+ * @example
+ * ```ts
+ * vi.mock('@/lib/messaging/email/unsubscribe', () => emailUnsubscribeMock)
+ * ```
+ */
+export const emailUnsubscribeMock = {
+  generateUnsubscribeToken: emailUnsubscribeMockFns.mockGenerateUnsubscribeToken,
+  verifyUnsubscribeToken: emailUnsubscribeMockFns.mockVerifyUnsubscribeToken,
+  isTransactionalEmail: emailUnsubscribeMockFns.mockIsTransactionalEmail,
+  getEmailPreferences: emailUnsubscribeMockFns.mockGetEmailPreferences,
+  updateEmailPreferences: emailUnsubscribeMockFns.mockUpdateEmailPreferences,
+  isUnsubscribed: emailUnsubscribeMockFns.mockIsUnsubscribed,
+  unsubscribeFromAll: emailUnsubscribeMockFns.mockUnsubscribeFromAll,
+}
diff --git a/packages/testing/src/mocks/index.ts b/packages/testing/src/mocks/index.ts
index ab1fb27bcd0..7c2a8990833 100644
--- a/packages/testing/src/mocks/index.ts
+++ b/packages/testing/src/mocks/index.ts
@@ -255,6 +255,10 @@ export {
   emailTemplatesMock,
   emailTemplatesMockFns,
 } from './email-templates.mock'
+export {
+  emailUnsubscribeMock,
+  emailUnsubscribeMockFns,
+} from './email-unsubscribe.mock'
 export {
   embeddingsMock,
   embeddingsMockFns,