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

* fix(billing): revoke a removed member's sessions after the workspace handoff locks

Member removal bumped the organization's security policy version, which
row-locks the organization, and only then locked every organization
workspace row for the shared-resource handoff. Storage writers lock a
workspace row first and then its organization payer row, so a removal
racing an upload or delete in one of the organization's workspaces
deadlocked. Session revocation now runs after the handoff holds the
workspace rows, keeping workspace rows before the organization row.

* perf(billing): admit wand and chat-title model calls through the usage gate cache

Both are single model calls gated on the payer's period usage, and both
summed the ledger on every request. They now use the execution gate
policy: a refusal is always re-read, and an admission is reused for at
most the near-limit TTL when headroom is thin. A new chat turn's dispatch
and the copilot validate callback keep reading the gate fresh.

* perf(workflows): withhold the deployed state blob when this process already caches it

loadDeployedWorkflowState selected the full state json on every call and only
then consulted the in-process cache keyed by the immutable deployment version
id, so a cache hit still shipped the whole snapshot from Postgres. The read now
remembers which version each workflow last cached and asks Postgres to return
NULL in place of state for that id, so a hit and a miss both stay one round
trip and only a miss carries the blob. A hint left behind by an evicted entry
falls through to the existing by-id read. The unused created_at projection is
dropped too.

An integration test pins the cold, warm, cutover, evicted-hint and undeployed
paths against real Postgres.

* perf(audit-logs): read each scope branch through its keyset index

The organization audit listing filtered on one predicate, workspace_id IN
(org workspaces) OR (workspace_id IS NULL AND names the org), ordered by the
millisecond-truncated timestamp. No index can return that OR in order, so
Postgres collected and sorted every row in scope to return one page; for a busy
organization the plan is a parallel sequential scan of audit_log, which on the
replica runs long enough to be cancelled by recovery conflicts.

The scope is a disjoint union, so the page's ids now come from one backward
scan of audit_log_workspace_created_at_id_idx per workspace (a lateral join
over the workspace ids) plus the organization-level rows, each stopping after
limit + 1 matches; the first limit + 1 of their union are exactly the first
limit + 1 of the whole scope, and the rows are then read by primary key in the
same order. The keyset cursor becomes the equivalent row comparison, which the
index scan seeks to directly instead of walking every newer row. queryAuditLogs
takes the scope structurally; buildOrgScopeCondition is unchanged for the
single-record read and the data drain.

An integration test pages through mixed workspace, organization-level, foreign
and departed-actor rows with sub-millisecond ties at several page sizes, with
and without the byte budget, and asserts the exact sequence the single-predicate
query returns.

* perf(tables): fetch only the mutated workspace's table triggers

fireTableTrigger runs after every committed table row mutation and loaded every
active table-trigger webhook in the deployment, joined with its full workflow
row, only to discard all but the mutated workspace's in JavaScript. It is the
most frequent webhook query in production and its largest egress source.

fetchActiveWebhooks takes an optional workspace scope that becomes a
workflow.workspace_id predicate, and the table trigger passes its workspace, so
a mutation in a workspace without table triggers reads nothing. The in-memory
workspace check stays as a second guard. An integration test pins the scoped
result to the unscoped result restricted to the workspace.

* perf(audit): hold a run's rollups for a minute instead of two seconds

A workflow run's repeated table row changes and credential accesses merge into
one audit row only while they fall inside the rollup window, two seconds after
the first. Repeats spaced further apart, which is most scheduled runs that loop
over slow tool calls, each opened a new row, so a run wrote several rows per
table and credential where the audit log docs describe one.

Inside a run's rollup scope, whose end writes every pending row anyway, the
window is now a minute: a run's repeats share a row however far apart they
fall, and a run whose process dies loses at most a minute of merged entries,
within the best-effort durability the docs state. Rollups recorded outside a
run keep the two-second window, and the per-row entry cap is unchanged, so no
identifier or count is dropped.

* perf(file-search): lock only the files a backfill page actually seeds

