fix migration

Author: TheodoreSpeaks

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

@@ -1,7 +1,7 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { getJobQueue } from '@/lib/core/async-jobs'
+import { dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
 
 export const dynamic = 'force-dynamic'
 
@@ -12,14 +12,13 @@ export async function GET(request: NextRequest) {
     const authError = verifyCronAuth(request, 'soft-delete cleanup')
     if (authError) return authError
 
-    const jobQueue = await getJobQueue()
-    const jobId = await jobQueue.enqueue('cleanup-soft-deletes', {})
+    const result = await dispatchCleanupJobs('cleanup-soft-deletes', 'softDeleteRetentionHours')
 
-    logger.info('Soft-delete cleanup job dispatched', { jobId })
+    logger.info('Soft-delete cleanup jobs dispatched', result)
 
-    return NextResponse.json({ triggered: true, jobId })
+    return NextResponse.json({ triggered: true, ...result })
   } catch (error) {
-    logger.error('Failed to dispatch soft-delete cleanup job:', { error })
+    logger.error('Failed to dispatch soft-delete cleanup jobs:', { error })
     return NextResponse.json({ error: 'Failed to dispatch soft-delete cleanup' }, { status: 500 })
   }
 }

apps/sim/app/api/cron/cleanup-tasks/route.ts

@@ -1,7 +1,7 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { getJobQueue } from '@/lib/core/async-jobs'
+import { dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
 
 export const dynamic = 'force-dynamic'
 
@@ -12,14 +12,13 @@ export async function GET(request: NextRequest) {
     const authError = verifyCronAuth(request, 'task cleanup')
     if (authError) return authError
 
-    const jobQueue = await getJobQueue()
-    const jobId = await jobQueue.enqueue('cleanup-tasks', {})
+    const result = await dispatchCleanupJobs('cleanup-tasks', 'taskCleanupHours')
 
-    logger.info('Task cleanup job dispatched', { jobId })
+    logger.info('Task cleanup jobs dispatched', result)
 
-    return NextResponse.json({ triggered: true, jobId })
+    return NextResponse.json({ triggered: true, ...result })
   } catch (error) {
-    logger.error('Failed to dispatch task cleanup job:', { error })
+    logger.error('Failed to dispatch task cleanup jobs:', { error })
     return NextResponse.json({ error: 'Failed to dispatch task cleanup' }, { status: 500 })
   }
 }

apps/sim/app/api/cron/redact-task-context/route.ts

@@ -1,7 +1,7 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { getJobQueue } from '@/lib/core/async-jobs'
+import { dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
 
 export const dynamic = 'force-dynamic'
 
@@ -12,14 +12,13 @@ export async function GET(request: NextRequest) {
     const authError = verifyCronAuth(request, 'logs cleanup')
     if (authError) return authError
 
-    const jobQueue = await getJobQueue()
-    const jobId = await jobQueue.enqueue('cleanup-logs', {})
+    const result = await dispatchCleanupJobs('cleanup-logs', 'logRetentionHours')
 
-    logger.info('Log cleanup job dispatched', { jobId })
+    logger.info('Log cleanup jobs dispatched', result)
 
-    return NextResponse.json({ triggered: true, jobId })
+    return NextResponse.json({ triggered: true, ...result })
   } catch (error) {
-    logger.error('Failed to dispatch log cleanup job:', { error })
+    logger.error('Failed to dispatch log cleanup jobs:', { error })
     return NextResponse.json({ error: 'Failed to dispatch log cleanup' }, { status: 500 })
   }
 }

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

@@ -6,6 +6,7 @@ import { type NextRequest, NextResponse } from 'next/server'
 import { z } from 'zod'
 import { AuditAction, AuditResourceType, recordAudit } from '@/lib/audit/log'
 import { getSession } from '@/lib/auth'
