feat(cleanup): add bounded manual retention runs

apps/sim/app/api/cron/cleanup-soft-deletes/route.ts

@@ -1,18 +1,36 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
+import { softDeletesCleanupContract } from '@/lib/api/contracts/cleanup'
+import { parseRequest } from '@/lib/api/server/validation'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
+import { dispatchBoundedCleanup, dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 
 export const dynamic = 'force-dynamic'
 
 const logger = createLogger('SoftDeleteCleanupAPI')
 
+/** Cron-secret maintenance protocol is global; workspace principal authorization does not apply. */
 export const GET = withRouteHandler(async (request: NextRequest) => {
   try {
     const authError = verifyCronAuth(request, 'soft-delete cleanup')
     if (authError) return authError
 
+    const parsed = await parseRequest(
+      softDeletesCleanupContract,
+      request,
+      {},
+      {
+        rejectDuplicateQueryValues: true,
+        rejectBlankQueryValues: true,
+      }
+    )
+    if (!parsed.success) return parsed.response
+    if (parsed.data.query) {
+      const result = await dispatchBoundedCleanup('cleanup-soft-deletes', parsed.data.query)
+      return NextResponse.json(result, { status: 202 })
+    }
+
     const result = await dispatchCleanupJobs('cleanup-soft-deletes')
 
     logger.info('Soft-delete cleanup jobs dispatched', result)

apps/sim/app/api/logs/cleanup/route.test.ts

@@ -0,0 +1,75 @@
+import { createMockRequest } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { auth, bounded, scheduled } = vi.hoisted(() => ({
+  auth: vi.fn(),
+  bounded: vi.fn(),
+  scheduled: vi.fn(),
+}))
+vi.mock('@/lib/auth/internal', () => ({ verifyCronAuth: auth }))
+vi.mock('@/lib/billing/cleanup-dispatcher', () => ({
+  dispatchBoundedCleanup: bounded,
+  dispatchCleanupJobs: scheduled,
+}))
+
+import { GET as softDeletes } from '@/app/api/cron/cleanup-soft-deletes/route'
+import { GET as logs } from '@/app/api/logs/cleanup/route'
+
+for (const [path, GET, type, limit] of [
+  ['/api/logs/cleanup', logs, 'cleanup-logs', 'workflowLogs'],
+  ['/api/cron/cleanup-soft-deletes', softDeletes, 'cleanup-soft-deletes', 'workflows'],
+] as const) {
+  describe(path, () => {
+    beforeEach(() => {
+      vi.clearAllMocks()
+      auth.mockReturnValue(null)
+      bounded.mockResolvedValue({ runId: 'run-one', mode: 'bounded' })
+      scheduled.mockResolvedValue({
+        jobIds: ['batch-one'],
+        jobCount: 1,
+        chunkCount: 2,
+        workspaceCount: 3,
+      })
+    })
+    const request = (query = '') =>
+      createMockRequest('GET', undefined, {}, `http://localhost:3000${path}${query}`)
+    it('authenticates before parsing invalid limits', async () => {
+      auth.mockReturnValue(new Response(null, { status: 401 }))
+      expect((await GET(request('?unknown=1'))).status).toBe(401)
+      expect(bounded).not.toHaveBeenCalled()
+      expect(scheduled).not.toHaveBeenCalled()
+    })
+    it('keeps no-parameter scheduled dispatch unchanged', async () => {
+      const response = await GET(request())
+      expect(response.status).toBe(200)
+      expect(scheduled).toHaveBeenCalledWith(type)
+      expect(bounded).not.toHaveBeenCalled()
+    })
+    it('accepts one bounded run', async () => {
+      const response = await GET(request(`?${limit}=2&requestId=wave-1&dryRun=true`))
+      expect(response.status).toBe(202)
+      expect(bounded).toHaveBeenCalledWith(
+        type,
+        expect.objectContaining({
+          limits: expect.objectContaining({ [limit]: 2 }),
+          requestId: 'wave-1',
+          dryRun: true,
+          batchSize: 25,
+        })
+      )
+      expect(scheduled).not.toHaveBeenCalled()
+    })
+    it.each(['?dryRun=true', '?unknown=1', '?batchSize=3', `?${limit}=2&${limit}=3&requestId=r`])(
+      'rejects invalid query %s',
+      async (query) => {
+        expect((await GET(request(query))).status).toBe(400)
+        expect(bounded).not.toHaveBeenCalled()
+        expect(scheduled).not.toHaveBeenCalled()
+      }
+    )
+    it('reports a dispatch failure', async () => {
+      bounded.mockRejectedValue(new Error('Trigger unavailable'))
+      expect((await GET(request(`?${limit}=2&requestId=r`))).status).toBe(500)
+    })
+  })
+}

apps/sim/app/api/logs/cleanup/route.ts

@@ -1,18 +1,36 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
+import { logsCleanupContract } from '@/lib/api/contracts/cleanup'
+import { parseRequest } from '@/lib/api/server/validation'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
+import { dispatchBoundedCleanup, dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 
 export const dynamic = 'force-dynamic'
 
 const logger = createLogger('LogsCleanupAPI')
 
+/** Cron-secret maintenance protocol is global; workspace principal authorization does not apply. */
 export const GET = withRouteHandler(async (request: NextRequest) => {
   try {
     const authError = verifyCronAuth(request, 'logs cleanup')
     if (authError) return authError
 
+    const parsed = await parseRequest(
+      logsCleanupContract,
+      request,
+      {},
+      {
+        rejectDuplicateQueryValues: true,
+        rejectBlankQueryValues: true,
+      }
+    )
+    if (!parsed.success) return parsed.response
+    if (parsed.data.query) {
+      const result = await dispatchBoundedCleanup('cleanup-logs', parsed.data.query)
+      return NextResponse.json(result, { status: 202 })
+    }
+
     const result = await dispatchCleanupJobs('cleanup-logs')
 
     logger.info('Log cleanup jobs dispatched', result)

apps/sim/app/api/webhooks/outbox/process/route.ts

@@ -9,6 +9,7 @@ import { enterpriseOwnerClaimOutboxHandlers } from '@/lib/billing/enterprise-own
 import { enterpriseIssuanceOutboxHandlers } from '@/lib/billing/enterprise-provisioning'
 import { membershipBillingOutboxHandlers } from '@/lib/billing/organizations/membership-reconciliation'
 import { billingOutboxHandlers } from '@/lib/billing/webhooks/outbox-handlers'
+import { retentionStorageOutboxHandlers } from '@/lib/cleanup/storage-outbox'
 import { processOutboxEvents } from '@/lib/core/outbox/service'
 import { DeadlineExceededError } from '@/lib/core/utils/deadline'
 import { generateRequestId } from '@/lib/core/utils/request'
@@ -33,6 +34,7 @@ export const dynamic = 'force-dynamic'
 export const maxDuration = 800
 
 const handlers = {
+  ...retentionStorageOutboxHandlers,
   ...slackSearchOutboxHandlers,
   ...adminInvitationOperationOutboxHandlers,
   ...adminMemberOperationOutboxHandlers,