feat(execution-queuing): async api mode + ratelimiting by subscription tier (#702)

* v1 queuing system

* working async queue

* working impl of sync + async request formats

* fix tests

* fix rate limit calc

* fix rate limiting issues

* regen migration

* fix test

* fix instrumentation script issues

* remove use workflow queue env var

* make modal have async examples

* Remove conflicting 54th migration before merging staging

* new migration files

* remove console log

* update modal correctly

* working sync executor

* works for sync

* remove useless stats endpoint

* fix tests

* add sync exec timeout

* working impl with cron job

* migrate to trigger.dev

* remove migration

* remove unused code

* update readme

* restructure jobs API response

* add logging for async execs

* improvement: example ui/ux

* use getBaseUrl() func

---------

Co-authored-by: Waleed Latif <walif6@gmail.com>
Co-authored-by: Emir Karabeg <emirkarabeg@berkeley.edu>

Author: 3 people

README.md

@@ -147,6 +147,7 @@ bun run dev:sockets
 - **Docs**: [Fumadocs](https://fumadocs.vercel.app/)
 - **Monorepo**: [Turborepo](https://turborepo.org/)
 - **Realtime**: [Socket.io](https://socket.io/)
+- **Background Jobs**: [Trigger.dev](https://trigger.dev/)
 
 ## Contributing
 

apps/sim/.gitignore

@@ -50,3 +50,5 @@ next-env.d.ts
 
 # Uploads
 /uploads
+
+.trigger

apps/sim/app/api/chat/utils.test.ts

@@ -7,6 +7,38 @@ import type { NextResponse } from 'next/server'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { env } from '@/lib/env'
 
+// Mock all the problematic imports that cause timeouts
+vi.mock('@/db', () => ({
+  db: {
+    select: vi.fn(),
+    update: vi.fn(),
+  },
+}))
+
+vi.mock('@/lib/utils', () => ({
+  decryptSecret: vi.fn().mockResolvedValue({ decrypted: 'test-secret' }),
+}))
+
+vi.mock('@/lib/logs/enhanced-logging-session', () => ({
+  EnhancedLoggingSession: vi.fn().mockImplementation(() => ({
+    safeStart: vi.fn().mockResolvedValue(undefined),
+    safeComplete: vi.fn().mockResolvedValue(undefined),
+    safeCompleteWithError: vi.fn().mockResolvedValue(undefined),
+  })),
+}))
+
+vi.mock('@/executor', () => ({
+  Executor: vi.fn(),
+}))
+
+vi.mock('@/serializer', () => ({
+  Serializer: vi.fn(),
+}))
+
+vi.mock('@/stores/workflows/server-utils', () => ({
+  mergeSubblockState: vi.fn().mockReturnValue({}),
+}))
+
 describe('Chat API Utils', () => {
   beforeEach(() => {
     vi.resetModules()

apps/sim/app/api/jobs/[jobId]/route.ts

@@ -0,0 +1,110 @@
+import { runs } from '@trigger.dev/sdk/v3'
+import { eq } from 'drizzle-orm'
+import { type NextRequest, NextResponse } from 'next/server'
+import { getSession } from '@/lib/auth'
+import { createLogger } from '@/lib/logs/console-logger'
+import { db } from '@/db'
+import { apiKey as apiKeyTable } from '@/db/schema'
+import { createErrorResponse } from '../../workflows/utils'
+
+const logger = createLogger('TaskStatusAPI')
+
+export async function GET(
+  request: NextRequest,
+  { params }: { params: Promise<{ jobId: string }> }
+) {
+  const { jobId: taskId } = await params
+  const requestId = crypto.randomUUID().slice(0, 8)
+
+  try {
+    logger.debug(`[${requestId}] Getting status for task: ${taskId}`)
+
+    // Try session auth first (for web UI)
+    const session = await getSession()
+    let authenticatedUserId: string | null = session?.user?.id || null
+
+    if (!authenticatedUserId) {
+      const apiKeyHeader = request.headers.get('x-api-key')
+      if (apiKeyHeader) {
+        const [apiKeyRecord] = await db
+          .select({ userId: apiKeyTable.userId })
+          .from(apiKeyTable)
+          .where(eq(apiKeyTable.key, apiKeyHeader))
+          .limit(1)
+
+        if (apiKeyRecord) {
+          authenticatedUserId = apiKeyRecord.userId
+        }
+      }
+    }
+
+    if (!authenticatedUserId) {
+      return createErrorResponse('Authentication required', 401)
+    }
+
+    // Fetch task status from Trigger.dev
+    const run = await runs.retrieve(taskId)
+
+    logger.debug(`[${requestId}] Task ${taskId} status: ${run.status}`)
+
+    // Map Trigger.dev status to our format
+    const statusMap = {
+      QUEUED: 'queued',
+      WAITING_FOR_DEPLOY: 'queued',
+      EXECUTING: 'processing',
+      RESCHEDULED: 'processing',
+      FROZEN: 'processing',
+      COMPLETED: 'completed',
+      CANCELED: 'cancelled',
+      FAILED: 'failed',
+      CRASHED: 'failed',
+      INTERRUPTED: 'failed',
+      SYSTEM_FAILURE: 'failed',
+      EXPIRED: 'failed',
+    } as const
+
+    const mappedStatus = statusMap[run.status as keyof typeof statusMap] || 'unknown'
+
+    // Build response based on status
+    const response: any = {
+      success: true,
+      taskId,
+      status: mappedStatus,
+      metadata: {
+        startedAt: run.startedAt,
+      },
+    }
+
+    // Add completion details if finished
+    if (mappedStatus === 'completed') {
+      response.output = run.output // This contains the workflow execution results
+      response.metadata.completedAt = run.finishedAt
+      response.metadata.duration = run.durationMs
+    }
+
+    // Add error details if failed
+    if (mappedStatus === 'failed') {
+      response.error = run.error
+      response.metadata.completedAt = run.finishedAt
+      response.metadata.duration = run.durationMs
+    }
+
+    // Add progress info if still processing
+    if (mappedStatus === 'processing' || mappedStatus === 'queued') {
+      response.estimatedDuration = 180000 // 3 minutes max from our config
+    }
+
+    return NextResponse.json(response)
+  } catch (error: any) {
+    logger.error(`[${requestId}] Error fetching task status:`, error)
+
+    if (error.message?.includes('not found') || error.status === 404) {
+      return createErrorResponse('Task not found', 404)
+    }
+
+    return createErrorResponse('Failed to fetch task status', 500)
+  }
+}
+
+// TODO: Implement task cancellation via Trigger.dev API if needed
+// export async function DELETE() { ... }