The backfill walk locked every file of its page FOR SHARE, and its idempotent insert
fired the revision owner trigger, which locks the file again, for every file that
already had a revision. Each pass wrote a row lock into every live file's heap tuple.

Walk without locks, check for an existing revision per row, and lock and reread only
the files still missing one before inserting. The lock still keeps a content change
from committing between the read and the insert, and a file deleted since the walk is
now skipped instead of failing the page. Every owner the page passes is still woken.

* perf(knowledge): count a completed listing window by window

A completed listing counted the connector's owned, listed, absent and tombstoned
documents in one statement over every owned document. On the largest connectors that
single read outlasts the statement timeout, so the reconciliation pass failed before
it could walk anything.

Sum the counts over the same bounded id windows the reconciliation walks scan, one
statement per window, with the lease heartbeat between windows and the run deadline
ending an unfinished count the way it ends an unfinished walk.

* perf(files): coalesce workspace list invalidation broadcasts per tab

Every file mutation broadcasts a list-changed event to each open browser in the
workspace, and each one invalidated the whole workspace file list at once. An API
client uploading files one by one made every viewer refetch the full list once per
upload, and invalidating a query mid-fetch cancels and restarts it, so a large
workspace's list could be read many times over without one response landing.

Deliver a room's changes one at a time: the first change is delivered at once, and
changes that arrive while a delivery's refetch is in flight, or within a second after
it settles, fold into one delivery after that.

* fix(copilot): purge the copilot backend and stored files when a chat is hard-deleted

DELETE /api/copilot/chat/delete removed the copilot_chats row but never told the
copilot backend, so the backend's chat, events and runs for that chat were orphaned
for good, and the chat's copilot/inline-image storage objects were never deleted.
deleteCopilotChat now collects the chat's file refs with prepareChatCleanup before
the delete transaction (the message rows that name them cascade away with it) and
runs execute() after commit, which re-checks the row is gone, calls the backend's
/api/tasks/cleanup and deletes the files. A failed purge is logged and does not turn
the committed delete into an error response.

Autopsy: the only callers of the backend cleanup were the retention jobs and the
organization cleanup, and no test covered deleteCopilotChat's side effects outside
Sim's rows: the integration suite only asserted that it stopped live runs. The new
deletion.test.ts asserts the purge is prepared for the chat and executed after the
delete, and fails on the previous code.

Adds a central @sim/testing mock for @/lib/cleanup/chat-cleanup (its fourth test
user) and moves the three existing local factories onto it.

* fix(copilot): queue copilot backend cleanup for chats an account removal cascades away

copilot_chats cascades from user, workspace and workflow (and workflow from user and
workspace), so deleting an account through POST /api/users/me/deletion, or a user row
through /admin/remove-user, erased chat rows without ever telling the copilot backend,
which kept those chats' data forever.

A new outbox event, copilot.chat.backend-cleanup, carries up to 100 chat ids. Its
handler purges only chats whose rows are gone and throws on a failed backend call, so
the outbox retries it. Account deletion collects the chats its workspace and user
deletes cascade (the account's own chats, chats in the doomed workspaces, and chats on
workflows the account created or those workspaces hold) and queues the events in its
transaction, so a rollback queues nothing. /admin/remove-user deletes outside any
transaction, so its Better Auth delete hooks read the chat ids before the delete and
queue the events after it commits, as they already do for run stops. Nothing is queued
without COPILOT_API_KEY, matching memory cleanup.

Workspace and workflow rows have no other hard-delete path: the UI archives them, and
the soft-delete retention job already runs prepareChatCleanup for the workflows it
purges.

Autopsy: the account-removal integration suites asserted run stops, storage and memory
cleanup, but no test looked at what the copilot backend kept for the chats the
cascade erased. chat-backend-cleanup.integration.ts drives both removal paths against
real PostgreSQL and a local backend fixture and asserts exactly the erased chats are
purged; its account-deletion case fails on the previous code.

