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

@@ -66,13 +66,18 @@ const {
  }
 })
 
-vi.mock('@sim/db', () => ({
-db: {
+vi.mock('@sim/db', () => {
+const db = {
  delete: mockDelete,
  select: mockSelect,
  transaction: mockTransaction,
- },
-}))
+ }
+ return {
+ db,
+ // Cleanup-pool client shares the instance so the seeded chains still apply.
+ 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 { db, dbFor } from '@sim/db'
 import {
  copilotChats,
  document,
@@ -36,6 +36,13 @@ import { deleteFileMetadata } from '@/lib/uploads/server/metadata'
 
 const logger = createLogger('CleanupSoftDeletes')
 
+/**
+ * Cleanup queries run on the dedicated cleanup pool. The one exception is the
+ * billable-file transaction below, which couples row deletion with a storage
+ * billing decrement — billing writes stay on the default client.
+ */
+const cleanupDb = dbFor('cleanup')
+
 const KB_ORPHAN_BINDING_BATCH_SIZE = 500
 const KB_ORPHAN_BINDING_TOTAL_LIMIT = 5_000
 /**
@@ -80,7 +87,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,
@@ -97,7 +104,7 @@ async function selectExpiredWorkspaceFiles(
  .limit(chunkLimit)
  ),
  selectRowsByIdChunks(workspaceIds, (chunkIds, chunkLimit) =>
-db
+cleanupDb
  .select({
  id: workspaceFiles.id,
  key: workspaceFiles.key,
@@ -199,7 +206,7 @@ async function deleteExpiredLegacyWorkspaceFileRows(
  const result = { deleted: 0, failed: 0 }
  for (const batch of chunkArray(rows, DEFAULT_DELETE_CHUNK_SIZE)) {
  try {
- const deleted = await db
+ const deleted = await cleanupDb
  .delete(workspaceFile)
  .where(
  and(
@@ -239,7 +246,7 @@ async function deleteExpiredUnbilledWorkspaceFileRows(
  for (const [context, contextRows] of rowsByContext) {
  for (const batch of chunkArray(contextRows, DEFAULT_DELETE_CHUNK_SIZE)) {
  try {
- const deleted = await db
+ const deleted = await cleanupDb
  .delete(workspaceFiles)
  .where(
  and(
@@ -342,7 +349,7 @@ async function hardDeleteKnowledgeBaseDocuments(
  label: string
 ): Promise<void> {
  for (let batch = 0; batch < KB_DOCUMENT_DELETE_MAX_BATCHES; batch++) {
- const documentRows = await db
+ const documentRows = await cleanupDb
  .select({ id: document.id })
  .from(document)
  .where(inArray(document.knowledgeBaseId, knowledgeBaseIds))
@@ -357,7 +364,7 @@ async function hardDeleteKnowledgeBaseDocuments(
  }
  }
 
- const remaining = await db
+ const remaining = await cleanupDb
  .select({ id: document.id })
  .from(document)
  .where(inArray(document.knowledgeBaseId, knowledgeBaseIds))
@@ -377,8 +384,9 @@ async function cleanupExpiredKnowledgeBases(
  workspaceIds,
  tableName: `${label}/knowledgeBase`,
  batchSize: KB_RETENTION_BATCH_SIZE,
+ dbClient: cleanupDb,
  selectChunk: (chunkIds, limit) =>
-db
+cleanupDb
  .select({ id: knowledgeBase.id })
  .from(knowledgeBase)
  .where(
@@ -457,7 +465,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(
@@ -535,7 +543,7 @@ export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise
  // prematurely purge data.
  const [doomedWorkflows, fileScope, expiredSoftDeletedChats] = await Promise.all([
  selectRowsByIdChunks(workspaceIds, (chunkIds, chunkLimit) =>
-db
+cleanupDb
  .select({ id: workflow.id })
  .from(workflow)
  .where(
@@ -549,7 +557,7 @@ export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise
  ),
  selectExpiredWorkspaceFiles(workspaceIds, retentionDate),
  selectRowsByIdChunks(workspaceIds, (chunkIds, chunkLimit) =>
-db
+cleanupDb
  .select({ id: copilotChats.id })
  .from(copilotChats)
  .where(
@@ -570,7 +578,7 @@ export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise
  const doomedChatIds = new Set(softDeletedChatIds)
  if (doomedWorkflowIds.length > 0) {
  const workflowChats = await selectRowsByIdChunks(doomedWorkflowIds, (chunkIds, chunkLimit) =>
-db
+cleanupDb
  .select({ id: copilotChats.id })
  .from(copilotChats)
  .where(inArray(copilotChats.workflowId, chunkIds))
@@ -595,7 +603,7 @@ export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise
  // spares their backend data and files.
  for (const batch of chunkArray(doomedWorkflowIds, DEFAULT_DELETE_CHUNK_SIZE)) {
  try {
- const deleted = await db
+ const deleted = await cleanupDb
  .delete(workflow)
  .where(
  and(
@@ -618,7 +626,7 @@ export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise
  // also re-checks row existence before purging external data).
  for (const batch of chunkArray(softDeletedChatIds, DEFAULT_DELETE_CHUNK_SIZE)) {
  try {
- const deleted = await db
+ const deleted = await cleanupDb
  .delete(copilotChats)
  .where(
  and(
@@ -667,6 +675,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,
@@ -22,6 +22,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.
@@ -47,7 +50,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(
@@ -63,7 +66,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)
+ )
  )
 }
 
@@ -82,7 +87,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(
@@ -110,6 +115,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
@@ -122,7 +128,7 @@ export async function runCleanupTasks(payload: CleanupJobPayload): Promise<void>
  const chatsResult = { deleted: 0, failed: 0 }
  for (const batch of chunkArray(doomedChatIds, DEFAULT_DELETE_CHUNK_SIZE)) {
  try {
- const deleted = await db
+ const deleted = await cleanupDb
  .delete(copilotChats)
  .where(and(inArray(copilotChats.id, batch), lt(copilotChats.updatedAt, retentionDate)))
  .returning({ id: copilotChats.id })
@@ -141,6 +147,7 @@ export async function runCleanupTasks(payload: CleanupJobPayload): Promise<void>
  workspaceIds,
  retentionDate,
  tableName: `${label}/mothershipInboxTask`,
+ dbClient: cleanupDb,
  })
 
  const totalDeleted =