feat(db): role-keyed dbFor clients for cleanup and exec workloads

apps/sim/background/cleanup-logs.test.ts

@@ -80,12 +80,17 @@ const {
   }
 })
 
-vi.mock('@sim/db', () => ({
-  db: {
+vi.mock('@sim/db', () => {
+  const db = {
     execute: mockExecute,
     select: mockSelect,
-  },
-}))
+  }
+  return {
+    db,
+    // Cleanup-pool client shares the instance so the seeded chains still apply.
+    dbFor: () => db,
+  }
+})
 
 vi.mock('@sim/db/schema', () => ({
   executionLargeValueDependencies: {
@@ -255,7 +260,7 @@ describe('cleanup logs worker', () => {
       workspaceIds: ['workspace-1'],
     })
 
-    expect(mockMarkLargeValuesDeleted).toHaveBeenCalledWith([largeValueKey])
+    expect(mockMarkLargeValuesDeleted).toHaveBeenCalledWith([largeValueKey], expect.anything())
     expect(mockDeleteFileMetadata).toHaveBeenCalledTimes(2)
   })
 

apps/sim/background/cleanup-logs.ts

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { dbFor } from '@sim/db'
 import {
   executionLargeValueDependencies,
   executionLargeValueReferences,
@@ -30,6 +30,9 @@ import { deleteFileMetadata } from '@/lib/uploads/server/metadata'
 
 const logger = createLogger('CleanupLogs')
 
+/** All cleanup queries run on the dedicated cleanup pool. */
+const cleanupDb = dbFor('cleanup')
+
 interface FileDeleteStats {
   filesTotal: number
   filesDeleted: number
@@ -107,7 +110,7 @@ async function deleteLargeValueKeys(keys: string[]): Promise<{ deleted: number;
 
   if (deletedKeys.length > 0) {
     try {
-      await markLargeValuesDeleted(deletedKeys)
+      await markLargeValuesDeleted(deletedKeys, cleanupDb)
     } catch (error) {
       logger.error('Failed to mark large execution values as deleted:', { error })
       return { deleted: 0, failed: result.failed.length + deletedKeys.length }
@@ -153,7 +156,7 @@ async function cleanupLargeExecutionValues(
         LARGE_VALUE_CLEANUP_BATCH_SIZE,
         LARGE_VALUE_CLEANUP_TOTAL_KEY_LIMIT - attempted
       )
-      const rows = await db
+      const rows = await cleanupDb
         .select({ key: executionLargeValues.key })
         .from(executionLargeValues)
         .where(
@@ -219,7 +222,7 @@ async function cleanupLegacyLargeExecutionValues(
         LARGE_VALUE_CLEANUP_BATCH_SIZE,
         LARGE_VALUE_CLEANUP_TOTAL_KEY_LIMIT - attempted
       )
-      const rows = await db
+      const rows = await cleanupDb
         .select({ key: workspaceFiles.key })
         .from(workspaceFiles)
         .where(
@@ -353,7 +356,11 @@ async function cleanupLargeValueMetadata(workspaceIds: string[], label: string):
     const tombstonesDeletedBefore = new Date(
       Date.now() - LARGE_VALUE_TOMBSTONE_RETENTION_HOURS * 60 * 60 * 1000
     )
-    const result = await pruneLargeValueMetadata({ workspaceIds, tombstonesDeletedBefore })
+    const result = await pruneLargeValueMetadata({
+      workspaceIds,
+      tombstonesDeletedBefore,
+      dbClient: cleanupDb,
+    })
     logger.info(
       `[${label}/execution_large_value_metadata] Pruned ${result.referencesDeleted} stale references, ${result.dependenciesDeleted} dependencies, ${result.tombstonesDeleted} tombstones`
     )
@@ -377,8 +384,9 @@ async function cleanupWorkflowExecutionLogs(
     tableDef: workflowExecutionLogs,
     workspaceIds,
     tableName: `${label}/workflow_execution_logs`,
+    dbClient: cleanupDb,
     selectChunk: (chunkIds, limit) =>
-      db
+      cleanupDb
         .select({
           id: workflowExecutionLogs.id,
           files: workflowExecutionLogs.files,
@@ -465,6 +473,7 @@ export async function runCleanupLogs(payload: CleanupJobPayload): Promise<void>
     workspaceIds,
     retentionDate,
     tableName: `${label}/job_execution_logs`,
+    dbClient: cleanupDb,
   })
 
   if (runGlobalHousekeeping && plan === 'free') {

apps/sim/background/cleanup-soft-deletes.test.ts

@@ -43,7 +43,11 @@ const {
   }
 })
 
-vi.mock('@sim/db', () => ({ db: { select: mockSelect } }))
+vi.mock('@sim/db', () => {
+  const db = { select: mockSelect }
+  // Cleanup-pool client shares the instance so the seeded chains still apply.
+  return { db, dbFor: () => db }
+})
 
 vi.mock('@sim/db/schema', () => {
   const table = (cols: string[]) =>

apps/sim/background/cleanup-soft-deletes.ts

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { dbFor } from '@sim/db'
 import {
   copilotChats,
   document,
@@ -29,6 +29,9 @@ import { deleteFileMetadata } from '@/lib/uploads/server/metadata'
 
 const logger = createLogger('CleanupSoftDeletes')
 
+/** All cleanup queries run on the dedicated cleanup pool. */
+const cleanupDb = dbFor('cleanup')
+
 const KB_ORPHAN_BINDING_BATCH_SIZE = 500
 const KB_ORPHAN_BINDING_TOTAL_LIMIT = 5_000
 /**
@@ -57,7 +60,7 @@ async function selectExpiredWorkspaceFiles(
 ): Promise<WorkspaceFileScope> {
   const [legacyRows, multiContextRows] = await Promise.all([
     selectRowsByIdChunks(workspaceIds, (chunkIds, chunkLimit) =>
-      db
+      cleanupDb
         .select({ id: workspaceFile.id, key: workspaceFile.key })
         .from(workspaceFile)
         .where(
@@ -70,7 +73,7 @@ async function selectExpiredWorkspaceFiles(
         .limit(chunkLimit)
     ),
     selectRowsByIdChunks(workspaceIds, (chunkIds, chunkLimit) =>
-      db
+      cleanupDb
         .select({
           id: workspaceFiles.id,
           key: workspaceFiles.key,
@@ -194,7 +197,7 @@ async function cleanupOrphanedKnowledgeBaseBindings(
         KB_ORPHAN_BINDING_BATCH_SIZE,
         KB_ORPHAN_BINDING_TOTAL_LIMIT - attempted
       )
-      const rows = await db
+      const rows = await cleanupDb
         .select({ key: workspaceFiles.key })
         .from(workspaceFiles)
         .where(
@@ -268,7 +271,7 @@ export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise
   // different subsets above the LIMIT cap and orphan or prematurely purge data.
   const [doomedWorkflows, fileScope] = await Promise.all([
     selectRowsByIdChunks(workspaceIds, (chunkIds, chunkLimit) =>
-      db
+      cleanupDb
         .select({ id: workflow.id })
         .from(workflow)
         .where(
@@ -288,7 +291,7 @@ export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise
 
   if (doomedWorkflowIds.length > 0) {
     const doomedChats = await selectRowsByIdChunks(doomedWorkflowIds, (chunkIds, chunkLimit) =>
-      db
+      cleanupDb
         .select({ id: copilotChats.id })
         .from(copilotChats)
         .where(inArray(copilotChats.workflowId, chunkIds))
@@ -310,23 +313,26 @@ export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise
     workflow,
     workflow.id,
     doomedWorkflowIds,
-    `${label}/workflow`
+    `${label}/workflow`,
+    cleanupDb
   )
   totalDeleted += workflowResult.deleted
 
   const legacyFileResult = await deleteRowsById(
     workspaceFile,
     workspaceFile.id,
     fileScope.legacyRows.map((r) => r.id),
-    `${label}/workspaceFile`
+    `${label}/workspaceFile`,
+    cleanupDb
   )
   totalDeleted += legacyFileResult.deleted
 
   const multiContextFileResult = await deleteRowsById(
     workspaceFiles,
     workspaceFiles.id,
     fileScope.multiContextRows.map((r) => r.id),
-    `${label}/workspaceFiles`
+    `${label}/workspaceFiles`,
+    cleanupDb
   )
   totalDeleted += multiContextFileResult.deleted
 
@@ -339,6 +345,7 @@ export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise
       retentionDate,
       tableName: `${label}/${target.name}`,
       requireTimestampNotNull: true,
+      dbClient: cleanupDb,
     })
     totalDeleted += result.deleted
   }

apps/sim/background/cleanup-tasks.ts

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { dbFor } from '@sim/db'
 import {
   copilotAsyncToolCalls,
   copilotChats,
@@ -21,6 +21,9 @@ import { prepareChatCleanup } from '@/lib/cleanup/chat-cleanup'
 
 const logger = createLogger('CleanupTasks')
 
+/** All cleanup queries run on the dedicated cleanup pool. */
+const cleanupDb = dbFor('cleanup')
+
 /**
  * Delete copilot run checkpoints and async tool calls via join through copilotRuns.
  * These tables don't have a direct workspaceId — we find qualifying run IDs first.
@@ -46,7 +49,7 @@ async function cleanupRunChildren(
   if (workspaceIds.length === 0) return []
 
   const runIds = await selectRowsByIdChunks(workspaceIds, (chunkIds, chunkLimit) =>
-    db
+    cleanupDb
       .select({ id: copilotRuns.id })
       .from(copilotRuns)
       .where(
@@ -62,7 +65,9 @@ async function cleanupRunChildren(
   const ids = runIds.map((r) => r.id)
 
   return Promise.all(
-    RUN_CHILD_TABLES.map((t) => deleteRowsById(t.table, t.runIdCol, ids, `${label}/${t.name}`))
+    RUN_CHILD_TABLES.map((t) =>
+      deleteRowsById(t.table, t.runIdCol, ids, `${label}/${t.name}`, cleanupDb)
+    )
   )
 }
 
@@ -81,7 +86,7 @@ export async function runCleanupTasks(payload: CleanupJobPayload): Promise<void>
   )
 
   const doomedChats = await selectRowsByIdChunks(workspaceIds, (chunkIds, chunkLimit) =>
-    db
+    cleanupDb
       .select({ id: copilotChats.id })
       .from(copilotChats)
       .where(
@@ -106,7 +111,8 @@ export async function runCleanupTasks(payload: CleanupJobPayload): Promise<void>
     copilotFeedback,
     copilotFeedback.chatId,
     doomedChatIds,
-    `${label}/copilotFeedback`
+    `${label}/copilotFeedback`,
+    cleanupDb
   )
 
   // Delete copilot runs (has workspaceId directly, cascades checkpoints)
@@ -117,6 +123,7 @@ export async function runCleanupTasks(payload: CleanupJobPayload): Promise<void>
     workspaceIds,
     retentionDate,
     tableName: `${label}/copilotRuns`,
+    dbClient: cleanupDb,
   })
 
   // Delete copilot chats using the exact IDs collected above so the chat
@@ -125,7 +132,8 @@ export async function runCleanupTasks(payload: CleanupJobPayload): Promise<void>
     copilotChats,
     copilotChats.id,
     doomedChatIds,
-    `${label}/copilotChats`
+    `${label}/copilotChats`,
+    cleanupDb
   )
 
   // Delete mothership inbox tasks (has workspaceId directly)
@@ -136,6 +144,7 @@ export async function runCleanupTasks(payload: CleanupJobPayload): Promise<void>
     workspaceIds,
     retentionDate,
     tableName: `${label}/mothershipInboxTask`,
+    dbClient: cleanupDb,
   })
 
   const totalDeleted =

apps/sim/lib/cleanup/batch-delete.ts

@@ -5,6 +5,13 @@ import type { PgColumn, PgTable } from 'drizzle-orm/pg-core'
 
 const logger = createLogger('BatchDelete')
 
+/**
+ * Structural client surface the delete helpers need. Satisfied by the global
+ * `db`, a `dbFor(...)` sub-pool client, and a transaction handle, so callers
+ * pick which pool the deletes run on (cleanup jobs pass `dbFor('cleanup')`).
+ */
+export type BatchDeleteClient = Pick<typeof db, 'select' | 'delete'>
+
 export const DEFAULT_BATCH_SIZE = 2000
 /** 50 × 2000 = 100K row cap per cleanup run; drains long-tail tenants in days, not weeks. */
 export const DEFAULT_MAX_BATCHES_PER_TABLE = 50
@@ -84,6 +91,8 @@ export interface ChunkedBatchDeleteOptions<TRow extends { id: string }> {
    */
   totalRowLimit?: number
   workspaceChunkSize?: number
+  /** Client the DELETEs run on. Defaults to the global pool. */
+  dbClient?: BatchDeleteClient
 }
 
 /**
@@ -107,6 +116,7 @@ export async function chunkedBatchDelete<TRow extends { id: string }>({
   maxBatches = DEFAULT_MAX_BATCHES_PER_TABLE,
   totalRowLimit = DEFAULT_BATCH_SIZE * DEFAULT_MAX_BATCHES_PER_TABLE,
   workspaceChunkSize = DEFAULT_WORKSPACE_CHUNK_SIZE,
+  dbClient = db,
 }: ChunkedBatchDeleteOptions<TRow>): Promise<TableCleanupResult> {
   const result: TableCleanupResult = { table: tableName, deleted: 0, failed: 0 }
 
@@ -149,7 +159,7 @@ export async function chunkedBatchDelete<TRow extends { id: string }>({
         if (onBatch) await onBatch(rows)
 
         const ids = rows.map((r) => r.id)
-        const deleted = await db
+        const deleted = await dbClient
           .delete(tableDef)
           .where(inArray(sql`id`, ids))
           .returning({ id: sql`id` })
@@ -189,6 +199,8 @@ export interface BatchDeleteOptions {
   batchSize?: number
   maxBatches?: number
   workspaceChunkSize?: number
+  /** Client the SELECTs and DELETEs run on. Defaults to the global pool. */
+  dbClient?: BatchDeleteClient
 }
 
 /**
@@ -204,16 +216,18 @@ export async function batchDeleteByWorkspaceAndTimestamp({
   retentionDate,
   tableName,
   requireTimestampNotNull = false,
+  dbClient = db,
   ...rest
 }: BatchDeleteOptions): Promise<TableCleanupResult> {
   return chunkedBatchDelete({
     tableDef,
     workspaceIds,
     tableName,
+    dbClient,
     selectChunk: (chunkIds, limit) => {
       const predicates = [inArray(workspaceIdCol, chunkIds), lt(timestampCol, retentionDate)]
       if (requireTimestampNotNull) predicates.push(isNotNull(timestampCol))
-      return db
+      return dbClient
         .select({ id: sql<string>`id` })
         .from(tableDef)
         .where(and(...predicates))
@@ -232,6 +246,7 @@ export async function deleteRowsById(
   idCol: PgColumn,
   ids: string[],
   tableName: string,
+  dbClient: BatchDeleteClient = db,
   chunkSize: number = DEFAULT_DELETE_CHUNK_SIZE
 ): Promise<TableCleanupResult> {
   const result: TableCleanupResult = { table: tableName, deleted: 0, failed: 0 }
@@ -240,7 +255,7 @@ export async function deleteRowsById(
   const chunks = chunkArray(ids, chunkSize)
   for (const [chunkIdx, chunkIds] of chunks.entries()) {
     try {
-      const deleted = await db
+      const deleted = await dbClient
         .delete(tableDef)
         .where(inArray(idCol, chunkIds))
         .returning({ id: idCol })