perf(db): cut hot-path database work and close copilot cleanup gaps

apps/sim/app/api/v1/audit-logs/route.ts

@@ -25,12 +25,7 @@ import { generateId } from '@sim/utils/id'
 import { type NextRequest, NextResponse } from 'next/server'
 import { v1ListAuditLogsContract } from '@/lib/api/contracts/v1/audit-logs'
 import { parseRequest } from '@/lib/api/server'
-import {
-  buildFilterConditions,
-  buildOrgScopeCondition,
-  getOrgWorkspaceIds,
-  queryAuditLogs,
-} from '@/lib/audit-logs/query'
+import { buildFilterConditions, getOrgWorkspaceIds, queryAuditLogs } from '@/lib/audit-logs/query'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { validateV1EnterpriseAuditAccess } from '@/app/api/v1/audit-logs/auth'
 import { formatAuditLogEntry } from '@/app/api/v1/audit-logs/format'
@@ -99,12 +94,12 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       )
     }
 
-    const scopeCondition = buildOrgScopeCondition({
+    const scope = {
       organizationId,
       orgWorkspaceIds,
       orgMemberIds,
       includeDeparted: params.includeDeparted,
-    })
+    }
     const filterConditions = buildFilterConditions({
       action: params.action,
       resourceType: params.resourceType,
@@ -116,7 +111,8 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     })
 
     const { data, nextCursor } = await queryAuditLogs(
-      [scopeCondition, ...filterConditions],
+      scope,
+      filterConditions,
       params.limit,
       params.cursor
     )

apps/sim/app/api/webhooks/cleanup/idempotency/route.ts

@@ -2,7 +2,7 @@ import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { cleanupExpiredIdempotencyKeys, getIdempotencyKeyStats } from '@/lib/core/idempotency'
+import { cleanupExpiredIdempotencyKeys, estimateIdempotencyKeyCount } from '@/lib/core/idempotency'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 
@@ -21,34 +21,27 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       return authError
     }
 
-    const statsBefore = await getIdempotencyKeyStats()
-    logger.info(
-      `Pre-cleanup stats: ${statsBefore.totalKeys} keys across ${Object.keys(statsBefore.keysByNamespace).length} namespaces`
-    )
-
     const result = await cleanupExpiredIdempotencyKeys({
       maxAgeSeconds: 7 * 24 * 60 * 60, // 7 days
       batchSize: 1000,
     })
 
