improvement(deployment): solve multiple client side races, and deployed state management issues (#4502)

* improvement(deploy): state transitions

* more fixes

* improvements and shift to outbox policy with eager call

* address comments

* ux improvement

* address comments

* address bugbot

Author: icecrasher321

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

@@ -16,7 +16,6 @@ import {
   workflowsOrchestrationMock,
   workflowsOrchestrationMockFns,
   workflowsPersistenceUtilsMock,
-  workflowsPersistenceUtilsMockFns,
 } from '@sim/testing'
 import { NextRequest } from 'next/server'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
@@ -28,7 +27,7 @@ const { mockCheckChatAccess } = vi.hoisted(() => ({
 const mockCreateSuccessResponse = workflowsApiUtilsMockFns.mockCreateSuccessResponse
 const mockCreateErrorResponse = workflowsApiUtilsMockFns.mockCreateErrorResponse
 const mockEncryptSecret = encryptionMockFns.mockEncryptSecret
-const mockDeployWorkflow = workflowsPersistenceUtilsMockFns.mockDeployWorkflow
+const mockPerformFullDeploy = workflowsOrchestrationMockFns.mockPerformFullDeploy
 const mockPerformChatUndeploy = workflowsOrchestrationMockFns.mockPerformChatUndeploy
 const mockNotifySocketDeploymentChanged =
   workflowsOrchestrationMockFns.mockNotifySocketDeploymentChanged
@@ -73,7 +72,7 @@ describe('Chat Edit API Route', () => {
     })
 
     mockEncryptSecret.mockResolvedValue({ encrypted: 'encrypted-password' })
-    mockDeployWorkflow.mockResolvedValue({ success: true, version: 1 })
+    mockPerformFullDeploy.mockResolvedValue({ success: true, version: 1 })
     mockNotifySocketDeploymentChanged.mockResolvedValue(undefined)
   })
 

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')
 
@@ -126,21 +126,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
@@ -156,6 +143,27 @@ 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(),
       }

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

@@ -0,0 +1,97 @@
+/**
+ * @vitest-environment node
+ */
+import {
+  authMockFns,
+  dbChainMock,
+  dbChainMockFns,
+  resetDbChainMock,
+  workflowsApiUtilsMock,
+  workflowsApiUtilsMockFns,
+  workflowsOrchestrationMock,
+  workflowsOrchestrationMockFns,
+} from '@sim/testing'
+import { NextRequest } from 'next/server'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockCheckWorkflowAccessForFormCreation } = vi.hoisted(() => ({
+  mockCheckWorkflowAccessForFormCreation: vi.fn(),
+}))
+
+const mockCreateErrorResponse = workflowsApiUtilsMockFns.mockCreateErrorResponse
+const mockPerformFullDeploy = workflowsOrchestrationMockFns.mockPerformFullDeploy
+
+vi.mock('@sim/db', () => dbChainMock)
+
+vi.mock('@sim/utils/id', () => ({
+  generateId: vi.fn(() => 'form-123'),
+}))
+
+vi.mock('@/app/api/form/utils', () => ({
+  checkWorkflowAccessForFormCreation: mockCheckWorkflowAccessForFormCreation,
+  DEFAULT_FORM_CUSTOMIZATIONS: {},
+}))
+
+vi.mock('@/app/api/workflows/utils', () => workflowsApiUtilsMock)
+
+vi.mock('@/lib/core/config/feature-flags', () => ({
+  isDev: true,
+}))
+
+vi.mock('@/lib/core/utils/urls', () => ({
+  getEmailDomain: vi.fn(() => 'localhost:3000'),
+}))
+
+vi.mock('@/lib/workflows/orchestration', () => workflowsOrchestrationMock)
+
+import { POST } from '@/app/api/form/route'
+
+describe('Form API Route', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+
+    authMockFns.mockGetSession.mockResolvedValue({
+      user: {
+        id: 'user-123',
+        email: 'user@example.com',
+        name: 'Test User',
+      },
+    })
+    mockCreateErrorResponse.mockImplementation((message, status = 500) => {
+      return new Response(JSON.stringify({ error: message }), {
+        status,
+        headers: { 'Content-Type': 'application/json' },
+      })
+    })
+    mockCheckWorkflowAccessForFormCreation.mockResolvedValue({
+      hasAccess: true,
+      workflow: {
+        id: 'workflow-123',
+        isDeployed: false,
+        workspaceId: 'workspace-123',
+      },
+    })
+    dbChainMockFns.limit.mockResolvedValue([])
+  })
+
+  it('cleans up inserted form when deploy throws', async () => {
+    mockPerformFullDeploy.mockRejectedValue(new Error('Deploy exploded'))
+
+    const request = new NextRequest('http://localhost:3000/api/form', {
+      method: 'POST',
+      body: JSON.stringify({
+        workflowId: 'workflow-123',
+        identifier: 'test-form',
+        title: 'Test Form',
+      }),
+    })
+
+    const response = await POST(request)
+
+    expect(response.status).toBe(500)
+    expect(dbChainMockFns.insert).toHaveBeenCalled()
+    expect(dbChainMockFns.delete).toHaveBeenCalled()
+    expect(mockCreateErrorResponse).toHaveBeenCalledWith('Deploy exploded', 500)
+  })
+})

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

