From c23934f6bdb95b812789679168fee3b9903b74dc Mon Sep 17 00:00:00 2001 From: Tejabudumuru3 Date: Wed, 14 Oct 2026 16:46:29 +0530 Subject: [PATCH] feat: implement workflow execution history footer component and real-time Kafka event streaming architecture --- .../src/routes/userRoutes/userMiddleware.ts | 7 +- .../src/routes/userRoutes/userRoutes.ts | 5 +- .../app/components/ExecutionHistoryFooter.tsx | 1 + apps/web/app/components/nodes/BaseNode.tsx | 21 ++- apps/web/app/hooks/useWorkflowEvents.ts | 12 +- apps/web/app/lib/formatters.ts | 2 + apps/web/app/workflows/[id]/page.tsx | 4 +- apps/worker/src/engine/executor.ts | 152 ++++++------------ packages/db/prisma/schema.prisma | 1 + packages/kafka/src/index.ts | 22 +-- 10 files changed, 92 insertions(+), 135 deletions(-) diff --git a/apps/http-backend/src/routes/userRoutes/userMiddleware.ts b/apps/http-backend/src/routes/userRoutes/userMiddleware.ts index b51cf42..887acf4 100644 --- a/apps/http-backend/src/routes/userRoutes/userMiddleware.ts +++ b/apps/http-backend/src/routes/userRoutes/userMiddleware.ts @@ -24,12 +24,17 @@ export async function userMiddleware( try { // console.log(req.cookies); const authHeader = req.headers.authorization; + const auth = req.query.token + let payload: JwtPayload | null = null; if (authHeader?.startsWith("Bearer ")) { const token = authHeader.split(" ")[1]; payload = jwt.verify(token!, SECRET!) as JwtPayload; - } else { + } else if (auth) { + payload = jwt.verify(auth as string, SECRET!) as JwtPayload + } + else { payload = (await getToken({ req, secret: SECRET })) as JwtPayload | null; } diff --git a/apps/http-backend/src/routes/userRoutes/userRoutes.ts b/apps/http-backend/src/routes/userRoutes/userRoutes.ts index 1f38151..29c500e 100644 --- a/apps/http-backend/src/routes/userRoutes/userRoutes.ts +++ b/apps/http-backend/src/routes/userRoutes/userRoutes.ts @@ -1107,7 +1107,10 @@ router.get("/workflow/logs/:workflowId", userMiddleware, async (req: AuthRequest const executions = await prismaClient.workflowExecution.findMany({ where: { workflowId: workflowId }, include: { - nodeExecutions: { include: { node: true } }, + nodeExecutions: { + include: { node: true }, + orderBy: { startedAt: "asc" }, + }, }, orderBy: { startAt: "desc" }, }) diff --git a/apps/web/app/components/ExecutionHistoryFooter.tsx b/apps/web/app/components/ExecutionHistoryFooter.tsx index ec2ff6a..539e5f2 100644 --- a/apps/web/app/components/ExecutionHistoryFooter.tsx +++ b/apps/web/app/components/ExecutionHistoryFooter.tsx @@ -33,6 +33,7 @@ const StatusBadge: React.FC<{ status: string }> = ({ status }) => { Pending: 'background:rgba(245,158,11,0.15);color:#fbbf24;border:1px solid rgba(245,158,11,0.3);', Start: 'background:rgba(148,163,184,0.15);color:#cbd5e1;border:1px solid rgba(148,163,184,0.3);', ReConnecting: 'background:rgba(249,115,22,0.15);color:#fb923c;border:1px solid rgba(249,115,22,0.3);', + Skipped: 'background:rgba(100,116,139,0.15);color:#94a3b8;border:1px solid rgba(100,116,139,0.3);', }; return ( void; onTest?: () => void; onAddChild?: (sourceHandleId?: string) => void; @@ -120,10 +120,11 @@ export default function BaseNode({ id, type, data }: BaseNodeProps) { currentStatus === "running" ? "ring-4 ring-blue-500/80 shadow-[0_0_20px_rgba(59,130,246,0.6)] animate-pulse" : currentStatus === "completed" || currentStatus === "success" - ? "ring-4 ring-emerald-500/80 shadow-[0_0_15px_rgba(16,185,129,0.5)]" - : currentStatus === "failed" || currentStatus === "error" - ? "ring-4 ring-rose-500/80 shadow-[0_0_15px_rgba(244,63,94,0.6)]" - : ""; + ? "ring-4 ring-emerald-500/80 shadow-[0_0_15px_rgba(16,185,129,0.5)]" + : currentStatus === "failed" || currentStatus === "error" + ? "ring-4 ring-rose-500/80 shadow-[0_0_15px_rgba(244,63,94,0.6)]" + : currentStatus === "skipped" ? "ring-2 ring-slate-400/40 opacity-50 grayscale-[40%]" + : ""; // ── Placeholder Node (Make.com Style) ── if (isPlaceholder) { @@ -310,11 +311,10 @@ export default function BaseNode({ id, type, data }: BaseNodeProps) { position={Position.Right} id={output.id} style={{ top: topPosition }} - className={`!w-5 !h-5 flex items-center justify-center !border-2 !border-white transition-all cursor-crosshair shadow-md z-20 ${ - isAlternative + className={`!w-5 !h-5 flex items-center justify-center !border-2 !border-white transition-all cursor-crosshair shadow-md z-20 ${isAlternative ? "!bg-rose-500 !text-white" : "!bg-[#18181b] !text-white" - }`} + }`} title={`${output.label} (${output.id})`} > {onAddChild && ( @@ -343,11 +343,10 @@ export default function BaseNode({ id, type, data }: BaseNodeProps) { }} > {output.label} diff --git a/apps/web/app/hooks/useWorkflowEvents.ts b/apps/web/app/hooks/useWorkflowEvents.ts index a84e29e..26ac034 100644 --- a/apps/web/app/hooks/useWorkflowEvents.ts +++ b/apps/web/app/hooks/useWorkflowEvents.ts @@ -2,13 +2,14 @@ import { useEffect, useRef } from "react"; import { BACKEND_URL } from "@repo/common/zod"; +import { useSession } from "next-auth/react"; export interface WorkflowLiveEventCallbacks { workflowId: string | null | undefined; onConnected?: () => void; onTestWebhook?: (payload: any) => void; onNodeStart?: (data: { nodeId: string; executionId: string; startedAt: string }) => void; - onNodeFinish?: (data: { nodeId: string; executionId: string; status: "Completed" | "Failed"; error?: string }) => void; + onNodeFinish?: (data: { nodeId: string; executionId: string; status: "Completed" | "Failed" | "Skipped"; error?: string }) => void; onWorkflowFinish?: (data: { executionId: string; status: "Completed" | "Failed"; error?: string }) => void; } @@ -29,6 +30,9 @@ export function useWorkflowEvents({ onWorkflowFinish, }); + const { data: session } = useSession(); + const token = (session as any)?.accessToken; + useEffect(() => { callbacksRef.current = { onConnected, @@ -40,13 +44,13 @@ export function useWorkflowEvents({ }); useEffect(() => { - if (!workflowId) return; + if (!workflowId || !token) return; let eventSource: EventSource | null = null; let isCleanedUp = false; try { - const url = `${BACKEND_URL}/user/workflow/events/${workflowId}`; + const url = `${BACKEND_URL}/user/workflow/events/${workflowId}?token=${encodeURIComponent(token)}`; eventSource = new EventSource(url, { withCredentials: true }); eventSource.addEventListener("connected", () => { @@ -110,5 +114,5 @@ export function useWorkflowEvents({ eventSource = null; } }; - }, [workflowId]); + }, [workflowId, token]); } diff --git a/apps/web/app/lib/formatters.ts b/apps/web/app/lib/formatters.ts index aac56ba..b851f5a 100644 --- a/apps/web/app/lib/formatters.ts +++ b/apps/web/app/lib/formatters.ts @@ -151,6 +151,8 @@ export const getStatusIcon = (status: string): string => { return '↻'; case 'Start': return '●'; + case 'Skipped': + return '⊘'; default: return '○'; } diff --git a/apps/web/app/workflows/[id]/page.tsx b/apps/web/app/workflows/[id]/page.tsx index cbf882a..d818103 100644 --- a/apps/web/app/workflows/[id]/page.tsx +++ b/apps/web/app/workflows/[id]/page.tsx @@ -219,7 +219,7 @@ export default function WorkflowCanvas() { ...node, data: { ...node.data, - executionStatus: status === "Completed" ? "completed" : "failed", + executionStatus: status === "Completed" ? "completed" : status === "Skipped" ? "skipped" : "failed", }, } : node @@ -542,7 +542,7 @@ export default function WorkflowCanvas() { setNodes(finalNodes); setEdges(finalEdges); - dispatch(workflowActions.setEdge(finalEdges)); + dispatch(workflowActions.setEdge(healedEdges)); } catch (err: any) { if (!isCancelled) { console.error("Failed to load workflow:", err); diff --git a/apps/worker/src/engine/executor.ts b/apps/worker/src/engine/executor.ts index 8dd5362..882468a 100644 --- a/apps/worker/src/engine/executor.ts +++ b/apps/worker/src/engine/executor.ts @@ -86,107 +86,7 @@ export async function executeWorkflow( const nodes = data?.workflow.nodes; console.log(`Total nodes - ${nodes.length}`); - // for (const node of nodes) { - // console.log(`${node.name}, ${node.stage}, ${node.id}th - started Execution`); - // const nodeExecution = await prismaClient.nodeExecution.create({ - // data: { - // nodeId: node.id, - // workflowExecId: workflowExecutionId, - // status: "Start", - // inputData: currentInputData ? currentInputData : {}, - // startedAt: new Date() - // } - // }) - // const nodeType = node.AvailableNode.type; - - // // Create mutable copy of config - // let nodeConfig = { ...(node.config as Record) }; - - // // Build interpolation context from all previously executed nodes - // const interpolationContext = buildInterpolationContext(executedNodeOutputs); - // for (const out of executedNodeOutputs) { - // if (out.nodeId) interpolationContext[out.nodeId] = out.outputData; - // } - // console.log(`[Interpolation] Before: ${JSON.stringify(interpolationContext)}`); - // // Resolve any {{variable}} references in the config - // console.log(`[nodeConfig] Before: ${JSON.stringify(nodeConfig)}`); - // nodeConfig = resolveConfigVariables(nodeConfig, interpolationContext); - // console.log(`[Interpolation] After: ${JSON.stringify(nodeConfig)}`); - - // // NOTE: Removed legacy body concatenation that appended raw JSON to email body. - // // Variables should be resolved via the {{interpolation}} system instead. - // if (!node.CredentialsID) { - // await prismaClient.workflowExecution.update({ - // where: { id: workflowExecutionId }, - // data: { - // status: "Failed", - // error: "Credential id not found", - // completedAt: new Date(), - // }, - // }); - - // await prismaClient.nodeExecution.update({ - // where: { id: nodeExecution.id }, - // data: { - // status: "Failed", - // error: "Credential id not found", - // completedAt: new Date() - // } - // }) - // return; - // } - - // // Check if we need to loop (inputData is spreadsheet + config has column variables) - - - // let execute: { success: boolean; output?: any; error?: string }; - - - // if (!execute.success) { - // // Check if it's a partial loop failure (some rows succeeded) - // const isPartialFailure = execute.output?.successful > 0 && execute.output?.failed > 0; - - // await prismaClient.workflowExecution.update({ - // where: { id: workflowExecutionId }, - // data: { - // status: "Failed", - // error: execute.error, - // completedAt: new Date(), - // } - // }); - - // await prismaClient.nodeExecution.update({ - // where: { id: nodeExecution.id }, - // data: { - // status: "Failed", - // error: execute.error, - // outputData: isPartialFailure ? execute.output : undefined, - // completedAt: new Date() - // } - // }) - // return; - // } - // await prismaClient.nodeExecution.update({ - // where: { id: nodeExecution.id }, - // data: { - // completedAt: new Date(), - // outputData: execute.output, - // status: "Completed" - // } - // }) - - // // Store this node's output for variable resolution in subsequent nodes - // executedNodeOutputs.push({ - // nodeName: node.name, - // nodeId: node.id, - // outputData: execute.output - // }); - // console.log(`[Interpolation] Added ${node.name} output to context. Total nodes in context: ${executedNodeOutputs.length}`); - - // currentInputData = execute.output; - - // console.log("output: ", JSON.stringify(execute)); - // } + const allEdges = (data.workflow.Edges as any[]) || []; const triggerId = data.workflow.Trigger?.id; const triggerOutgoingEdges = triggerId ? allEdges.filter(e => e.source === triggerId) : []; @@ -235,7 +135,7 @@ export async function executeWorkflow( while (queue.length > 0) { const currentTask = queue.shift() if (!currentTask || !currentTask.nodeId) continue; - + // Prevent infinite loops or multiple executions of the same node if (executedNodeIds.has(currentTask.nodeId)) { console.log(`Node ${currentTask.nodeId} already executed, skipping...`); @@ -295,6 +195,45 @@ export async function executeWorkflow( itemsToProcess = currentInputData ? [{ json: currentInputData }] : [] } + if (itemsToProcess.length === 0) { + await prismaClient.nodeExecution.update({ + where: { + id: nodeExecution.id + }, + data: { + completedAt: new Date(), + status: "Skipped", + outputData: [] + } + }); + + await emitLiveEvent({ + type: "NODE_FINISH", + workflowId: data.workflow.id, + executionId: workflowExecutionId, + nodeId: node.id, + status: "Skipped", + completedAt: new Date().toISOString(), + }); + + // Cascade skip to child nodes + const allEdges = (data.workflow.Edges as any[]) || []; + const outgoingEdges = allEdges.filter( + (edge) => + edge.source === node.id && + !edge.target?.startsWith?.("action-placeholder-") && + data.workflow.nodes.some((n) => n.id === edge.target) + ); + + for (const edge of outgoingEdges) { + queue.push({ + nodeId: edge.target, + inputData: [] + }); + } + + continue; + } // Build interpolation context from all previously executed nodes const interpolationContext = buildInterpolationContext(executedNodeOutputs); @@ -423,10 +362,13 @@ export async function executeWorkflow( console.log(`[Interpolation] Added ${node.name} output to context. Total nodes in context: ${executedNodeOutputs.length}`); console.log("output: ", JSON.stringify(execute)); - const allEdges = (data.workflow.Edges as any[]) || [] + const allEdges = (data.workflow.Edges as any[]) || []; const outgoingEdges = allEdges.filter( - (edge) => edge.source === node.id - ) + (edge) => + edge.source === node.id && + !edge.target?.startsWith?.("action-placeholder-") && + data.workflow.nodes.some((n) => n.id === edge.target) + ); for (const edge of outgoingEdges) { // Resolve pin index from sourceHandle (e.g. "out-0", "out-1", "a-out" -> 0) diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index 95bed8b..92c0a31 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -144,4 +144,5 @@ enum WorkflowStatus { ReConnecting Failed Completed + Skipped } diff --git a/packages/kafka/src/index.ts b/packages/kafka/src/index.ts index 3454199..33c6f0a 100644 --- a/packages/kafka/src/index.ts +++ b/packages/kafka/src/index.ts @@ -38,7 +38,7 @@ export interface NodeFinishLiveEvent { workflowId: string; executionId: string; nodeId: string; - status: "Completed" | "Failed"; + status: "Completed" | "Failed" | "Skipped"; completedAt: string; durationMs?: number; error?: string; @@ -48,7 +48,7 @@ export interface WorkflowFinishLiveEvent { type: "WORKFLOW_FINISH"; workflowId: string; executionId: string; - status: "Completed" | "Failed"; + status: "Completed" | "Failed" | "Skipped"; completedAt: string; error?: string; } @@ -79,15 +79,15 @@ const kafkaConfig: KafkaConfig = { requestTimeout: 30000, ...(isSaslEnabled ? { - ssl: { - rejectUnauthorized: process.env.KAFKA_SSL_REJECT_UNAUTHORIZED === "true", - }, - sasl: { - mechanism, - username: username!, - password: password!, - } as SASLOptions, - } + ssl: { + rejectUnauthorized: process.env.KAFKA_SSL_REJECT_UNAUTHORIZED === "true", + }, + sasl: { + mechanism, + username: username!, + password: password!, + } as SASLOptions, + } : {}), };