+import { getRetentionDefaultHours } from '@/lib/billing/cleanup-dispatcher'
 import { getHighestPrioritySubscription } from '@/lib/billing/core/plan'
 import { isEnterprisePlan } from '@/lib/billing/core/subscription'
 import {
@@ -21,40 +22,19 @@ const logger = createLogger('DataRetentionAPI')
 const MIN_HOURS = 24
 const MAX_HOURS = 43800 // 5 years
 
-const FREE_LOG_RETENTION_HOURS = 7 * 24
-const FREE_SOFT_DELETE_RETENTION_HOURS = 7 * 24
-const FREE_TASK_REDACTION_HOURS = null // never
-const FREE_TASK_CLEANUP_HOURS = null // never
-
-const PRO_LOG_RETENTION_HOURS = 30 * 24
-const PRO_SOFT_DELETE_RETENTION_HOURS = 30 * 24
-const PRO_TASK_REDACTION_HOURS = 30 * 24
-const PRO_TASK_CLEANUP_HOURS = null // never
-
 interface PlanDefaults {
-  logRetentionHours: number
-  softDeleteRetentionHours: number
-  taskRedactionHours: number | null
+  logRetentionHours: number | null
+  softDeleteRetentionHours: number | null
   taskCleanupHours: number | null
 }
 
 function getPlanDefaults(plan: 'free' | 'pro' | 'enterprise'): PlanDefaults {
-  switch (plan) {
-    case 'enterprise':
-    case 'pro':
-      return {
-        logRetentionHours: PRO_LOG_RETENTION_HOURS,
-        softDeleteRetentionHours: PRO_SOFT_DELETE_RETENTION_HOURS,
-        taskRedactionHours: PRO_TASK_REDACTION_HOURS,
-        taskCleanupHours: PRO_TASK_CLEANUP_HOURS,
-      }
-    default:
-      return {
-        logRetentionHours: FREE_LOG_RETENTION_HOURS,
-        softDeleteRetentionHours: FREE_SOFT_DELETE_RETENTION_HOURS,
-        taskRedactionHours: FREE_TASK_REDACTION_HOURS,
-        taskCleanupHours: FREE_TASK_CLEANUP_HOURS,
-      }
+  const tier: 'free' | 'paid' | 'enterprise' =
+    plan === 'free' ? 'free' : plan === 'enterprise' ? 'enterprise' : 'paid'
+  return {
+    logRetentionHours: getRetentionDefaultHours(tier, 'logRetentionHours'),
+    softDeleteRetentionHours: getRetentionDefaultHours(tier, 'softDeleteRetentionHours'),
+    taskCleanupHours: getRetentionDefaultHours(tier, 'taskCleanupHours'),
   }
 }
 
@@ -71,7 +51,6 @@ async function resolveWorkspacePlan(
 const updateRetentionSchema = z.object({
   logRetentionHours: z.number().int().min(MIN_HOURS).max(MAX_HOURS).nullable().optional(),
   softDeleteRetentionHours: z.number().int().min(MIN_HOURS).max(MAX_HOURS).nullable().optional(),
-  taskRedactionHours: z.number().int().min(MIN_HOURS).max(MAX_HOURS).nullable().optional(),
   taskCleanupHours: z.number().int().min(MIN_HOURS).max(MAX_HOURS).nullable().optional(),
 })
 
@@ -98,7 +77,6 @@ export async function GET(_request: NextRequest, { params }: { params: Promise<{
       .select({
         logRetentionHours: workspace.logRetentionHours,
         softDeleteRetentionHours: workspace.softDeleteRetentionHours,
-        taskRedactionHours: workspace.taskRedactionHours,
         taskCleanupHours: workspace.taskCleanupHours,
         billedAccountUserId: workspace.billedAccountUserId,
       })
@@ -123,20 +101,17 @@ export async function GET(_request: NextRequest, { params }: { params: Promise<{
         configured: {
           logRetentionHours: ws.logRetentionHours,
           softDeleteRetentionHours: ws.softDeleteRetentionHours,
-          taskRedactionHours: ws.taskRedactionHours,
           taskCleanupHours: ws.taskCleanupHours,
         },
         effective: isEnterpriseWorkspace
           ? {
               logRetentionHours: ws.logRetentionHours,
               softDeleteRetentionHours: ws.softDeleteRetentionHours,
-              taskRedactionHours: ws.taskRedactionHours,
               taskCleanupHours: ws.taskCleanupHours,
             }
           : {
               logRetentionHours: defaults.logRetentionHours,
               softDeleteRetentionHours: defaults.softDeleteRetentionHours,
-              taskRedactionHours: defaults.taskRedactionHours,
               taskCleanupHours: defaults.taskCleanupHours,
             },
       },
@@ -196,10 +171,7 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
     if (parsed.data.softDeleteRetentionHours !== undefined) {
       updateData.softDeleteRetentionHours = parsed.data.softDeleteRetentionHours
     }
-    if (parsed.data.taskRedactionHours !== undefined) {
-      updateData.taskRedactionHours = parsed.data.taskRedactionHours
-    }
-    if (parsed.data.taskCleanupHours !== undefined) {
+if (parsed.data.taskCleanupHours !== undefined) {
       updateData.taskCleanupHours = parsed.data.taskCleanupHours
     }
 
@@ -210,7 +182,6 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
       .returning({
         logRetentionHours: workspace.logRetentionHours,
         softDeleteRetentionHours: workspace.softDeleteRetentionHours,
-        taskRedactionHours: workspace.taskRedactionHours,
         taskCleanupHours: workspace.taskCleanupHours,
       })
 
@@ -242,13 +213,11 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
         configured: {
           logRetentionHours: updated.logRetentionHours,
           softDeleteRetentionHours: updated.softDeleteRetentionHours,
-          taskRedactionHours: updated.taskRedactionHours,
           taskCleanupHours: updated.taskCleanupHours,
         },
         effective: {
           logRetentionHours: updated.logRetentionHours,
           softDeleteRetentionHours: updated.softDeleteRetentionHours,
-          taskRedactionHours: updated.taskRedactionHours,
           taskCleanupHours: updated.taskCleanupHours,
         },
       },

apps/sim/app/api/workspaces/[id]/data-retention/route.ts

@@ -1,10 +1,14 @@
 import { db } from '@sim/db'
-import { workflowExecutionLogs } from '@sim/db/schema'
+import { workflowExecutionLogs, workspace } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { task } from '@trigger.dev/sdk'
-import { and, inArray, lt } from 'drizzle-orm'
+import { and, eq, inArray, lt } from 'drizzle-orm'
 import { snapshotService } from '@/lib/logs/execution/snapshot/service'
-import { resolveRetentionGroups } from '@/lib/retention/workspace-tiers'
+import {
+  type CleanupJobPayload,
+  getRetentionDefaultHours,
+  resolveTierWorkspaceIds,
+} from '@/lib/billing/cleanup-dispatcher'
 import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
 
 const logger = createLogger('CleanupLogs')
@@ -102,36 +106,79 @@ async function cleanupTier(
   return results
 }
 
-export const cleanupLogsTask = task({
-  id: 'cleanup-logs',
-  run: async () => {
-    const startTime = Date.now()
+async function resolvePayload(payload: CleanupJobPayload): Promise<{
+  workspaceIds: string[]
+  retentionHours: number
+  tierLabel: string
+} | null> {
+  if (payload.tier === 'free' || payload.tier === 'paid') {
+    const retentionHours = getRetentionDefaultHours(payload.tier, 'logRetentionHours')
+    if (retentionHours === null) {
+      logger.info(`[${payload.tier}] No default retention, skipping`)
+      return null
+    }
+    const workspaceIds = await resolveTierWorkspaceIds(payload.tier)
+    return { workspaceIds, retentionHours, tierLabel: payload.tier }
+  }
+
+  // enterprise
+  const [ws] = await db
+    .select({ logRetentionHours: workspace.logRetentionHours })
+    .from(workspace)
+    .where(eq(workspace.id, payload.workspaceId))
+    .limit(1)
 
-    logger.info('Starting log cleanup task')
+  if (!ws?.logRetentionHours) {
+    logger.info(`[enterprise/${payload.workspaceId}] No retention configured, skipping`)
+    return null
+  }
+
+  return {
+    workspaceIds: [payload.workspaceId],
+    retentionHours: ws.logRetentionHours,
+    tierLabel: `enterprise/${payload.workspaceId}`,
+  }
+}
 
-    const groups = await resolveRetentionGroups('logRetentionHours')
+export async function runCleanupLogs(payload: CleanupJobPayload): Promise<void> {
+  const startTime = Date.now()
 
-    for (const group of groups) {
-      const results = await cleanupTier(group.workspaceIds, group.retentionDate, group.tierLabel)
-      logger.info(`[${group.tierLabel}] Result: ${results.deleted} deleted, ${results.deleteFailed} failed out of ${results.total} candidates`)
-    }
+  const resolved = await resolvePayload(payload)
+  if (!resolved) return
+
+  const { workspaceIds, retentionHours, tierLabel } = resolved
 
-    // Snapshot cleanup — use shortest retention + 1 day
+  if (workspaceIds.length === 0) {
+    logger.info(`[${tierLabel}] No workspaces to process`)
+    return
+  }
+
+  const retentionDate = new Date(Date.now() - retentionHours * 60 * 60 * 1000)
+  logger.info(
+    `[${tierLabel}] Cleaning ${workspaceIds.length} workspaces, cutoff: ${retentionDate.toISOString()}`
+  )
+
+  const results = await cleanupTier(workspaceIds, retentionDate, tierLabel)
+  logger.info(
+    `[${tierLabel}] Result: ${results.deleted} deleted, ${results.deleteFailed} failed out of ${results.total} candidates`
+  )
+
+  // Snapshot cleanup runs only on the free job to avoid running it N times for N enterprise workspaces.
+  if (payload.tier === 'free') {
     try {
-      const shortestDays = Math.min(
-        ...groups.map((g) => (Date.now() - g.retentionDate.getTime()) / (24 * 60 * 60 * 1000))
-      )
-      if (Number.isFinite(shortestDays)) {
-        const snapshotsCleaned = await snapshotService.cleanupOrphanedSnapshots(
-          Math.floor(shortestDays) + 1
-        )
-        logger.info(`Cleaned up ${snapshotsCleaned} orphaned snapshots`)
-      }
+      const retentionDays = Math.floor(retentionHours / 24)
+      const snapshotsCleaned = await snapshotService.cleanupOrphanedSnapshots(retentionDays + 1)
+      logger.info(`Cleaned up ${snapshotsCleaned} orphaned snapshots`)
     } catch (snapshotError) {
       logger.error('Error cleaning up orphaned snapshots:', { snapshotError })
     }
+  }
+
+  const timeElapsed = (Date.now() - startTime) / 1000
+  logger.info(`[${tierLabel}] Job completed in ${timeElapsed.toFixed(2)}s`)
+}
 
-    const timeElapsed = (Date.now() - startTime) / 1000
-    logger.info(`Log cleanup task completed in ${timeElapsed.toFixed(2)}s`)
-  },
+export const cleanupLogsTask = task({
+  id: 'cleanup-logs',
+  run: runCleanupLogs,
 })