remove worker code

Author: icecrasher321

README.md

@@ -74,10 +74,6 @@ docker compose -f docker-compose.prod.yml up -d
 
 Open [http://localhost:3000](http://localhost:3000)
 
-#### Background worker note
-
-The Docker Compose stack starts a dedicated worker container by default. If `REDIS_URL` is not configured, the worker will start, log that it is idle, and do no queue processing. This is expected. Queue-backed API, webhook, and schedule execution requires Redis; installs without Redis continue to use the inline execution path.
-
 Sim also supports local models via [Ollama](https://ollama.ai) and [vLLM](https://docs.vllm.ai/) — see the [Docker self-hosting docs](https://docs.sim.ai/self-hosting/docker) for setup details.
 
 ### Self-hosted: Manual Setup
@@ -123,12 +119,10 @@ cd packages/db && bun run db:migrate
 5. Start development servers:
 
 ```bash
-bun run dev:full  # Starts Next.js app, realtime socket server, and the BullMQ worker
+bun run dev:full  # Starts Next.js app and realtime socket server
 ```
 
-If `REDIS_URL` is not configured, the worker will remain idle and execution continues inline.
-
-Or run separately: `bun run dev` (Next.js), `cd apps/sim && bun run dev:sockets` (realtime), and `cd apps/sim && bun run worker` (BullMQ worker).
+Or run separately: `bun run dev` (Next.js) and `cd apps/sim && bun run dev:sockets` (realtime).
 
 ## Copilot API Keys
 

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

@@ -6,16 +6,16 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 const {
   mockCheckHybridAuth,
-  mockGetDispatchJobRecord,
   mockGetJobQueue,
   mockVerifyWorkflowAccess,
   mockGetWorkflowById,
+  mockGetJob,
 } = vi.hoisted(() => ({
   mockCheckHybridAuth: vi.fn(),
-  mockGetDispatchJobRecord: vi.fn(),
   mockGetJobQueue: vi.fn(),
   mockVerifyWorkflowAccess: vi.fn(),
   mockGetWorkflowById: vi.fn(),
+  mockGetJob: vi.fn(),
 }))
 
 vi.mock('@sim/logger', () => ({
@@ -32,19 +32,9 @@ vi.mock('@/lib/auth/hybrid', () => ({
 }))
 
 vi.mock('@/lib/core/async-jobs', () => ({
-  JOB_STATUS: {
-    PENDING: 'pending',
-    PROCESSING: 'processing',
-    COMPLETED: 'completed',
-    FAILED: 'failed',
-  },
   getJobQueue: mockGetJobQueue,
 }))
 
-vi.mock('@/lib/core/workspace-dispatch/store', () => ({
-  getDispatchJobRecord: mockGetDispatchJobRecord,
-}))
-
 vi.mock('@/lib/core/utils/request', () => ({
   generateRequestId: vi.fn().mockReturnValue('request-1'),
 }))
@@ -85,71 +75,51 @@ describe('GET /api/jobs/[jobId]', () => {
     })
 
     mockGetJobQueue.mockResolvedValue({
-      getJob: vi.fn().mockResolvedValue(null),
+      getJob: mockGetJob,
     })
   })
 
-  it('returns dispatcher-aware waiting status with metadata', async () => {
-    mockGetDispatchJobRecord.mockResolvedValue({
-      id: 'dispatch-1',
-      workspaceId: 'workspace-1',
-      lane: 'runtime',
-      queueName: 'workflow-execution',
-      bullmqJobName: 'workflow-execution',
-      bullmqPayload: {},
+  it('returns job status with metadata', async () => {
+    mockGetJob.mockResolvedValue({
+      id: 'job-1',
+      status: 'pending',
       metadata: {
         workflowId: 'workflow-1',
       },
-      priority: 10,
-      status: 'waiting',
-      createdAt: 1000,
-      admittedAt: 2000,
     })
 
     const response = await GET(createMockRequest(), {
-      params: Promise.resolve({ jobId: 'dispatch-1' }),
+      params: Promise.resolve({ jobId: 'job-1' }),
     })
     const body = await response.json()
 
     expect(response.status).toBe(200)
-    expect(body.status).toBe('waiting')
-    expect(body.metadata.queueName).toBe('workflow-execution')
-    expect(body.metadata.lane).toBe('runtime')
-    expect(body.metadata.workspaceId).toBe('workspace-1')
+    expect(body.status).toBe('pending')
+    expect(body.metadata.workflowId).toBe('workflow-1')
   })
 
-  it('returns completed output from dispatch state', async () => {
-    mockGetDispatchJobRecord.mockResolvedValue({
-      id: 'dispatch-2',
-      workspaceId: 'workspace-1',
-      lane: 'interactive',
-      queueName: 'workflow-execution',
-      bullmqJobName: 'direct-workflow-execution',
-      bullmqPayload: {},
+  it('returns completed output from job', async () => {
+    mockGetJob.mockResolvedValue({
+      id: 'job-2',
+      status: 'completed',
       metadata: {
         workflowId: 'workflow-1',
       },
-      priority: 1,
-      status: 'completed',
-      createdAt: 1000,
-      startedAt: 2000,
-      completedAt: 7000,
       output: { success: true },
     })
 
     const response = await GET(createMockRequest(), {
-      params: Promise.resolve({ jobId: 'dispatch-2' }),
+      params: Promise.resolve({ jobId: 'job-2' }),
     })
     const body = await response.json()
 
     expect(response.status).toBe(200)
     expect(body.status).toBe('completed')
     expect(body.output).toEqual({ success: true })
-    expect(body.metadata.duration).toBe(5000)
   })
 
-  it('returns 404 when neither dispatch nor BullMQ job exists', async () => {
-    mockGetDispatchJobRecord.mockResolvedValue(null)
+  it('returns 404 when job does not exist', async () => {
+    mockGetJob.mockResolvedValue(null)
 
     const response = await GET(createMockRequest(), {
       params: Promise.resolve({ jobId: 'missing-job' }),

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

@@ -3,8 +3,6 @@ import { type NextRequest, NextResponse } from 'next/server'
 import { checkHybridAuth } from '@/lib/auth/hybrid'
 import { getJobQueue } from '@/lib/core/async-jobs'
 import { generateRequestId } from '@/lib/core/utils/request'
-import { presentDispatchOrJobStatus } from '@/lib/core/workspace-dispatch/status'
-import { getDispatchJobRecord } from '@/lib/core/workspace-dispatch/store'
 import { createErrorResponse } from '@/app/api/workflows/utils'
 
 const logger = createLogger('TaskStatusAPI')
@@ -25,15 +23,14 @@ export async function GET(
 
     const authenticatedUserId = authResult.userId
 
-    const dispatchJob = await getDispatchJobRecord(taskId)
     const jobQueue = await getJobQueue()
-    const job = dispatchJob ? null : await jobQueue.getJob(taskId)
+    const job = await jobQueue.getJob(taskId)
 
-    if (!job && !dispatchJob) {
+    if (!job) {
       return createErrorResponse('Task not found', 404)
     }
 
-    const metadataToCheck = dispatchJob?.metadata ?? job?.metadata
+    const metadataToCheck = job.metadata
 
     if (metadataToCheck?.workflowId) {
       const { verifyWorkflowAccess } = await import('@/socket/middleware/permissions')
@@ -61,25 +58,22 @@ export async function GET(
       return createErrorResponse('Access denied', 403)
     }
 
-    const presented = presentDispatchOrJobStatus(dispatchJob, job)
-    const response: any = {
+    const response: Record<string, unknown> = {
       success: true,
       taskId,
-      status: presented.status,
-      metadata: presented.metadata,
+      status: job.status,
+      metadata: job.metadata,
     }
 
-    if (presented.output !== undefined) response.output = presented.output
-    if (presented.error !== undefined) response.error = presented.error
-    if (presented.estimatedDuration !== undefined) {
-      response.estimatedDuration = presented.estimatedDuration
-    }
+    if (job.output !== undefined) response.output = job.output
+    if (job.error !== undefined) response.error = job.error
 
     return NextResponse.json(response)
-  } catch (error: any) {
+  } catch (error: unknown) {
+    const errorMessage = error instanceof Error ? error.message : String(error)
     logger.error(`[${requestId}] Error fetching task status:`, error)
 
-    if (error.message?.includes('not found') || error.status === 404) {
+    if (errorMessage?.includes('not found')) {
       return createErrorResponse('Task not found', 404)
     }
 

apps/sim/app/api/resume/[workflowId]/[executionId]/[contextId]/route.ts

@@ -1,13 +1,11 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
 import { AuthType } from '@/lib/auth/hybrid'
-import { getJobQueue, shouldUseBullMQ } from '@/lib/core/async-jobs'
-import { createBullMQJobData } from '@/lib/core/bullmq'
+import { getJobQueue } from '@/lib/core/async-jobs'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { SSE_HEADERS } from '@/lib/core/utils/sse'
 import { getBaseUrl } from '@/lib/core/utils/urls'
 import { generateId } from '@/lib/core/utils/uuid'
-import { enqueueWorkspaceDispatch } from '@/lib/core/workspace-dispatch'
 import { setExecutionMeta } from '@/lib/execution/event-buffer'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
@@ -227,26 +225,10 @@ export async function POST(
 
       let jobId: string
       try {
-        const useBullMQ = shouldUseBullMQ()
-        if (useBullMQ) {
-          jobId = await enqueueWorkspaceDispatch({
-            id: enqueueResult.resumeExecutionId,
-            workspaceId: workflow.workspaceId,
-            lane: 'runtime',
-            queueName: 'resume-execution',
-            bullmqJobName: 'resume-execution',
-            bullmqPayload: createBullMQJobData(resumePayload, {
-              workflowId,
-              userId,
-            }),
-            metadata: { workflowId, userId },
-          })
-        } else {
-          const jobQueue = await getJobQueue()
-          jobId = await jobQueue.enqueue('resume-execution', resumePayload, {
-            metadata: { workflowId, workspaceId: workflow.workspaceId, userId },
-          })
-        }
+        const jobQueue = await getJobQueue()
+        jobId = await jobQueue.enqueue('resume-execution', resumePayload, {
+          metadata: { workflowId, workspaceId: workflow.workspaceId, userId },
+        })
         logger.info('Enqueued async resume execution', {
           jobId,
           resumeExecutionId: enqueueResult.resumeExecutionId,

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

@@ -14,7 +14,6 @@ const {
   mockDbReturning,
   mockDbUpdate,
   mockEnqueue,
-  mockEnqueueWorkspaceDispatch,
   mockStartJob,
   mockCompleteJob,
   mockMarkJobFailed,
@@ -24,7 +23,6 @@ const {
   const mockDbSet = vi.fn().mockReturnValue({ where: mockDbWhere })
   const mockDbUpdate = vi.fn().mockReturnValue({ set: mockDbSet })
   const mockEnqueue = vi.fn().mockResolvedValue('job-id-1')
-  const mockEnqueueWorkspaceDispatch = vi.fn().mockResolvedValue('job-id-1')
   const mockStartJob = vi.fn().mockResolvedValue(undefined)
   const mockCompleteJob = vi.fn().mockResolvedValue(undefined)
   const mockMarkJobFailed = vi.fn().mockResolvedValue(undefined)
@@ -42,7 +40,6 @@ const {
     mockDbReturning,
     mockDbUpdate,
     mockEnqueue,
-    mockEnqueueWorkspaceDispatch,
     mockStartJob,
     mockCompleteJob,
     mockMarkJobFailed,
@@ -75,15 +72,6 @@ vi.mock('@/lib/core/async-jobs', () => ({
   shouldExecuteInline: vi.fn().mockReturnValue(false),
 }))
 
-vi.mock('@/lib/core/bullmq', () => ({
-  isBullMQEnabled: vi.fn().mockReturnValue(true),
-  createBullMQJobData: vi.fn((payload: unknown) => ({ payload })),
-}))
-
-vi.mock('@/lib/core/workspace-dispatch', () => ({
-  enqueueWorkspaceDispatch: mockEnqueueWorkspaceDispatch,
-}))
-
 vi.mock('@/lib/workflows/utils', () => ({
   getWorkflowById: vi.fn().mockResolvedValue({
     id: 'workflow-1',
@@ -175,8 +163,6 @@ const SINGLE_JOB = [
     cronExpression: '0 * * * *',
     failedCount: 0,
     lastQueuedAt: undefined,
-    sourceUserId: 'user-1',
-    sourceWorkspaceId: 'workspace-1',
     sourceType: 'job',
   },
 ]
@@ -250,56 +236,48 @@ describe('Scheduled Workflow Execution API Route', () => {
     expect(data).toHaveProperty('executedCount', 2)
   })
 
-  it('should queue mothership jobs to BullMQ when available', async () => {
+  it('should execute mothership jobs inline', async () => {
     mockDbReturning.mockReturnValueOnce([]).mockReturnValueOnce(SINGLE_JOB)
 
     const response = await GET(createMockRequest())
 
     expect(response.status).toBe(200)
-    expect(mockEnqueueWorkspaceDispatch).toHaveBeenCalledWith(
+    expect(mockExecuteJobInline).toHaveBeenCalledWith(
       expect.objectContaining({
-        workspaceId: 'workspace-1',
-        lane: 'runtime',
-        queueName: 'mothership-job-execution',
-        bullmqJobName: 'mothership-job-execution',
-        bullmqPayload: {
-          payload: {
-            scheduleId: 'job-1',
-            cronExpression: '0 * * * *',
-            failedCount: 0,
-            now: expect.any(String),
-          },
-        },
+        scheduleId: 'job-1',
+        cronExpression: '0 * * * *',
+        failedCount: 0,
+        now: expect.any(String),
       })
     )
-    expect(mockExecuteJobInline).not.toHaveBeenCalled()
   })
 
-  it('should enqueue preassigned correlation metadata for schedules', async () => {
-    mockDbReturning.mockReturnValue(SINGLE_SCHEDULE)
+  it('should enqueue schedule with correlation metadata via job queue', async () => {
+    mockDbReturning.mockReturnValueOnce(SINGLE_SCHEDULE).mockReturnValueOnce([])
 
     const response = await GET(createMockRequest())
 
     expect(response.status).toBe(200)
-    expect(mockEnqueueWorkspaceDispatch).toHaveBeenCalledWith(
+    expect(mockEnqueue).toHaveBeenCalledWith(
+      'schedule-execution',
+      expect.objectContaining({
+        scheduleId: 'schedule-1',
+        workflowId: 'workflow-1',
+        executionId: 'schedule-execution-1',
+        requestId: 'test-request-id',
+      }),
       expect.objectContaining({
-        id: 'schedule-execution-1',
-        workspaceId: 'workspace-1',
-        lane: 'runtime',
-        queueName: 'schedule-execution',
-        bullmqJobName: 'schedule-execution',
-        metadata: {
+        metadata: expect.objectContaining({
           workflowId: 'workflow-1',
-          correlation: {
+          workspaceId: 'workspace-1',
+          correlation: expect.objectContaining({
             executionId: 'schedule-execution-1',
             requestId: 'test-request-id',
             source: 'schedule',
             workflowId: 'workflow-1',
             scheduleId: 'schedule-1',
-            triggerType: 'schedule',
-            scheduledFor: '2025-01-01T00:00:00.000Z',
-          },
-        },
+          }),
+        }),
       })
     )
   })