improvement(billing): skip locked rows in the identity purge, share the auth-request lookup

Concurrent account deletions no longer wait on each other's purge batch.
syncUsageLimitsFromSubscription reads in parallel again, and email
canonicalization reuses normalizeEmail.

Author: waleedlatif1

apps/sim/lib/billing/core/usage.ts

@@ -191,6 +191,13 @@ function calculateOrgUsageLimit(
  */
 const usersCreatedInAuthRequest = new WeakMap<object, string>()
 
+function isUserCreatedInAuthRequest(
+  userId: string,
+  authRequest: object | null | undefined
+): boolean {
+  return authRequest ? usersCreatedInAuthRequest.get(authRequest) === userId : false
+}
+
 /**
  * Handle new user setup when they join the platform.
  * Creates the userStats record with the free allowance, which is zero when the
@@ -253,7 +260,7 @@ export async function handleNewSignInAccount(
   },
   authRequest: object | null | undefined
 ): Promise<void> {
-  if (!authRequest || usersCreatedInAuthRequest.get(authRequest) !== userId) return
+  if (!isUserCreatedInAuthRequest(userId, authRequest)) return
   if (!(await isDeletedAccountIdentity({ oauthSubjects: [{ providerId, accountId }] }))) return
   await forfeitFreeTierAllowance(userId)
 }
@@ -289,9 +296,7 @@ export async function ensureUserStatsExists(
   if (created.length === 0) return
 
   try {
-    const includeSignIns = authRequest
-      ? usersCreatedInAuthRequest.get(authRequest) === userId
-      : false
+    const includeSignIns = isUserCreatedInAuthRequest(userId, authRequest)
     if (await userHasDeletedAccountIdentity(userId, { includeSignIns })) {
       await forfeitFreeTierAllowance(userId)
     }
@@ -705,15 +710,10 @@ export async function syncUsageLimitsFromSubscription(
   userId: string,
   executor: DbOrTx = db
 ): Promise<void> {
-  const subscription = await getHighestPriorityPersonalSubscription(userId, {
-    onError: 'throw',
-    executor,
-  })
-  const currentUserStats = await executor
-    .select()
-    .from(userStats)
-    .where(eq(userStats.userId, userId))
-    .limit(1)
+  const [subscription, currentUserStats] = await Promise.all([
+    getHighestPriorityPersonalSubscription(userId, { onError: 'throw', executor }),
+    executor.select().from(userStats).where(eq(userStats.userId, userId)).limit(1),
+  ])
 
   if (currentUserStats.length === 0) {
     throw new Error(`User stats not found for userId: ${userId}`)

apps/sim/lib/users/deleted-identity.ts

@@ -3,6 +3,7 @@ import { account, deletedAccountIdentity, user } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { hmacSha256Hex } from '@sim/security/hmac'
 import { getErrorMessage } from '@sim/utils/errors'
+import { normalizeEmail } from '@sim/utils/string'
 import { and, eq, gt, inArray, lte, sql } from 'drizzle-orm'
 import { env } from '@/lib/core/config/env'
 import { isBillingEnabled } from '@/lib/core/config/env-flags'
@@ -102,7 +103,7 @@ function getIdentityKey(): string {
  * `googlemail.com` alias.
  */
 function canonicalizeEmail(email: string): string | null {
-  const normalized = email.trim().toLowerCase()
+  const normalized = normalizeEmail(email)
   const at = normalized.lastIndexOf('@')
   if (at <= 0 || at === normalized.length - 1) return null
   let local = normalized.slice(0, at)
@@ -164,18 +165,18 @@ export async function recordDeletedAccountIdentities(tx: DbOrTx, userId: string)
       const hashes = identityHashes(await loadSignInIdentity(savepoint, userId))
       if (hashes.length === 0) return
 
-      await savepoint
-        .delete(deletedAccountIdentity)
-        .where(
-          inArray(
-            deletedAccountIdentity.identityHash,
-            savepoint
-              .select({ identityHash: deletedAccountIdentity.identityHash })
-              .from(deletedAccountIdentity)
-              .where(lte(deletedAccountIdentity.expiresAt, sql`now()`))
-              .limit(EXPIRED_PURGE_BATCH)
-          )
+      await savepoint.delete(deletedAccountIdentity).where(
+        inArray(
+          deletedAccountIdentity.identityHash,
+          savepoint
+            .select({ identityHash: deletedAccountIdentity.identityHash })
+            .from(deletedAccountIdentity)
+            .where(lte(deletedAccountIdentity.expiresAt, sql`now()`))
+            .limit(EXPIRED_PURGE_BATCH)
+            // Concurrent deletions skip each other's purge batch instead of waiting on its commit.
+            .for('update', { skipLocked: true })
         )
+      )
 
       const expiresAt = new Date(Date.now() + DELETED_IDENTITY_RETENTION_MS)
       await savepoint