* improvement(outbox): prune completed checkpoint expiry, file, memory and chat cleanup events

Retention pruned completed rows of only two event types, so outbox_event kept every
other completed row forever; the embedding checkpoint expiry events alone are most of
the table. Completed rows older than the 7-day window are now also pruned for:

- the three checkpoint expiry events. They are deterministic-id gates: a writer inserts
  the row before uploading and skips the upload once the row is no longer pending. The
  row completes at the checkpoint's 48-hour expiry, and after it is pruned the next
  writer of that key inserts a fresh pending cleanup before uploading, so no object is
  left without a cleanup. The new integration case drives writeFramedCheckpoint through
  completion, pruning and a rewrite to prove it.
- workspace-file.live-doc.reconcile, workspace-file.storage.cleanup,
  mothership.memory.cleanup and copilot.chat.backend-cleanup: random ids, looked up only
  by the immediate post-commit process call while pending.

Still kept: workflow deployment and workspace operation events (workspace operation
receipts read their effect rows back by id and treat a missing row as a failed
effect), enterprise/admin operation records, idempotency-keyed events, and every
pending, processing and dead-letter row. The per-type batch and cadence are unchanged,
so an existing backlog drains over successive maintenance runs.

* perf(idempotency): drop the full-table stat scans and resume each cleanup batch where the last ended

The daily /api/webhooks/cleanup/idempotency run read the whole idempotency_key table
four times for its before/after stats (count/min/max and a GROUP BY namespace, twice),
which only fed two log lines and the cron response. It now reports what it actually
deleted per namespace, counted from the delete's returned keys, and an estimated row
count from pg_class.reltuples (one catalog read) after the sweep.

The sweep itself started every batch from the oldest expired row. The global sweep
keeps the permanent admin-credit-grant and workflow-execution-id claims, whose expired
rows sit at the front of the created_at order forever, so each 1,000-row batch walked
all of them again before finding anything to delete. Batches are now ordered by
created_at and resume at the timestamp the previous successful batch ended on (>=, so
keys sharing that timestamp are not skipped), and the retained rows are read once per
run instead of once per batch.

The new integration test sweeps expired keys across several small batches, including
a tie on the boundary timestamp, and asserts every expired key goes while the
permanent and fresh ones stay; it fails if the resume skips the tie.

* perf(db): drop prefix-covered indexes and vacuum churn-heavy tables sooner

Drop 32 b-tree indexes whose columns are a leading prefix of another
non-partial b-tree on the same table, so the remaining index serves every
lookup, range scan and foreign-key check they did. Each coverer is named
beside its drop in the migration. Indexes whose only longer sibling is
partial (user_table_def_workspace_id_idx, workflow_mcp_tool_server_id_idx)
stay, since a partial index cannot serve FK checks.

Lower the vacuum and insert-vacuum scale factors to 2% on embedding,
workspace_files and copilot_async_tool_calls, which accumulate dead and
unvacuumed rows well past the 20% defaults between runs. embedding_search
is left alone: each vacuum there also scans its HNSW index.

* perf(db): cover usage_log period sums with one index led by both period stamps

The period-sum read (entity plus both period stamps) bounded its scan on
only three columns of the covering index, because user_id and created_at
sat between the two stamps. With the planner also underestimating that
match (the stamps are functionally dependent on the entity), it chose the
slimmer index without cost and fetched every matched row from the heap.

One index keyed (entity type, entity id, period start, period end,
user_id) with created_at, source and cost as trailing payload now serves
the period sums, the per-user sums, the weekly-refresh buckets and the
cycle-close range index-only, and replaces both previous indexes.

A replay after a failed build moves only an invalid copy of the new index
aside to be dropped, so a valid index an earlier attempt built is kept
rather than rebuilt.

* chore(outbox): read the retention suite's database through a typed handle

The checkpoint-gate case added three more non-null assertions on the hoisted database holder. The suite now keeps the drizzle handle it builds and reads through it, so the file carries none and its explicit-any baseline entry goes.