-    const statsAfter = await getIdempotencyKeyStats()
-    logger.info(`Post-cleanup stats: ${statsAfter.totalKeys} keys remaining`)
+    const estimatedKeysRemaining = await estimateIdempotencyKeyCount()
+    logger.info('Idempotency cleanup finished', {
+      deleted: result.deleted,
+      deletedByNamespace: result.deletedByNamespace,
+      estimatedKeysRemaining,
+    })
 
     return NextResponse.json({
       success: true,
       message: 'Idempotency key cleanup completed',
       requestId,
       result: {
         deleted: result.deleted,
+        deletedByNamespace: result.deletedByNamespace,
         errors: result.errors,
-        statsBefore: {
-          totalKeys: statsBefore.totalKeys,
-          keysByNamespace: statsBefore.keysByNamespace,
-        },
-        statsAfter: {
-          totalKeys: statsAfter.totalKeys,
-          keysByNamespace: statsAfter.keysByNamespace,
-        },
+        estimatedKeysRemaining,
       },
     })
   } catch (error) {

apps/sim/app/workspace/[workspaceId]/files/hooks/use-workspace-files-room.ts

@@ -19,10 +19,11 @@ export function useWorkspaceFilesRoom(workspaceId: string): void {
   useWorkspaceInvalidationRoom(
     workspaceId,
     ROOM_TYPES.WORKSPACE_FILES,
-    () => {
-      invalidateWorkspaceFileBrowsers(queryClient, workspaceId)
-      void queryClient.invalidateQueries({ queryKey: dashboardKeys.workspace(workspaceId) })
-    },
+    () =>
+      Promise.all([
+        invalidateWorkspaceFileBrowsers(queryClient, workspaceId),
+        queryClient.invalidateQueries({ queryKey: dashboardKeys.workspace(workspaceId) }),
+      ]),
     WORKSPACE_FILE_BROWSER_INVALIDATION_KEY
   )
 }

apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.test.tsx

@@ -13,7 +13,10 @@ vi.mock('@/app/workspace/providers/socket-provider', () => ({
   useSocket: () => mockUseSocket(),
 }))
 
-import { useWorkspaceInvalidationRoom } from '@/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room'
+import {
+  INVALIDATION_COALESCE_MS,
+  useWorkspaceInvalidationRoom,
+} from '@/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room'
 
 function fakeSocket() {
   return { connected: true, emit: vi.fn(), on: vi.fn(), off: vi.fn() }
@@ -25,7 +28,7 @@ const roots: Root[] = []
 function mount(
   workspaceId: string,
   roomType: RoomType,
-  onChanged: () => void = () => {},
+  onChanged: () => unknown = () => {},
   dedupeKey?: string
 ): Root {
   function Probe() {
@@ -44,6 +47,11 @@ function unmount(root: Root) {
   act(() => root.unmount())
 }
 
+/** Delivers a `${roomType}-changed` broadcast to the handler the room registered. */
+function broadcast(event: string, workspaceId: string) {
+  for (const [name, handler] of socket.on.mock.calls) if (name === event) handler({ workspaceId })
+}
+
 const emitted = (event: string, workspaceId?: string) =>
   socket.emit.mock.calls.filter(
     ([name, payload]) =>
@@ -99,4 +107,50 @@ describe('useWorkspaceInvalidationRoom', () => {
 
     expect(emitted('leave-workspace-files', 'ws-1')).toBe(1)
   })
+
+  describe('change delivery', () => {
+    beforeEach(() => {
+      vi.useFakeTimers()
+    })
+
+    afterEach(() => {
+      vi.useRealTimers()
+    })
+
+    it('delivers a burst of broadcasts once at once and once after it, not once per broadcast', async () => {
+      const onChanged = vi.fn()
+      mount('ws-1', 'workspace-files', onChanged)
+
+      for (let index = 0; index < 20; index++) broadcast('workspace-files-changed', 'ws-1')
+      await vi.advanceTimersByTimeAsync(0)
+      expect(onChanged).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(INVALIDATION_COALESCE_MS)
+      expect(onChanged).toHaveBeenCalledTimes(2)
+
+      await vi.advanceTimersByTimeAsync(10 * INVALIDATION_COALESCE_MS)
+      expect(onChanged).toHaveBeenCalledTimes(2)
+    })
+
+    it('waits for a refetch in flight instead of starting another beside it', async () => {
+      let settle: () => void = () => {}
+      const onChanged = vi.fn(
+        () =>
+          new Promise<void>((resolve) => {
+            settle = resolve
+          })
+      )
+      mount('ws-1', 'workspace-files', onChanged)
+
+      broadcast('workspace-files-changed', 'ws-1')
+      await vi.advanceTimersByTimeAsync(5 * INVALIDATION_COALESCE_MS)
+      broadcast('workspace-files-changed', 'ws-1')
+      await vi.advanceTimersByTimeAsync(5 * INVALIDATION_COALESCE_MS)
+      expect(onChanged).toHaveBeenCalledTimes(1)
+
+      settle()
+      await vi.advanceTimersByTimeAsync(INVALIDATION_COALESCE_MS)
+      expect(onChanged).toHaveBeenCalledTimes(2)
+    })
+  })
 })

apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.ts

@@ -3,6 +3,7 @@
 import { useEffect, useRef } from 'react'
 import { createLogger } from '@sim/logger'
 import type { RoomType } from '@sim/realtime-protocol/rooms'
+import { sleep } from '@sim/utils/helpers'
 import type { Socket } from 'socket.io-client'
 import { useSocket } from '@/app/workspace/providers/socket-provider'
 
@@ -11,6 +12,11 @@ const logger = createLogger('WorkspaceInvalidationRoom')
 /** Retry cap + base delay for a retryable join failure on an otherwise-live socket. */
 const MAX_JOIN_RETRIES = 3
 const JOIN_RETRY_BASE_MS = 1000
+/**
+ * The least time between two deliveries of a room's changes. A burst of broadcasts (an API client
+ * uploading files one by one) otherwise refetches every viewer's list once per broadcast.
+ */
+export const INVALIDATION_COALESCE_MS = 1000
 
 interface JoinErrorPayload {
   workspaceId: string
@@ -19,8 +25,14 @@ interface JoinErrorPayload {
   retryable?: boolean
 }
 
+/**
+ * Runs when the room's resources changed. A returned promise is awaited before the next delivery,
+ * so a refetch still in flight is neither cancelled nor overlapped by the next one.
+ */
+type RoomChangedCallback = () => unknown
+
 interface SharedRoomSubscription {
-  callbacks: Map<string | symbol, Set<() => void>>
+  callbacks: Map<string | symbol, Set<RoomChangedCallback>>
   dispose: () => void
 }
 
@@ -36,9 +48,12 @@ function createSharedRoomSubscription(
   const errorEvent = `${joinEvent}-error`
   const leaveEvent = `leave-${roomType}`
   const changedEvent = `${roomType}-changed`
-  const callbacks = new Map<string | symbol, Set<() => void>>()
+  const callbacks = new Map<string | symbol, Set<RoomChangedCallback>>()
   let retries = 0
   let retryTimer: ReturnType<typeof setTimeout> | null = null
+  let delivering = false
+  let changedSinceDelivery = false
+  let disposed = false
 
   const join = () => socket.emit(joinEvent, { workspaceId })
   const handleConnect = () => {
@@ -61,9 +76,31 @@ function createSharedRoomSubscription(
     if (retryTimer) clearTimeout(retryTimer)
     retryTimer = setTimeout(join, JOIN_RETRY_BASE_MS * retries)
   }
+  /**
+   * The first change is delivered at once. Changes arriving while a delivery runs, or within
+   * {@link INVALIDATION_COALESCE_MS} after it settles, are folded into one delivery after that.
+   */
+  const deliver = async () => {
+    delivering = true
+    try {
+      do {
+        changedSinceDelivery = false
+        await Promise.allSettled(
+          [...callbacks.values()].map(async (group) => group.values().next().value?.())
+        )
+        await sleep(INVALIDATION_COALESCE_MS)
+      } while (changedSinceDelivery && !disposed)
+    } finally {
+      delivering = false
+    }
+  }
   const handleChanged = (data: { workspaceId: string }) => {
     if (data.workspaceId !== workspaceId) return
-    for (const group of callbacks.values()) group.values().next().value?.()
+    if (delivering) {
+      changedSinceDelivery = true
+      return
+    }
+    void deliver()
   }
 
   if (socket.connected) join()
@@ -75,6 +112,7 @@ function createSharedRoomSubscription(
   return {
     callbacks,
     dispose: () => {
+      disposed = true
       if (retryTimer) clearTimeout(retryTimer)
       socket.off('connect', handleConnect)
       socket.off(successEvent, handleJoinSuccess)
@@ -88,16 +126,18 @@ function createSharedRoomSubscription(
 /**
  * Joins a workspace-scoped, presence-free "invalidation room" over the shared socket and runs
  * `onChanged` whenever the server broadcasts `${roomType}-changed` for this workspace, so the list
- * refetches without waiting for staleness. Shared core behind {@link useWorkspaceFilesRoom} and
- * {@link useWorkspaceTablesRoom}; event names derive from `roomType`.
+ * refetches without waiting for staleness. A burst of broadcasts is coalesced: one delivery at a
+ * time, at most one per {@link INVALIDATION_COALESCE_MS} after the previous one settles. Shared core
+ * behind {@link useWorkspaceFilesRoom} and {@link useWorkspaceTablesRoom}; event names derive from
+ * `roomType`.
  *
  * These rooms carry no presence — "who's in a resource" comes from the per-resource room, not from
  * who's browsing the section. Mutations happen server-side (HTTP + copilot) and fan out this signal.
  */
 export function useWorkspaceInvalidationRoom(
   workspaceId: string,
   roomType: RoomType,
-  onChanged: () => void,
+  onChanged: RoomChangedCallback,
   dedupeKey?: string
 ): void {
   const { socket } = useSocket()

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/hooks/use-resource-folders.ts

@@ -1,6 +1,6 @@
 'use client'
 
-import { useCallback, useMemo } from 'react'
+import { useMemo } from 'react'
 import { ROOM_TYPES } from '@sim/realtime-protocol/rooms'
 import { getErrorMessage } from '@sim/utils/errors'
 import { useQueryClient } from '@tanstack/react-query'
@@ -74,13 +74,10 @@ export function useResourceFolders(
    * broadcast (knowledge bases and tables still wait out their stale time).
    */
   const queryClient = useQueryClient()
-  const onFilesChanged = useCallback(() => {
-    if (workspaceId) invalidateWorkspaceFileBrowsers(queryClient, workspaceId)
-  }, [queryClient, workspaceId])
   useWorkspaceInvalidationRoom(
     isFileResource ? (workspaceId ?? '') : '',
     ROOM_TYPES.WORKSPACE_FILES,
-    onFilesChanged,
+    () => (workspaceId ? invalidateWorkspaceFileBrowsers(queryClient, workspaceId) : undefined),
     WORKSPACE_FILE_BROWSER_INVALIDATION_KEY
   )
 

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

@@ -6,6 +6,7 @@ import {
   schemaMock,
 } from '@sim/testing'
 import { billingStorageMock } from '@sim/testing/mocks/billing-storage.mock'
+import { chatCleanupMock, chatCleanupMockFns } from '@sim/testing/mocks/chat-cleanup.mock'
 import {
   knowledgeDocumentsServiceMock,
   knowledgeDocumentsServiceMockFns,
@@ -25,7 +26,6 @@ const {
   mockScopedChunkedBatchDelete,
   mockDeleteRowsById,
   mockKnowledgeBaseContainerDelete,
-  mockPrepareChatCleanup,
   mockSelectRowsByIdChunks,
   mockSettleDetachedConnectorReservations,
   mockDeduplicateFolderName,
@@ -36,7 +36,6 @@ const {
   mockScopedChunkedBatchDelete: vi.fn(async () => ({ deleted: 0, failed: 0 })),
   mockDeleteRowsById: vi.fn(async () => ({ deleted: 0, failed: 0 })),
   mockKnowledgeBaseContainerDelete: vi.fn(),
-  mockPrepareChatCleanup: vi.fn(async () => ({ execute: vi.fn(async () => undefined) })),
   mockSelectRowsByIdChunks: vi.fn(async () => [] as unknown[]),
   mockSettleDetachedConnectorReservations: vi.fn(async () => undefined),
 }))
@@ -58,7 +57,7 @@ vi.mock('@/lib/cleanup/queue', () => ({
   retentionCleanupQueue: { name: 'retention-cleanup', concurrencyLimit: 1 },
 }))
 
-vi.mock('@/lib/cleanup/chat-cleanup', () => ({ prepareChatCleanup: mockPrepareChatCleanup }))
+vi.mock('@/lib/cleanup/chat-cleanup', () => chatCleanupMock)
 
 vi.mock('@/lib/billing/storage', () => billingStorageMock)
 
@@ -82,6 +81,8 @@ vi.mock('@/lib/folders/naming', () => ({ deduplicateFolderName: mockDeduplicateF
 
 import { runCleanupSoftDeletes } from '@/background/cleanup-soft-deletes'
 
+const { mockPrepareChatCleanup } = chatCleanupMockFns
+
 const { mockHardDeleteDocuments } = knowledgeDocumentsServiceMockFns
 mockHardDeleteDocuments.mockImplementation(async (ids: string[]) => ids.length)