feat: live execution engine — BullMQ, WebSocket, node status streaming

Author: DIYA73

apps/api/src/executions/execution.processor.ts

@@ -31,6 +31,9 @@ export class ExecutionProcessor {
     await this.executionsService.setStatus(executionId, ExecutionStatus.RUNNING);
     await this.executionsService.appendLog(executionId, 'system', '▶ Starting flow execution');
 
+    // Give WebSocket client time to connect and join the room
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
     try {
       const sorted = this.topologicalSort(nodes, edges);
       const context: Record<string, unknown> = { input: input || {} };

apps/api/src/flows/flows.module.ts

@@ -3,9 +3,10 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 import { FlowsController } from './flows.controller';
 import { FlowsService } from './flows.service';
 import { Flow } from './entities/flow.entity';
+import { ExecutionsModule } from '../executions/executions.module';
 
 @Module({
-  imports: [TypeOrmModule.forFeature([Flow])],
+  imports: [TypeOrmModule.forFeature([Flow]), ExecutionsModule],
   controllers: [FlowsController],
   providers: [FlowsService],
   exports: [FlowsService],

apps/api/src/flows/flows.service.ts

@@ -1,15 +1,17 @@
 import { Injectable, NotFoundException } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 import { Repository } from 'typeorm';
-import { Flow, FlowStatus, FlowGraph } from './entities/flow.entity';
+import { Flow, FlowStatus } from './entities/flow.entity';
 import { CreateFlowDto } from './dto/create-flow.dto';
 import { UpdateFlowDto } from './dto/update-flow.dto';
+import { ExecutionsService } from '../executions/executions.service';
 
 @Injectable()
 export class FlowsService {
   constructor(
     @InjectRepository(Flow)
     private readonly flowRepo: Repository<Flow>,
+    private readonly executionsService: ExecutionsService,
   ) {}
 
   async create(workspaceId: string, dto: CreateFlowDto): Promise<Flow> {
@@ -55,9 +57,9 @@ export class FlowsService {
   }
 
   async execute(id: string, workspaceId: string): Promise<{ id: string; status: string }> {
-    await this.findOne(id, workspaceId);
-    // Execution engine will be wired in next phase
-    return { id: `exec_${Date.now()}`, status: 'queued' };
+    const flow = await this.findOne(id, workspaceId);
+    const execution = await this.executionsService.enqueue(flow);
+    return { id: execution.id, status: execution.status };
   }
 
   async remove(id: string, workspaceId: string): Promise<void> {

apps/api/src/nodes/nodes.service.ts

@@ -20,29 +20,31 @@ export class NodesService {
     input: Record<string, unknown>,
     executionId: string,
   ): Promise<unknown> {
-    switch (node.type) {
-      case 'ai-llm':       return this.runAiLlm(node, input);
-      case 'web-scraper':  return this.runWebScraper(node, input);
-      case 'api-caller':   return this.runApiCaller(node, input);
-      case 'code-runner':  return this.runCodeRunner(node, input);
-      case 'email-sender': return this.runEmailSender(node, input);
+    const nodeType = ((node.data as Record<string, unknown>)?.type as string) || node.type;
+    switch (nodeType) {
+      case 'ai-llm':         return this.runAiLlm(node, input);
+      case 'web-scraper':    return this.runWebScraper(node, input);
+      case 'api-caller':     return this.runApiCaller(node, input);
+      case 'code-runner':    return this.runCodeRunner(node, input);
+      case 'email-sender':   return this.runEmailSender(node, input);
       case 'data-transform': return this.runDataTransform(node, input);
       case 'webhook-output': return this.runWebhookOutput(node, input);
-      case 'condition':    return this.runCondition(node, input);
+      case 'condition':      return this.runCondition(node, input);
       default:
-        throw new BadRequestException(`Unknown node type: ${node.type}`);
+        throw new BadRequestException(`Unknown node type: ${nodeType}`);
     }
   }
 
   // ─── AI LLM Node ──────────────────────────────────────────
   private async runAiLlm(node: FlowNode, input: Record<string, unknown>): Promise<unknown> {
-    const { prompt, model = 'gpt-4o', systemPrompt } = node.data.config as {
+    const { prompt, model = 'gpt-4o', systemPrompt } = (node.data.config || {}) as {
       prompt: string;
       model?: string;
       systemPrompt?: string;
     };
 
-    // Allow prompt to reference upstream data with {{key}} syntax
+    if (!prompt) return { text: '[no prompt configured]', usage: null };
+
     const resolvedPrompt = this.interpolate(prompt, input);
 
     const messages: OpenAI.ChatCompletionMessageParam[] = [];
@@ -81,13 +83,15 @@ export class NodesService {
 
   // ─── API Caller Node ──────────────────────────────────────
   private async runApiCaller(node: FlowNode, input: Record<string, unknown>): Promise<unknown> {
-    const { url, method = 'GET', headers = {}, body } = node.data.config as {
+    const { url, method = 'GET', headers = {}, body } = (node.data.config || {}) as {
       url: string;
       method?: string;
       headers?: Record<string, string>;
       body?: string;
     };
 
+    if (!url) return { statusCode: 0, data: '[no url configured]', headers: {} };
+
     const resolvedUrl = this.interpolate(url, input);
     const resolvedBody = body ? this.interpolate(body, input) : undefined;
 

apps/web/src/app/flows/[id]/page.tsx

@@ -33,7 +33,9 @@ function FlowEditor({ flowId }: { flowId: string }) {
   const [showTriggers, setShowTriggers] = useState(false);
 
   // 🔴 live node status updates via WebSocket
-  useExecutionSocket(executionId);
+  useExecutionSocket(executionId, () => {
+    setIsRunning(false);
+  });
 
   useEffect(() => {
     if (flowId === 'new') return;
@@ -82,16 +84,15 @@ function FlowEditor({ flowId }: { flowId: string }) {
   };
 
   const handleRun = async () => {
+    if (flowId === 'new') { alert('Save the flow first!'); return; }
     setIsRunning(true);
     setShowLogs(true);
     setShowTriggers(false);
     try {
-      if (flowId === 'new') { alert('Save the flow first!'); return; }
       const execution = await flowsApi.execute(flowId);
       setExecutionId(execution.id);
     } catch (err) {
       console.error('Run failed:', err);
-    } finally {
       setIsRunning(false);
     }
   };

apps/web/src/lib/hooks/useExecutionSocket.ts

@@ -6,7 +6,10 @@ let socket: Socket | null = null;
 
 export type NodeExecutionStatus = 'idle' | 'running' | 'success' | 'error';
 
-export function useExecutionSocket(executionId: string | null) {
+export function useExecutionSocket(
+  executionId: string | null,
+  onConnected?: () => void,
+) {
   const { setNodes } = useReactFlow();
 
   const updateNodeStatus = useCallback(
@@ -31,6 +34,7 @@ export function useExecutionSocket(executionId: string | null) {
 
     socket.on('connect', () => {
       socket?.emit('join:execution', { executionId });
+      onConnected?.();
     });
 
     socket.on('execution:node:status', (payload: {
@@ -51,5 +55,5 @@ export function useExecutionSocket(executionId: string | null) {
       socket?.disconnect();
       socket = null;
     };
-  }, [executionId, updateNodeStatus]);
+  }, [executionId, updateNodeStatus, onConnected]);
 }