Function cleanupLegacyLargeExecutionValues

apps/sim/background/cleanup-logs.ts:198–349

(
  workspaceIds: string[],
  retentionDate: Date,
  label: string
)

}

async function cleanupLegacyLargeExecutionValues(
  workspaceIds: string[],
  retentionDate: Date,
  label: string
): Promise<LargeValueCleanupStats> {
  const stats: LargeValueCleanupStats = {
    largeValuesTotal: 0,
    largeValuesDeleted: 0,
    largeValuesDeleteFailed: 0,
  }
  if (workspaceIds.length === 0) return stats

  const legacyRetentionDate = new Date(
    retentionDate.getTime() - LEGACY_LARGE_VALUE_CLEANUP_GRACE_HOURS * 60 * 60 * 1000
  )
  const workspaceChunks = chunkArray(workspaceIds, 50)
  let attempted = 0

  for (const chunkIds of workspaceChunks) {
    while (attempted < LARGE_VALUE_CLEANUP_TOTAL_KEY_LIMIT) {
      const limit = Math.min(
        LARGE_VALUE_CLEANUP_BATCH_SIZE,
        LARGE_VALUE_CLEANUP_TOTAL_KEY_LIMIT - attempted
      )
      const rows = await db
        .select({ key: workspaceFiles.key })
        .from(workspaceFiles)
        .where(
          and(
            inArray(workspaceFiles.workspaceId, chunkIds),
            eq(workspaceFiles.context, 'execution'),
            isNull(workspaceFiles.deletedAt),
            lt(workspaceFiles.uploadedAt, legacyRetentionDate),
            sql`${workspaceFiles.key} LIKE 'execution/%/%/%/large-value-lv_%.json'`,
            sql`NOT EXISTS (
              SELECT 1
              FROM ${executionLargeValues} AS registered_value
              WHERE registered_value.key = ${workspaceFiles.key}
            )`,
            sql`NOT EXISTS (
              SELECT 1
              FROM ${executionLargeValueReferences} AS ref
              WHERE ref.key = ${workspaceFiles.key}
                AND (
                  (
                    ref.source = 'execution_log'
                    AND EXISTS (
                      SELECT 1
                      FROM ${workflowExecutionLogs} AS ref_wel
                      WHERE ref_wel.execution_id = ref.execution_id
                    )
                  )
                  OR (
                    ref.source = 'paused_snapshot'
                    AND EXISTS (
                      SELECT 1
                      FROM ${pausedExecutions} AS ref_pe
                      WHERE ref_pe.execution_id = ref.execution_id