improvement(cron): run stale execution and file version cleanup off the request path (#8535)

* improvement(cron): run stale execution cleanup as a background task

* fix(cron): run sync-log retention test against the cleanup task

* fix(cron): assert stale cleanup route by its response only

* fix(cron): cover stale cleanup dispatch window through the route response

* improvement(cron): dispatch file version cleanup from a background task

* fix(cron): retry and serialize cleanup dispatch on the task definition

Author: TheodoreSpeaks

apps/sim/app/api/cron/cleanup-file-versions/route.test.ts

@@ -0,0 +1,73 @@
+import { createMockRequest } from '@sim/testing'
+import { asyncJobsMock, asyncJobsMockFns } from '@sim/testing/mocks/async-jobs.mock'
+import { authInternalMock, authInternalMockFns } from '@sim/testing/mocks/auth-internal.mock'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+vi.mock('@/lib/auth/internal', () => authInternalMock)
+vi.mock('@/lib/core/async-jobs', () => asyncJobsMock)
+
+import { GET } from '@/app/api/cron/cleanup-file-versions/route'
+
+const { mockVerifyCronAuth } = authInternalMockFns
+
+const mockEnqueue = asyncJobsMockFns.mockJobQueue.enqueue
+
+function request() {
+  return createMockRequest(
+    'GET',
+    undefined,
+    {},
+    'http://localhost:3000/api/cron/cleanup-file-versions'
+  )
+}
+
+/** The job id a dispatch was keyed to, as the queue reports it back. */
+async function dispatchedJobId(): Promise<string> {
+  const response = await GET(request())
+  expect(response.status).toBe(200)
+  const body = (await response.json()) as { jobId: string }
+  return body.jobId
+}
+
+describe('file version cleanup route', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+    vi.setSystemTime(new Date('2026-10-01T03:30:00Z'))
+    mockVerifyCronAuth.mockReturnValue(null)
+    mockEnqueue.mockReset()
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('fails the cron invocation when the dispatch cannot be enqueued', async () => {
+    mockEnqueue.mockRejectedValueOnce(new Error('queue unavailable'))
+
+    const response = await GET(request())
+
+    expect(response.status).toBe(500)
+  })
+
+  describe('with a queue that keys each job by the id it is given', () => {
+    beforeEach(() => {
+      mockEnqueue.mockImplementation(
+        async (_type: string, _payload: unknown, options: { jobId: string }) => options.jobId
+      )
+    })
+
+    it('dispatches a retry on the same day as the same job', async () => {
+      const first = await dispatchedJobId()
+      vi.setSystemTime(new Date('2026-10-01T23:59:59.999Z'))
+
+      expect(await dispatchedJobId()).toBe(first)
+    })
+
+    it('dispatches a new job on the next day', async () => {
+      const first = await dispatchedJobId()
+      vi.setSystemTime(new Date('2026-10-02T00:00:00.000Z'))
+
+      expect(await dispatchedJobId()).not.toBe(first)
+    })
+  })
+})

apps/sim/app/api/cron/cleanup-file-versions/route.ts

@@ -1,24 +1,41 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
+import { getJobQueue } from '@/lib/core/async-jobs'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { CLEANUP_DISPATCH_MAX_ATTEMPTS } from '@/background/cleanup-dispatch'
 
 export const dynamic = 'force-dynamic'
 
 const logger = createLogger('FileVersionCleanupAPI')
+const FILE_VERSION_CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000
 
 /** GET /api/cron/cleanup-file-versions — dispatch retention for superseded workspace file versions. */
 export const GET = withRouteHandler(async (request: NextRequest) => {
   try {
     const authError = verifyCronAuth(request, 'file version cleanup')
     if (authError) return authError
 
-    const result = await dispatchCleanupJobs('cleanup-file-versions')
+    const queue = await getJobQueue()
+    const scheduleWindow = Math.floor(Date.now() / FILE_VERSION_CLEANUP_INTERVAL_MS)
+    const jobId = await queue.enqueue(
+      'cleanup-dispatch',
+      { jobType: 'cleanup-file-versions' },
+      {
+        maxAttempts: CLEANUP_DISPATCH_MAX_ATTEMPTS,
+        jobId: `cleanup-dispatch:cleanup-file-versions:${scheduleWindow}`,
+        name: 'File version cleanup dispatch',
+        concurrencyKey: 'cleanup-dispatch:cleanup-file-versions',
+        concurrencyLimit: 1,
+        runner: async () => {
+          const { dispatchCleanupJobs } = await import('@/lib/billing/cleanup-dispatcher')
+          return dispatchCleanupJobs('cleanup-file-versions')
+        },
+      }
+    )
 
-    logger.info('File version cleanup jobs dispatched', result)
-
-    return NextResponse.json({ triggered: true, ...result })
+    logger.info('File version cleanup dispatch enqueued', { jobId })
+    return NextResponse.json({ triggered: true, jobId })
   } catch (error) {
     logger.error('Failed to dispatch file version cleanup jobs:', { error })
     return NextResponse.json({ error: 'Failed to dispatch file version cleanup' }, { status: 500 })