@@ -12,20 +12,28 @@ 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
 }
 
+async function cleanupFormAfterDeployFailure(formId: string) {
+  try {
+    await db.delete(form).where(eq(form.id, formId))
+  } catch (cleanupError) {
+    logger.error('Failed to clean up form after deploy failure:', cleanupError)
+  }
+}
+
 export const GET = withRouteHandler(async (request: NextRequest) => {
   try {
     const session = await getSession()
@@ -106,21 +114,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 +154,29 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
       updatedAt: new Date(),
     })
 
+    let result: Awaited<ReturnType<typeof performFullDeploy>>
+    try {
+      result = await performFullDeploy({
+        workflowId,
+        userId: session.user.id,
+        request,
+      })
+    } catch (error) {
+      await cleanupFormAfterDeployFailure(id)
+      throw error
+    }
+
+    if (!result.success) {
+      await cleanupFormAfterDeployFailure(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

@@ -20,6 +20,7 @@ const {
   mockExecuteJobInline,
   mockFeatureFlags,
   mockEnqueue,
+  mockGetJob,
   mockStartJob,
   mockCompleteJob,
   mockMarkJobFailed,
@@ -34,6 +35,7 @@ const {
     isDev: true,
   },
   mockEnqueue: vi.fn().mockResolvedValue('job-id-1'),
+  mockGetJob: vi.fn().mockResolvedValue(null),
   mockStartJob: vi.fn().mockResolvedValue(undefined),
   mockCompleteJob: vi.fn().mockResolvedValue(undefined),
   mockMarkJobFailed: vi.fn().mockResolvedValue(undefined),
@@ -54,6 +56,7 @@ vi.mock('@/lib/core/config/feature-flags', () => mockFeatureFlags)
 vi.mock('@/lib/core/async-jobs', () => ({
   getJobQueue: vi.fn().mockResolvedValue({
     enqueue: mockEnqueue,
+    getJob: mockGetJob,
     startJob: mockStartJob,
     completeJob: mockCompleteJob,
     markJobFailed: mockMarkJobFailed,
@@ -69,6 +72,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 +170,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 +186,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 +200,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 +223,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 +236,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 +253,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())
@@ -255,6 +268,8 @@ describe('Scheduled Workflow Execution API Route', () => {
         requestId: 'test-request-id',
       }),
       expect.objectContaining({
+        jobId: expect.stringMatching(/^schedule_[0-9a-f]{32}$/),
+        concurrencyKey: expect.stringMatching(/^schedule_[0-9a-f]{32}$/),
         metadata: expect.objectContaining({
           workflowId: 'workflow-1',
           workspaceId: 'workspace-1',