improvements and shift to outbox policy with eager call

Author: icecrasher321

apps/sim/app/api/chat/manage/[id]/route.ts

@@ -11,12 +11,12 @@ import { isDev } from '@/lib/core/config/feature-flags'
 import { encryptSecret } from '@/lib/core/security/encryption'
 import { getEmailDomain } from '@/lib/core/utils/urls'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { notifySocketDeploymentChanged, performChatUndeploy } from '@/lib/workflows/orchestration'
-import { deployWorkflow } from '@/lib/workflows/persistence/utils'
+import { performChatUndeploy, performFullDeploy } from '@/lib/workflows/orchestration'
 import { checkChatAccess } from '@/app/api/chat/utils'
 import { createErrorResponse, createSuccessResponse } from '@/app/api/workflows/utils'
 
 export const dynamic = 'force-dynamic'
+export const maxDuration = 120
 
 const logger = createLogger('ChatDetailAPI')
 
@@ -122,21 +122,8 @@ export const PATCH = withRouteHandler(
         }
       }
 
-      // Redeploy the workflow to ensure latest version is active
-      const deployResult = await deployWorkflow({
-        workflowId: existingChat[0].workflowId,
-        deployedBy: session.user.id,
-      })
-
-      if (!deployResult.success) {
-        logger.warn(
-          `Failed to redeploy workflow for chat update: ${deployResult.error}, continuing with chat update`
-        )
-      } else {
-        logger.info(
-          `Redeployed workflow ${existingChat[0].workflowId} for chat update (v${deployResult.version})`
-        )
-        await notifySocketDeploymentChanged(existingChat[0].workflowId)
+      if (workflowId && workflowId !== existingChat[0].workflowId) {
+        return createErrorResponse('Changing a chat deployment workflow is not supported', 400)
       }
 
       let encryptedPassword
@@ -152,11 +139,31 @@ export const PATCH = withRouteHandler(
         logger.info('Keeping existing password')
       }
 
+      // Redeploy the workflow to ensure latest version is active
+      const deployResult = await performFullDeploy({
+        workflowId: existingChat[0].workflowId,
+        userId: session.user.id,
+        request,
+      })
+
+      if (!deployResult.success) {
+        logger.warn(`Failed to redeploy workflow for chat update: ${deployResult.error}`)
+        const status =
+          deployResult.errorCode === 'validation'
+            ? 400
+            : deployResult.errorCode === 'not_found'
+              ? 404
+              : 500
+        return createErrorResponse(deployResult.error || 'Failed to redeploy workflow', status)
+      }
+      logger.info(
+        `Redeployed workflow ${existingChat[0].workflowId} for chat update (v${deployResult.version})`
+      )
+
       const updateData: Record<string, unknown> = {
         updatedAt: new Date(),
       }
 
-      if (workflowId) updateData.workflowId = workflowId
       if (identifier) updateData.identifier = identifier
       if (title) updateData.title = title
       if (description !== undefined) updateData.description = description

apps/sim/app/api/form/route.ts

@@ -12,15 +12,15 @@ import { isDev } from '@/lib/core/config/feature-flags'
 import { encryptSecret } from '@/lib/core/security/encryption'
 import { getEmailDomain } from '@/lib/core/utils/urls'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { notifySocketDeploymentChanged } from '@/lib/workflows/orchestration'
-import { deployWorkflow } from '@/lib/workflows/persistence/utils'
+import { performFullDeploy } from '@/lib/workflows/orchestration'
 import {
   checkWorkflowAccessForFormCreation,
   DEFAULT_FORM_CUSTOMIZATIONS,
 } from '@/app/api/form/utils'
 import { createErrorResponse, createSuccessResponse } from '@/app/api/workflows/utils'
 
 const logger = createLogger('FormAPI')
+export const maxDuration = 120
 
 function getErrorMessage(error: unknown, fallback: string): string {
   return error instanceof Error ? error.message : fallback
@@ -106,21 +106,6 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
       return createErrorResponse('Workflow not found or access denied', 404)
     }
 
-    const result = await deployWorkflow({
-      workflowId,
-      deployedBy: session.user.id,
-    })
-
-    if (!result.success) {
-      return createErrorResponse(result.error || 'Failed to deploy workflow', 500)
-    }
-
-    logger.info(
-      `${workflowRecord.isDeployed ? 'Redeployed' : 'Auto-deployed'} workflow ${workflowId} for form (v${result.version})`
-    )
-
-    await notifySocketDeploymentChanged(workflowId)
-
     let encryptedPassword = null
     if (authType === 'password' && password) {
       const { encrypted } = await encryptSecret(password)
@@ -161,6 +146,23 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
       updatedAt: new Date(),
     })
 
