fix trigger dev, bullmq, inline prioritization

Author: icecrasher321

apps/sim/app/api/schedules/execute/route.ts

@@ -6,6 +6,7 @@ import { v4 as uuidv4 } from 'uuid'
 import { verifyCronAuth } from '@/lib/auth/internal'
 import { getJobQueue } from '@/lib/core/async-jobs'
 import { createBullMQJobData, isBullMQEnabled } from '@/lib/core/bullmq'
+import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { enqueueWorkspaceDispatch } from '@/lib/core/workspace-dispatch'
 import {
@@ -153,7 +154,7 @@ export async function GET(request: NextRequest) {
           `[${requestId}] Queued schedule execution task ${jobId} for workflow ${schedule.workflowId}`
         )
 
-        if (!isBullMQEnabled()) {
+        if (!isBullMQEnabled() && !isTriggerDevEnabled) {
           try {
             await jobQueue.startJob(jobId)
             const output = await executeScheduleJob(payload)

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -6,6 +6,7 @@ import { AuthType, checkHybridAuth, hasExternalApiCredentials } from '@/lib/auth
 import { admissionRejectedResponse, tryAdmit } from '@/lib/core/admission/gate'
 import { getJobQueue } from '@/lib/core/async-jobs'
 import { createBullMQJobData, isBullMQEnabled } from '@/lib/core/bullmq'
+import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
 import {
   createTimeoutAbortController,
   getTimeoutErrorMessage,
@@ -238,7 +239,7 @@ async function handleAsyncExecution(params: AsyncExecutionParams): Promise<NextR
       jobId,
     })
 
-    if (!isBullMQEnabled() && jobQueue) {
+    if (!isBullMQEnabled() && !isTriggerDevEnabled && jobQueue) {
       const inlineJobQueue = jobQueue
       void (async () => {
         try {

apps/sim/lib/webhooks/processor.ts

@@ -8,7 +8,7 @@ import { isOrganizationOnTeamOrEnterprisePlan } from '@/lib/billing/core/subscri
 import { getInlineJobQueue, getJobQueue } from '@/lib/core/async-jobs'
 import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types'
 import { createBullMQJobData, isBullMQEnabled } from '@/lib/core/bullmq'
-import { isProd } from '@/lib/core/config/feature-flags'
+import { isProd, isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
 import { safeCompare } from '@/lib/core/security/encryption'
 import { enqueueWorkspaceDispatch } from '@/lib/core/workspace-dispatch'
 import { getEffectiveDecryptedEnv } from '@/lib/environment/utils'
@@ -1265,9 +1265,16 @@ export async function queueWebhookExecution(
 
     const isPolling = isPollingWebhookProvider(payload.provider)
 
-    if (isPolling && isBullMQEnabled()) {
-      const jobId = isBullMQEnabled()
-        ? await enqueueWorkspaceDispatch({
+    if (isPolling && (isTriggerDevEnabled || isBullMQEnabled())) {
+      const jobId = isTriggerDevEnabled
+        ? await (await getJobQueue()).enqueue('webhook-execution', payload, {
+            metadata: {
+              workflowId: foundWorkflow.id,
+              userId: actorUserId,
+              correlation,
+            },
+          })
+        : await enqueueWorkspaceDispatch({
             id: executionId,
             workspaceId: foundWorkflow.workspaceId,
             lane: 'runtime',
@@ -1284,13 +1291,6 @@ export async function queueWebhookExecution(
               correlation,
             },
           })
-        : await (await getJobQueue()).enqueue('webhook-execution', payload, {
-            metadata: {
-              workflowId: foundWorkflow.id,
-              userId: actorUserId,
-              correlation,
-            },
-          })
       logger.info(
         `[${options.requestId}] Queued polling webhook execution task ${jobId} for ${foundWebhook.provider} webhook via job queue`
       )