refactor(db): require a transaction for advisory lock helpers

apps/sim/ee/access-requests/lib/application/authorized-use-case.ts

@@ -8,7 +8,7 @@ import {
 import type { OperationUseCase } from '@/lib/core/application/operation'
 import { requireAllowedWorkspacePrincipal } from '@/lib/core/application/workspace-authorization'
 import { runWithOutboundOrganization } from '@/lib/core/network/context.server'
-import type { DbOrTx } from '@/lib/db/types'
+import type { DbClient, DbOrTx, DbTransaction } from '@/lib/db/types'
 import {
   type AccessRequestContext,
   authorizeAccessRequestScope,
@@ -25,25 +25,40 @@ interface AccessRequestPreparationArgs<I> {
   context: AccessRequestContext
 }
 
-interface AccessRequestUseCaseArgs<I> extends AccessRequestPreparationArgs<I> {
-  executor: DbOrTx
+/** A mutation executes inside the funnel's transaction; a read runs on the pool-level client. */
+interface AccessRequestUseCaseArgs<I, E extends DbOrTx = DbOrTx>
+  extends AccessRequestPreparationArgs<I> {
+  executor: E
 }
 
 interface AccessRequestUseCaseDefinition<I, R> {
   operation: AccessRequestOperation
   scope(input: I): AccessRequestScope
-  mutation?: boolean
   projectAudit?(args: AccessRequestUseCaseArgs<I> & { result: R }): WorkspaceUseCaseAuditEntry[]
 }
 
-interface PreparedAccessRequestUseCase<I, R, P> extends AccessRequestUseCaseDefinition<I, R> {
+interface PreparedAccessRequestUseCase<I, R, P, E extends DbOrTx>
+  extends AccessRequestUseCaseDefinition<I, R> {
   prepare(args: AccessRequestPreparationArgs<I>): Promise<P>
-  execute(args: AccessRequestUseCaseArgs<I> & { prepared: P }): Promise<R>
+  execute(args: AccessRequestUseCaseArgs<I, E> & { prepared: P }): Promise<R>
 }
 
-interface UnpreparedAccessRequestUseCase<I, R> extends AccessRequestUseCaseDefinition<I, R> {
+interface UnpreparedAccessRequestUseCase<I, R, E extends DbOrTx>
+  extends AccessRequestUseCaseDefinition<I, R> {
   prepare?: never
-  execute(args: AccessRequestUseCaseArgs<I> & { prepared: undefined }): Promise<R>
+  execute(args: AccessRequestUseCaseArgs<I, E> & { prepared: undefined }): Promise<R>
+}
+
+type AccessRequestUseCase<I, R, P, E extends DbOrTx> =
+  | PreparedAccessRequestUseCase<I, R, P, E>
+  | UnpreparedAccessRequestUseCase<I, R, E>
+
+type MutationAccessRequestUseCase<I, R, P> = AccessRequestUseCase<I, R, P, DbTransaction> & {
+  mutation: true
+}
+
+type ReadAccessRequestUseCase<I, R, P> = AccessRequestUseCase<I, R, P, DbClient> & {
+  mutation?: false
 }
 
 function requireAccessRequestPrincipal(
@@ -53,15 +68,33 @@ function requireAccessRequestPrincipal(
   requireAllowedWorkspacePrincipal(principal, operation)
 }
 
+/** Runs preparation before any transaction opens and binds its result to `execute`. */
+async function prepareExecution<I, R, P, E extends DbOrTx>(
+  definition: AccessRequestUseCase<I, R, P, E>,
+  args: AccessRequestPreparationArgs<I>
+): Promise<(args: AccessRequestUseCaseArgs<I, E>) => Promise<R>> {
+  if (definition.prepare) {
+    const prepared = await definition.prepare(args)
+    const executePrepared = definition.execute
+    return (executeArgs) => executePrepared({ ...executeArgs, prepared })
+  }
+  const executeUnprepared = definition.execute
+  return (executeArgs) => executeUnprepared({ ...executeArgs, prepared: undefined })
+}
+
 export function defineAuthorizedAccessRequestUseCase<I, R, P>(
-  definition: PreparedAccessRequestUseCase<I, R, P>
+  definition:
+    | (PreparedAccessRequestUseCase<I, R, P, DbTransaction> & { mutation: true })
+    | (PreparedAccessRequestUseCase<I, R, P, DbClient> & { mutation?: false })
 ): OperationUseCase<AccessRequestOperation, I, R>
 export function defineAuthorizedAccessRequestUseCase<I, R>(
-  definition: UnpreparedAccessRequestUseCase<I, R>
+  definition:
+    | (UnpreparedAccessRequestUseCase<I, R, DbTransaction> & { mutation: true })
+    | (UnpreparedAccessRequestUseCase<I, R, DbClient> & { mutation?: false })
 ): OperationUseCase<AccessRequestOperation, I, R>
 /** Shared human-credential funnel; preparation finishes before any transaction acquires locks. */
 export function defineAuthorizedAccessRequestUseCase<I, R, P = undefined>(
-  definition: PreparedAccessRequestUseCase<I, R, P> | UnpreparedAccessRequestUseCase<I, R>
+  definition: MutationAccessRequestUseCase<I, R, P> | ReadAccessRequestUseCase<I, R, P>
 ): OperationUseCase<AccessRequestOperation, I, R> {
   return {
     operation: definition.operation,
@@ -74,32 +107,29 @@ export function defineAuthorizedAccessRequestUseCase<I, R, P = undefined>(
       const scope = definition.scope(input)
       const initial = await authorizeAccessRequestScope(principal, definition.operation, scope)
       return runWithOutboundOrganization(initial.organizationId, async () => {
-        let execute: (args: AccessRequestUseCaseArgs<I>) => Promise<R>
-        if (definition.prepare) {
-          const prepared = await definition.prepare({ principal, input, context: initial })
-          const executePrepared = definition.execute
-          execute = (args) => executePrepared({ ...args, prepared })
+        const preparation = { principal, input, context: initial }
+        let context = initial
+        let result: R
+        if (definition.mutation) {
+          const execute = await prepareExecution(definition, preparation)
+          result = await db.transaction(async (executor) => {
+            if (initial.organizationId) {
+              await acquireOrganizationMutationLock(executor, initial.organizationId)
+            }
+            context = await authorizeAccessRequestScope(
+              principal,
+              definition.operation,
+              scope,
+              executor,
+              true,
+              initial
+            )
+            return execute({ principal, input, context, executor })
+          })
         } else {
-          const executeUnprepared = definition.execute
-          execute = (args) => executeUnprepared({ ...args, prepared: undefined })
+          const execute = await prepareExecution(definition, preparation)
+          result = await execute({ principal, input, context, executor: db })
         }
-        let context = initial
-        const result = definition.mutation
-          ? await db.transaction(async (executor) => {
-              if (initial.organizationId) {
-                await acquireOrganizationMutationLock(executor, initial.organizationId)
-              }
-              context = await authorizeAccessRequestScope(
-                principal,
-                definition.operation,
-                scope,
-                executor,
-                true,
-                initial
-              )
-              return execute({ principal, input, context, executor })
-            })
-          : await execute({ principal, input, context, executor: db })
         if (definition.projectAudit) {
           recordProjectedUseCaseAuditEntries(
             definition.operation,

apps/sim/ee/access-requests/lib/application/review.ts

@@ -8,7 +8,7 @@ import { setOrgMemberUsageLimit } from '@/lib/billing/organizations/member-limit
 import type { WorkspaceUseCaseAuditEntry } from '@/lib/core/application/authorized-workspace-use-case'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
-import type { DbOrTx } from '@/lib/db/types'
+import type { DbTransaction } from '@/lib/db/types'
 import { acquirePermissionGroupOrgLock } from '@/lib/permission-groups/locks'
 import { loadAccessRequestMembership } from '@/ee/access-requests/lib/application/authorization'
 import { defineAuthorizedAccessRequestUseCase } from '@/ee/access-requests/lib/application/authorized-use-case'
@@ -52,13 +52,17 @@ const organizationScope = (input: ReviewInput): AccessRequestScope => ({
   organizationId: input.organizationId,
 })
 
-/** Checks the requester's present scope before inspecting or modifying any governing policy. */
+/**
+ * Checks the requester's present scope before inspecting or modifying any governing policy.
+ * Given `lockingTx`, it locks the scope rows and the governing policy in that transaction.
+ */
 async function loadReviewPreview(
-  executor: DbOrTx,
   row: StoredAccessRequest,
   prepared: PreparedAccessRequestPolicy | null,
-  forUpdate = false
+  lockingTx?: DbTransaction
 ) {
+  const executor = lockingTx ?? db
+  const forUpdate = Boolean(lockingTx)
   const request = await presentAccessRequest(executor, row)
   if (row.status === 'fulfilled' && row.decision) {
     const snapshot = storedAccessRequestDecisionSchema.parse(row.decision)
@@ -118,8 +122,8 @@ async function loadReviewPreview(
     membershipId: membership?.membershipId ?? '',
     role: membership?.role ?? ('read' as const),
   }
-  if (forUpdate)
-    await acquirePermissionGroupOrgLock(executor, row.organizationId, {
+  if (lockingTx)
+    await acquirePermissionGroupOrgLock(lockingTx, row.organizationId, {
       lockTimeoutAlreadyBounded: true,
     })
   const catalog = prepared.catalog
@@ -233,7 +237,7 @@ export const previewAccessRequest = defineAuthorizedAccessRequestUseCase({
   },
   async execute({ input, executor, prepared }) {
     const row = await loadStoredAccessRequest(executor, input.organizationId, input.requestId)
-    return (await loadReviewPreview(executor, row, prepared)).preview
+    return (await loadReviewPreview(row, prepared)).preview
   },
 })
 
@@ -275,7 +279,7 @@ export const resolveAccessRequest = defineAuthorizedAccessRequestUseCase({
     }
     if (!prepared)
       throw new OrchestrationError('internal', 'Request preview preparation is missing')
-    const { preview, policy } = await loadReviewPreview(executor, row, prepared, true)
+    const { preview, policy } = await loadReviewPreview(row, prepared, executor)
     if (!preview.canApply)
       throw new OrchestrationError(
         'conflict',

apps/sim/ee/scim/lib/application/admin/mappings.ts

@@ -13,7 +13,7 @@ import { and, count, eq, sql } from 'drizzle-orm'
 import type { ScimGroupMappingView } from '@/lib/api/contracts/organization-scim'
 import { acquireOrganizationMutationLock } from '@/lib/billing/organizations/membership'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
-import type { DbOrTx } from '@/lib/db/types'
+import type { DbTransaction } from '@/lib/db/types'
 import { acquirePermissionGroupOrgLock } from '@/lib/permission-groups/locks'
 import {
   assertWorkspaceInOrganization,
@@ -147,7 +147,7 @@ async function requireGroup(connectionId: string, groupId: string) {
 }
 
 async function assertPermissionGroupTarget(
-  tx: DbOrTx,
+  tx: DbTransaction,
   organizationId: string,
   permissionGroupId: string
 ) {

apps/sim/ee/scim/lib/application/groups/manage-groups.ts

@@ -4,7 +4,7 @@ import { scimGroup } from '@sim/db/schema'
 import { and, eq, ne } from 'drizzle-orm'
 import type { ScimPatchOperation } from '@/lib/api/contracts/scim'
 import { acquireOrganizationMutationLock } from '@/lib/billing/organizations/membership'
-import type { DbOrTx } from '@/lib/db/types'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 import {
   defineAuthorizedScimUseCase,
   type ScimUseCaseArgs,
@@ -54,7 +54,7 @@ import {
  */
 function withGroupWrite<T>(
   context: ScimUseCaseContext,
-  work: (tx: DbOrTx) => Promise<T>
+  work: (tx: DbTransaction) => Promise<T>
 ): Promise<T> {
   return db.transaction(async (tx) => {
     await acquireOrganizationMutationLock(tx, context.organizationId)

apps/sim/ee/scim/lib/application/users/update-user.ts

@@ -4,7 +4,7 @@ import type { ScimUserAttributes } from '@sim/db/schema'
 import { normalizeEmail } from '@sim/utils/string'
 import type { ScimPatchOperation } from '@/lib/api/contracts/scim'
 import { acquireOrganizationUserMutationLocks } from '@/lib/billing/organizations/membership'
-import type { DbOrTx } from '@/lib/db/types'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 import { suspendMemberTx, unsuspendMemberTx } from '@/lib/organizations/members/lifecycle'
 import {
   invalidateAfterSessionRevocation,
@@ -49,7 +49,7 @@ export interface UpdateOutcome {
 }
 
 async function applyUserUpdate(
-  tx: DbOrTx,
+  tx: DbTransaction,
   context: ScimUseCaseContext,
   current: ScimUserRecord,
   next: ScimUserAttributes
@@ -135,7 +135,7 @@ export interface UpdateScimUserResult {
  * which take the advisory locks first and then touch rows referencing this one.
  */
 async function loadUserForUpdate(
-  tx: DbOrTx,
+  tx: DbTransaction,
   context: ScimUseCaseContext,
   scimUserId: string
 ): Promise<ScimUserRecord> {

apps/sim/ee/scim/lib/projection/auto-map.ts

@@ -1,7 +1,7 @@
 import { permissionGroup, scimGroupMapping } from '@sim/db/schema'
 import { generateId } from '@sim/utils/id'
 import { and, eq, inArray, ne } from 'drizzle-orm'
-import type { DbOrTx } from '@/lib/db/types'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 import { acquirePermissionGroupOrgLock } from '@/lib/permission-groups/locks'
 
 /**
@@ -94,7 +94,7 @@ export async function autoMapPermissionGroupByName(
  * mode, and both commit together.
  */
 export async function settleMappedPermissionGroupsExplicit(
-  tx: DbOrTx,
+  tx: DbTransaction,
   params: { organizationId: string; scimGroupId: string }
 ): Promise<void> {
   const inheriting = await tx

apps/sim/ee/scim/lib/projection/reconcile-user.ts

@@ -16,7 +16,7 @@ import { generateId } from '@sim/utils/id'
 import { and, eq, inArray } from 'drizzle-orm'
 import { acquireOrganizationUserMutationLocks } from '@/lib/billing/organizations/membership'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
-import type { DbOrTx } from '@/lib/db/types'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 import { changeMemberRoleTx } from '@/lib/organizations/members/lifecycle'
 import {
   addPermissionGroupMemberTx,
@@ -195,7 +195,7 @@ async function findForeignWorkspaces(
 
 /** Applies one grant. `skipped` means the grant describes nothing this server can apply. */
 async function applyGrant(
-  tx: DbOrTx,
+  tx: DbTransaction,
   params: {
     organizationId: string
     userId: string
@@ -259,7 +259,7 @@ async function applyGrant(
  * is simply inert for that one person.
  */
 async function setOrganizationRole(
-  tx: DbOrTx,
+  tx: DbTransaction,
   organizationId: string,
   userId: string,
   role: 'admin' | 'member'
@@ -293,7 +293,7 @@ async function setOrganizationRole(
  * has made the directory the source of truth.
  */
 async function withdrawGrant(
-  tx: DbOrTx,
+  tx: DbTransaction,
   params: {
     organizationId: string
     userId: string
@@ -398,7 +398,7 @@ async function withdrawGrant(
  * without a dry-run mode.
  */
 export async function reconcileUserProjection(
-  tx: DbOrTx,
+  tx: DbTransaction,
   params: {
     connectionId: string
     organizationId: string
@@ -576,7 +576,7 @@ export async function reconcileUserProjection(
 
 /** Reconciles several users, in a stable order so concurrent syncs cannot deadlock. */
 export async function reconcileUsersProjection(
-  tx: DbOrTx,
+  tx: DbTransaction,
   params: {
     connectionId: string
     organizationId: string

apps/sim/ee/workspace-forking/application/revision.ts

@@ -23,7 +23,7 @@ import {
   workspaceSandbox,
 } from '@sim/db/schema'
 import { and, type SQL, sql } from 'drizzle-orm'
-import type { DbOrTx } from '@/lib/db/types'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 import { acquireFolderMutationLock } from '@/lib/folders/locks'
 import { activeWorkspaceFileConditions } from '@/lib/workspace-files/query-scope'
 import {
@@ -141,7 +141,7 @@ export async function loadForkPreviewRevision(
 }
 
 /** Locks normalized graph rows as well as workflow metadata, including realtime-only writes. */
-export async function lockForkRevision(tx: DbOrTx, scope: ForkRevisionScope): Promise<void> {
+export async function lockForkRevision(tx: DbTransaction, scope: ForkRevisionScope): Promise<void> {
   const workspaceIds = [
     ...new Set([
       scope.sourceWorkspaceId,

apps/sim/ee/workspace-forking/lib/copy/workflow-mcp-attachments.ts

@@ -1,7 +1,7 @@
 import { workflowMcpServer, workflowMcpTool } from '@sim/db/schema'
 import { generateId } from '@sim/utils/id'
 import { and, eq, inArray, isNull } from 'drizzle-orm'
-import type { DbOrTx } from '@/lib/db/types'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 import { acquireWorkflowMcpServerLock } from '@/lib/mcp/server-locks'
 import { validateMcpToolMetadataForStorage } from '@/lib/mcp/tool-limits'
 import { getEdgeMappingRows } from '@/ee/workspace-forking/lib/mapping/mapping-store'
@@ -102,7 +102,7 @@ export async function copyForkWorkflowMcpAttachments(params: {
  * Returns the affected target server ids so the caller can notify them post-commit.
  */
 export async function reconcileForkWorkflowMcpAttachments(params: {
-  tx: DbOrTx
+  tx: DbTransaction
   childWorkspaceId: string
   /** True when the sync SOURCE is the parent workspace (a pull). */
   sourceIsParent: boolean