+    const result = await performFullDeploy({
+      workflowId,
+      userId: session.user.id,
+      request,
+    })
+
+    if (!result.success) {
+      await db.delete(form).where(eq(form.id, id))
+      const status =
+        result.errorCode === 'validation' ? 400 : result.errorCode === 'not_found' ? 404 : 500
+      return createErrorResponse(result.error || 'Failed to deploy workflow', status)
+    }
+
+    logger.info(
+      `${workflowRecord.isDeployed ? 'Redeployed' : 'Auto-deployed'} workflow ${workflowId} for form (v${result.version})`
+    )
+
     const baseDomain = getEmailDomain()
     const protocol = isDev ? 'http' : 'https'
     const formUrl = `${protocol}://${baseDomain}/form/${identifier}`

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

@@ -69,6 +69,7 @@ vi.mock('drizzle-orm', () => ({
   ne: vi.fn((field: unknown, value: unknown) => ({ field, value, type: 'ne' })),
   lte: vi.fn((field: unknown, value: unknown) => ({ field, value, type: 'lte' })),
   lt: vi.fn((field: unknown, value: unknown) => ({ field, value, type: 'lt' })),
+  inArray: vi.fn((field: unknown, values: unknown[]) => ({ field, values, type: 'inArray' })),
   not: vi.fn((condition: unknown) => ({ type: 'not', condition })),
   isNull: vi.fn((field: unknown) => ({ type: 'isNull', field })),
   or: vi.fn((...conditions: unknown[]) => ({ type: 'or', conditions })),
@@ -166,6 +167,8 @@ function createMockRequest(): NextRequest {
 describe('Scheduled Workflow Execution API Route', () => {
   beforeEach(() => {
     vi.clearAllMocks()
+    dbChainMockFns.limit.mockReset()
+    dbChainMockFns.returning.mockReset()
     resetDbChainMock()
     requestUtilsMockFns.mockGenerateRequestId.mockReturnValue('test-request-id')
     workflowsUtilsMockFns.mockGetWorkflowById.mockResolvedValue({
@@ -180,6 +183,7 @@ describe('Scheduled Workflow Execution API Route', () => {
   })
 
   it('should execute scheduled workflows with Trigger.dev disabled', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([{ id: 'schedule-1' }]).mockResolvedValueOnce([])
     dbChainMockFns.returning.mockReturnValueOnce(SINGLE_SCHEDULE).mockReturnValueOnce([])
 
     const response = await GET(createMockRequest())
@@ -193,6 +197,7 @@ describe('Scheduled Workflow Execution API Route', () => {
 
   it('should queue schedules to Trigger.dev when enabled', async () => {
     mockFeatureFlags.isTriggerDevEnabled = true
+    dbChainMockFns.limit.mockResolvedValueOnce([{ id: 'schedule-1' }]).mockResolvedValueOnce([])
     dbChainMockFns.returning.mockReturnValueOnce(SINGLE_SCHEDULE).mockReturnValueOnce([])
 
     const response = await GET(createMockRequest())
@@ -215,6 +220,9 @@ describe('Scheduled Workflow Execution API Route', () => {
   })
 
   it('should execute multiple schedules in parallel', async () => {
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([{ id: 'schedule-1' }, { id: 'schedule-2' }])
+      .mockResolvedValueOnce([])
     dbChainMockFns.returning.mockReturnValueOnce(MULTIPLE_SCHEDULES).mockReturnValueOnce([])
 
     const response = await GET(createMockRequest())
@@ -225,7 +233,8 @@ describe('Scheduled Workflow Execution API Route', () => {
   })
 
   it('should execute mothership jobs inline', async () => {
-    dbChainMockFns.returning.mockReturnValueOnce([]).mockReturnValueOnce(SINGLE_JOB)
+    dbChainMockFns.limit.mockResolvedValueOnce([]).mockResolvedValueOnce([{ id: 'job-1' }])
+    dbChainMockFns.returning.mockReturnValueOnce(SINGLE_JOB)
 
     const response = await GET(createMockRequest())
 
@@ -241,6 +250,7 @@ describe('Scheduled Workflow Execution API Route', () => {
   })
 
   it('should enqueue schedule with correlation metadata via job queue', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([{ id: 'schedule-1' }]).mockResolvedValueOnce([])
     dbChainMockFns.returning.mockReturnValueOnce(SINGLE_SCHEDULE).mockReturnValueOnce([])
 
     const response = await GET(createMockRequest())

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

@@ -2,7 +2,7 @@ import { db, workflowDeploymentVersion, workflowSchedule } from '@sim/db'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
-import { and, eq, isNull, lt, lte, ne, not, or, sql } from 'drizzle-orm'
+import { and, eq, inArray, isNull, lt, lte, ne, not, or, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
 import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
@@ -17,6 +17,9 @@ import {
 export const dynamic = 'force-dynamic'
 
 const logger = createLogger('ScheduledExecuteAPI')
+const MAX_CRON_CLAIMS = 20
+const RESERVED_WORKFLOW_CLAIMS = 10
+const RESERVED_JOB_CLAIMS = MAX_CRON_CLAIMS - RESERVED_WORKFLOW_CLAIMS
 
 const dueFilter = (queuedAt: Date) =>
   and(
@@ -30,27 +33,42 @@ const dueFilter = (queuedAt: Date) =>
     )
   )
 
-export const GET = withRouteHandler(async (request: NextRequest) => {
-  const requestId = generateRequestId()
-  logger.info(`[${requestId}] Scheduled execution triggered at ${new Date().toISOString()}`)
+const activeWorkflowDeploymentFilter = () =>
+  sql`${workflowSchedule.deploymentVersionId} = (select ${workflowDeploymentVersion.id} from ${workflowDeploymentVersion} where ${workflowDeploymentVersion.workflowId} = ${workflowSchedule.workflowId} and ${workflowDeploymentVersion.isActive} = true)`
 
-  const authError = verifyCronAuth(request, 'Schedule execution')
-  if (authError) {
-    return authError
-  }
+const workflowScheduleFilter = (queuedAt: Date) =>
+  and(
+    dueFilter(queuedAt),
+    or(eq(workflowSchedule.sourceType, 'workflow'), isNull(workflowSchedule.sourceType)),
+    activeWorkflowDeploymentFilter()
+  )
 
-  const queuedAt = new Date()
+const jobScheduleFilter = (queuedAt: Date) =>
+  and(dueFilter(queuedAt), eq(workflowSchedule.sourceType, 'job'))
 
-  try {
-    // Workflow schedules (require active deployment)
-    const dueSchedules = await db
+async function claimWorkflowSchedules(queuedAt: Date, limit: number) {
+  if (limit <= 0) return []
+
+  return db.transaction(async (tx) => {
+    const rows = await tx
+      .select({ id: workflowSchedule.id })
+      .from(workflowSchedule)
+      .where(workflowScheduleFilter(queuedAt))
+      .for('update', { skipLocked: true })
+      .limit(limit)
+
+    if (rows.length === 0) return []
+
+    return tx
       .update(workflowSchedule)
       .set({ lastQueuedAt: queuedAt, updatedAt: queuedAt })
       .where(
         and(
-          dueFilter(queuedAt),
-          or(eq(workflowSchedule.sourceType, 'workflow'), isNull(workflowSchedule.sourceType)),
-          sql`${workflowSchedule.deploymentVersionId} = (select ${workflowDeploymentVersion.id} from ${workflowDeploymentVersion} where ${workflowDeploymentVersion.workflowId} = ${workflowSchedule.workflowId} and ${workflowDeploymentVersion.isActive} = true)`
+          workflowScheduleFilter(queuedAt),
+          inArray(
+            workflowSchedule.id,
+            rows.map((row) => row.id)
+          )
         )
       )
       .returning({
@@ -62,21 +80,68 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
         failedCount: workflowSchedule.failedCount,
         nextRunAt: workflowSchedule.nextRunAt,
         lastQueuedAt: workflowSchedule.lastQueuedAt,
+        deploymentVersionId: workflowSchedule.deploymentVersionId,
         sourceType: workflowSchedule.sourceType,
       })
+  })
+}
 
-    // Jobs (no deployment, dispatch inline)
-    const dueJobs = await db
+async function claimJobSchedules(queuedAt: Date, limit: number) {
+  if (limit <= 0) return []
+
+  return db.transaction(async (tx) => {
+    const rows = await tx
+      .select({ id: workflowSchedule.id })
+      .from(workflowSchedule)
+      .where(jobScheduleFilter(queuedAt))
+      .for('update', { skipLocked: true })
+      .limit(limit)
+
+    if (rows.length === 0) return []
+
+    return tx
       .update(workflowSchedule)
       .set({ lastQueuedAt: queuedAt, updatedAt: queuedAt })
-      .where(and(dueFilter(queuedAt), eq(workflowSchedule.sourceType, 'job')))
+      .where(
+        and(
+          jobScheduleFilter(queuedAt),
+          inArray(
+            workflowSchedule.id,
+            rows.map((row) => row.id)
+          )
+        )
+      )
       .returning({
         id: workflowSchedule.id,
         cronExpression: workflowSchedule.cronExpression,
         failedCount: workflowSchedule.failedCount,
         lastQueuedAt: workflowSchedule.lastQueuedAt,
         sourceType: workflowSchedule.sourceType,
       })
+  })
+}
+
+export const GET = withRouteHandler(async (request: NextRequest) => {
+  const requestId = generateRequestId()
+  logger.info(`[${requestId}] Scheduled execution triggered at ${new Date().toISOString()}`)
+
+  const authError = verifyCronAuth(request, 'Schedule execution')
+  if (authError) {
+    return authError
+  }
+
+  const queuedAt = new Date()
+
+  try {
+    const dueSchedules = await claimWorkflowSchedules(queuedAt, RESERVED_WORKFLOW_CLAIMS)
+    const dueJobs = await claimJobSchedules(queuedAt, RESERVED_JOB_CLAIMS)
+    const remainingClaimBudget = Math.max(0, MAX_CRON_CLAIMS - dueSchedules.length - dueJobs.length)
+
+    if (remainingClaimBudget > 0 && dueSchedules.length === RESERVED_WORKFLOW_CLAIMS) {
+      dueSchedules.push(...(await claimWorkflowSchedules(queuedAt, remainingClaimBudget)))
+    } else if (remainingClaimBudget > 0 && dueJobs.length === RESERVED_JOB_CLAIMS) {
+      dueJobs.push(...(await claimJobSchedules(queuedAt, remainingClaimBudget)))
+    }
 
     const totalCount = dueSchedules.length + dueJobs.length
     logger.info(
@@ -108,6 +173,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
         requestId,
         correlation,
         blockId: schedule.blockId || undefined,
+        deploymentVersionId: schedule.deploymentVersionId || undefined,
         cronExpression: schedule.cronExpression || undefined,
         lastRanAt: schedule.lastRanAt?.toISOString(),
         failedCount: schedule.failedCount || 0,

apps/sim/app/api/v1/admin/types.ts

@@ -645,6 +645,7 @@ export interface AdminDeployResult {
 
 export interface AdminUndeployResult {
   isDeployed: boolean
+  warnings?: string[]
 }
 
 // =============================================================================

apps/sim/app/api/v1/admin/workflows/[id]/deploy/route.ts

@@ -18,6 +18,7 @@ import {
 import type { AdminDeployResult, AdminUndeployResult } from '@/app/api/v1/admin/types'
 
 const logger = createLogger('AdminWorkflowDeployAPI')
+export const maxDuration = 120
 
 interface RouteParams {
   id: string
@@ -109,6 +110,7 @@ export const DELETE = withRouteHandler(
 
       const response: AdminUndeployResult = {
         isDeployed: false,
+        warnings: result.warnings,
       }
 
       return singleResponse(response)

apps/sim/app/api/v1/admin/workflows/[id]/versions/[versionId]/activate/route.ts

@@ -14,6 +14,7 @@ import {
 } from '@/app/api/v1/admin/responses'
 
 const logger = createLogger('AdminWorkflowActivateVersionAPI')
+export const maxDuration = 120
 
 interface RouteParams {
   id: string