feat(async-jobs): async execution with job queue backends (#3134)

* feat(async-jobs): async execution with job queue backends

* added migration

* remove unused envvar, remove extraneous comments

* ack comment

* same for db

* added dedicated async envvars for timeouts, updated helm

* updated comment

* ack comment

* migrated routes to be more restful

* ack comments

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: waleedlatif1

apps/sim/app/api/creators/[id]/route.ts

@@ -21,6 +21,7 @@ const UpdateCreatorProfileSchema = z.object({
   name: z.string().min(1, 'Name is required').max(100, 'Max 100 characters').optional(),
   profileImageUrl: z.string().optional().or(z.literal('')),
   details: CreatorProfileDetailsSchema.optional(),
+  verified: z.boolean().optional(), // Verification status (super users only)
 })
 
 // Helper to check if user has permission to manage profile
@@ -97,11 +98,29 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
       return NextResponse.json({ error: 'Profile not found' }, { status: 404 })
     }
 
-    // Check permissions
-    const canEdit = await hasPermission(session.user.id, existing[0])
-    if (!canEdit) {
-      logger.warn(`[${requestId}] User denied permission to update profile: ${id}`)
-      return NextResponse.json({ error: 'Access denied' }, { status: 403 })
+    // Verification changes require super user permission
+    if (data.verified !== undefined) {
+      const { verifyEffectiveSuperUser } = await import('@/lib/templates/permissions')
+      const { effectiveSuperUser } = await verifyEffectiveSuperUser(session.user.id)
+      if (!effectiveSuperUser) {
+        logger.warn(`[${requestId}] Non-super user attempted to change creator verification: ${id}`)
+        return NextResponse.json(
+          { error: 'Only super users can change verification status' },
+          { status: 403 }
+        )
+      }
+    }
+
+    // For non-verified updates, check regular permissions
+    const hasNonVerifiedUpdates =
+      data.name !== undefined || data.profileImageUrl !== undefined || data.details !== undefined
+
+    if (hasNonVerifiedUpdates) {
+      const canEdit = await hasPermission(session.user.id, existing[0])
+      if (!canEdit) {
+        logger.warn(`[${requestId}] User denied permission to update profile: ${id}`)
+        return NextResponse.json({ error: 'Access denied' }, { status: 403 })
+      }
     }
 
     const updateData: any = {
@@ -111,6 +130,7 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
     if (data.name !== undefined) updateData.name = data.name
     if (data.profileImageUrl !== undefined) updateData.profileImageUrl = data.profileImageUrl
     if (data.details !== undefined) updateData.details = data.details
+    if (data.verified !== undefined) updateData.verified = data.verified
 
     const updated = await db
       .update(templateCreators)

apps/sim/app/api/creators/[id]/verify/route.ts

@@ -1,9 +1,10 @@
-import { db } from '@sim/db'
+import { asyncJobs, db } from '@sim/db'
 import { workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { and, eq, lt, sql } from 'drizzle-orm'
+import { and, eq, inArray, lt, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
+import { JOB_RETENTION_HOURS, JOB_STATUS } from '@/lib/core/async-jobs'
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 
 const logger = createLogger('CleanupStaleExecutions')
@@ -80,12 +81,102 @@ export async function GET(request: NextRequest) {
 
     logger.info(`Stale execution cleanup completed. Cleaned: ${cleaned}, Failed: ${failed}`)
 
+    // Clean up stale async jobs (stuck in processing)
+    let asyncJobsMarkedFailed = 0
+
+    try {
+      const staleAsyncJobs = await db
+        .update(asyncJobs)
+        .set({
+          status: JOB_STATUS.FAILED,
+          completedAt: new Date(),
+          error: `Job terminated: stuck in processing for more than ${STALE_THRESHOLD_MINUTES} minutes`,
+          updatedAt: new Date(),
+        })
+        .where(
+          and(eq(asyncJobs.status, JOB_STATUS.PROCESSING), lt(asyncJobs.startedAt, staleThreshold))
+        )
+        .returning({ id: asyncJobs.id })
+
+      asyncJobsMarkedFailed = staleAsyncJobs.length
+      if (asyncJobsMarkedFailed > 0) {
+        logger.info(`Marked ${asyncJobsMarkedFailed} stale async jobs as failed`)
+      }
+    } catch (error) {
+      logger.error('Failed to clean up stale async jobs:', {
+        error: error instanceof Error ? error.message : String(error),
+      })
+    }
+
+    // Clean up stale pending jobs (never started, e.g., due to server crash before startJob())
+    let stalePendingJobsMarkedFailed = 0
+
+    try {
+      const stalePendingJobs = await db
+        .update(asyncJobs)
+        .set({
+          status: JOB_STATUS.FAILED,
+          completedAt: new Date(),
+          error: `Job terminated: stuck in pending state for more than ${STALE_THRESHOLD_MINUTES} minutes (never started)`,
+          updatedAt: new Date(),
+        })
+        .where(
+          and(eq(asyncJobs.status, JOB_STATUS.PENDING), lt(asyncJobs.createdAt, staleThreshold))
+        )
+        .returning({ id: asyncJobs.id })
+
+      stalePendingJobsMarkedFailed = stalePendingJobs.length
+      if (stalePendingJobsMarkedFailed > 0) {
+        logger.info(`Marked ${stalePendingJobsMarkedFailed} stale pending jobs as failed`)
+      }
+    } catch (error) {
+      logger.error('Failed to clean up stale pending jobs:', {
+        error: error instanceof Error ? error.message : String(error),
+      })
+    }
+
+    // Delete completed/failed jobs older than retention period
+    const retentionThreshold = new Date(Date.now() - JOB_RETENTION_HOURS * 60 * 60 * 1000)
+    let asyncJobsDeleted = 0
+
+    try {
+      const deletedJobs = await db
+        .delete(asyncJobs)
+        .where(
+          and(
+            inArray(asyncJobs.status, [JOB_STATUS.COMPLETED, JOB_STATUS.FAILED]),
+            lt(asyncJobs.completedAt, retentionThreshold)
+          )
+        )
+        .returning({ id: asyncJobs.id })
+
+      asyncJobsDeleted = deletedJobs.length
+      if (asyncJobsDeleted > 0) {
+        logger.info(
+          `Deleted ${asyncJobsDeleted} old async jobs (retention: ${JOB_RETENTION_HOURS}h)`
+        )
+      }
+    } catch (error) {
+      logger.error('Failed to delete old async jobs:', {
+        error: error instanceof Error ? error.message : String(error),
+      })
+    }
+
     return NextResponse.json({
       success: true,
-      found: staleExecutions.length,
-      cleaned,
-      failed,
-      thresholdMinutes: STALE_THRESHOLD_MINUTES,
+      executions: {
+        found: staleExecutions.length,
+        cleaned,
+        failed,
+        thresholdMinutes: STALE_THRESHOLD_MINUTES,
+      },
+      asyncJobs: {
+        staleProcessingMarkedFailed: asyncJobsMarkedFailed,
+        stalePendingMarkedFailed: stalePendingJobsMarkedFailed,
+        oldDeleted: asyncJobsDeleted,
+        staleThresholdMinutes: STALE_THRESHOLD_MINUTES,
+        retentionHours: JOB_RETENTION_HOURS,
+      },
     })
   } catch (error) {
     logger.error('Error in stale execution cleanup job:', error)