* improvement(files): drop the unobserved folder-tree room callback memo and a pre-existing-behavior test

useWorkspaceInvalidationRoom reads its callback through a ref and no effect lists it, so the folder tree's useCallback bought nothing. The cross-workspace broadcast test guarded a filter the coalescing change left untouched.

* revert(billing): read the usage gate fresh for wand and chat-title model calls

Reverts c921209. The cached admission it routed both calls through also
caches the payer's billing-block result, so a payer blocked for payment after
an admission was cached still passed for up to the cache TTL. The saved ledger
sum per call was small next to that. The requestChatTitle test's cache reset
existed only for that cached path and goes with it.

* fix(copilot): record a deleted chat's backend and file purge durably with its delete

A permanent chat delete purged the copilot backend and the chat's stored files
after commit and only logged a failure, so a failed purge was lost: the chat row
was gone and nothing could find it again. /admin/remove-user read the cascaded
chats in its before hook and queued their backend cleanup in its after hook,
logging either failure and letting the removal proceed with nothing recorded,
and neither removal path purged the stored files of the chats it erased.

The copilot.chat.backend-cleanup event now also carries the chats' stored-file
keys, collected while the rows naming them exist (100 chat ids or 100 files per
event). Its handler purges the backend data and deletes the objects of chats
whose rows are gone, and throws on any failure, an already-missing object
excepted, so the outbox retries; chat ids reaching a processor without
COPILOT_API_KEY throw as the organization cleanup does. File events are queued
even without the key.

- deleteCopilotChat reads the keys under the chat row's lock and queues the
  event in its transaction, so the purge is recorded exactly when the delete
  commits; the post-commit best-effort purge is gone.
- Account deletion reads the chats and keys before its transaction, as its
  storage sweep does, re-reads the chat ids under the account lock, reads the
  keys of chats created since, and queues it all in the transaction.
- The admin remove-user before hook queues the event before the delete, due an
  hour later so the delete has committed; a failed read or enqueue throws,
  which makes Better Auth abort the delete. A delete that never happens leaves
  the chats alive, and the handler purges nothing for them.

collectChatFiles moves to chat-files.ts so the auth graph does not load the
backend and storage clients; it still adds five modules to the routes that
import auth, and their module baselines are re-recorded.

Autopsy: deletion.test.ts only asserted calls on a mocked prepareChatCleanup,
so it could not see that the purge was not durable, and the integration
stand-in accepted any request. The integration suite now drives the real chat
delete, account deletion and admin hook against PostgreSQL, local storage and
a stand-in that refuses anything but an authenticated POST /api/tasks/cleanup:
a rolled-back delete records nothing, a failed backend call or object delete
is retried until it succeeds, a missing key holds the backend purge, and a
failed enqueue refuses the admin removal. Seven of its eight cases fail on the
previous code.

* docs(knowledge): say what the windowed reconciliation counts do and do not guarantee

The per-window sums are not one snapshot. They differ from one only by the
rows writers outside the sync lease change during the count, and they only
gate whether a walk runs: every soft and hard delete re-checks absence in its
own statement, as it already had to after the single count statement.

* chore(copilot): prove the cleanup rollback tests fail on the fixture trigger, not any error

* docs(auth): state what the admin removal cleanup hook cannot cover

* revert(audit): keep a run's rollup window at two seconds

Reverts 03ca9c5. A rollup holds entries in process memory until its window closes, so a minute-long window widens what a crashed process loses from the audit trail thirty-fold. Platforms that aggregate audit events summarize records already written durably; keep the short window.

* perf(billing): bound cycle close's previous-period lookup below the elapsed period

The covering usage_log period index added in 0441 has unique keys, so B-tree deduplication can't compress an entity's current period the way the old four-column index did, and max(billing_period_start) for the previous period walked every current-period entry first. The elapsed period starts before it ends, so bounding the start below it changes no answer and lets the scan skip the current period.

Author: waleedlatif1

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)