Merge pull request #325 from badaitech/feat/execution-consumer-optimization

yakud · web-flow · commit 6306f2c125cf · 2025-09-24T23:22:36.000+03:00
feat: optimize flow node and edge addition with batch processing for improved performance
diff --git a/.changeset/pre.json b/.changeset/pre.json
@@ -45,6 +45,7 @@
     "tall-bags-follow",
     "tiny-things-report",
     "tired-days-sleep",
-    "tough-rice-spend"
+    "tough-rice-spend",
+    "vast-dryers-beg"
   ]
 }
diff --git a/apps/chaingraph-backend/CHANGELOG.md b/apps/chaingraph-backend/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @badaitech/chaingraph-backend
 
+## 0.5.5-dev.30
+
+### Patch Changes
+
+- feat: optimize flow node and edge addition with batch processing for improved performance
+
 ## 0.5.5-dev.29
 
 ### Patch Changes
diff --git a/apps/chaingraph-backend/package.json b/apps/chaingraph-backend/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-backend",
   "type": "module",
-  "version": "0.5.5-dev.29",
+  "version": "0.5.5-dev.30",
   "private": false,
   "description": "Backend server for the Chaingraph project",
   "license": "BUSL-1.1",
diff --git a/apps/chaingraph-execution-api/CHANGELOG.md b/apps/chaingraph-execution-api/CHANGELOG.md
@@ -1,5 +1,15 @@
 # @badaitech/chaingraph-execution-api
 
+## 0.5.5-dev.30
+
+### Patch Changes
+
+- feat: optimize flow node and edge addition with batch processing for improved performance
+- Updated dependencies
+  - @badaitech/chaingraph-executor@0.5.5-dev.30
+  - @badaitech/chaingraph-types@0.5.5-dev.30
+  - @badaitech/chaingraph-trpc@0.5.5-dev.30
+
 ## 0.5.5-dev.29
 
 ### Patch Changes
diff --git a/apps/chaingraph-execution-api/package.json b/apps/chaingraph-execution-api/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-execution-api",
   "type": "module",
-  "version": "0.5.5-dev.29",
+  "version": "0.5.5-dev.30",
   "private": false,
   "description": "Chaingraph tRPC Server - Scalable API server for Chaingraph execution management",
   "license": "BUSL-1.1",
diff --git a/apps/chaingraph-execution-worker/CHANGELOG.md b/apps/chaingraph-execution-worker/CHANGELOG.md
@@ -1,5 +1,16 @@
 # @badaitech/chaingraph-execution-worker
 
+## 0.5.5-dev.30
+
+### Patch Changes
+
+- feat: optimize flow node and edge addition with batch processing for improved performance
+- Updated dependencies
+  - @badaitech/chaingraph-executor@0.5.5-dev.30
+  - @badaitech/chaingraph-nodes@0.5.5-dev.30
+  - @badaitech/chaingraph-types@0.5.5-dev.30
+  - @badaitech/chaingraph-trpc@0.5.5-dev.30
+
 ## 0.5.5-dev.29
 
 ### Patch Changes
diff --git a/apps/chaingraph-execution-worker/package.json b/apps/chaingraph-execution-worker/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-execution-worker",
   "type": "module",
-  "version": "0.5.5-dev.29",
+  "version": "0.5.5-dev.30",
   "private": false,
   "description": "Chaingraph Execution Worker Service which connected to Kafka and process the flow executions",
   "license": "BUSL-1.1",
diff --git a/apps/chaingraph-frontend/CHANGELOG.md b/apps/chaingraph-frontend/CHANGELOG.md
@@ -1,5 +1,17 @@
 # @badaitech/chaingraph-frontend
 
+## 0.5.5-dev.30
+
+### Patch Changes
+
+- feat: optimize flow node and edge addition with batch processing for improved performance
+- Updated dependencies
+  - @badaitech/chaingraph-executor@0.5.5-dev.30
+  - @badaitech/chaingraph-nodes@0.5.5-dev.30
+  - @badaitech/chaingraph-types@0.5.5-dev.30
+  - @badaitech/chaingraph-trpc@0.5.5-dev.30
+  - @badaitech/badai-api@0.5.5-dev.30
+
 ## 0.5.5-dev.29
 
 ### Patch Changes
diff --git a/apps/chaingraph-frontend/package.json b/apps/chaingraph-frontend/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-frontend",
   "type": "module",
-  "version": "0.5.5-dev.29",
+  "version": "0.5.5-dev.30",
   "private": false,
   "description": "Frontend application for the Chaingraph project",
   "license": "BUSL-1.1",
diff --git a/packages/badai-api-example/package.json b/packages/badai-api-example/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/badai-api-example",
   "type": "module",
-  "version": "0.5.5-dev.29",
+  "version": "0.5.5-dev.30",
   "private": true,
   "description": "Bad AI GraphQL API client example",
   "license": "BUSL-1.1",
diff --git a/packages/badai-api/CHANGELOG.md b/packages/badai-api/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @badaitech/badai-api
 
+## 0.5.5-dev.30
+
+### Patch Changes
+
+- feat: optimize flow node and edge addition with batch processing for improved performance
+
 ## 0.5.5-dev.29
 
 ### Patch Changes
diff --git a/packages/badai-api/package.json b/packages/badai-api/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/badai-api",
   "type": "module",
-  "version": "0.5.5-dev.29",
+  "version": "0.5.5-dev.30",
   "private": false,
   "description": "Bad AI GraphQL API client",
   "license": "BUSL-1.1",
diff --git a/packages/chaingraph-executor/package.json b/packages/chaingraph-executor/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-executor",
   "type": "module",
-  "version": "0.5.5-dev.29",
+  "version": "0.5.5-dev.30",
   "private": false,
   "description": "Chaingraph Message Bus for inter-service communication",
   "license": "BUSL-1.1",
diff --git a/packages/chaingraph-nodes/CHANGELOG.md b/packages/chaingraph-nodes/CHANGELOG.md
@@ -1,5 +1,13 @@
 # @badaitech/chaingraph-nodes
 
+## 0.5.5-dev.30
+
+### Patch Changes
+
+- feat: optimize flow node and edge addition with batch processing for improved performance
+- Updated dependencies
+  - @badaitech/chaingraph-types@0.5.5-dev.30
+
 ## 0.5.5-dev.29
 
 ### Patch Changes
diff --git a/packages/chaingraph-nodes/package.json b/packages/chaingraph-nodes/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-nodes",
   "type": "module",
-  "version": "0.5.5-dev.29",
+  "version": "0.5.5-dev.30",
   "private": false,
   "description": "Node components for the Chaingraph project",
   "license": "BUSL-1.1",
diff --git a/packages/chaingraph-trpc/CHANGELOG.md b/packages/chaingraph-trpc/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @badaitech/chaingraph-trpc
 
+## 0.5.5-dev.30
+
+### Patch Changes
+
+- feat: optimize flow node and edge addition with batch processing for improved performance
+
 ## 0.5.5-dev.29
 
 ### Patch Changes
diff --git a/packages/chaingraph-trpc/package.json b/packages/chaingraph-trpc/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-trpc",
   "type": "module",
-  "version": "0.5.5-dev.29",
+  "version": "0.5.5-dev.30",
   "private": false,
   "description": "tRPC client and server for the Chaingraph V2",
   "license": "BUSL-1.1",
diff --git a/packages/chaingraph-trpc/server/procedures/flow/flow-fork.ts b/packages/chaingraph-trpc/server/procedures/flow/flow-fork.ts
@@ -83,17 +83,8 @@ export const fork = authedProcedure
     const forkedFlow = new Flow(metadataWithoutId)
 
     // Transfer nodes and edges to the new flow
-    await Promise.all(
-      clonedFlow.nodes.values().map((node) => {
-        return forkedFlow.addNode(node, true)
-      }),
-    )
-
-    await Promise.all(
-      clonedFlow.edges.values().map((edge) => {
-        return forkedFlow.addEdge(edge)
-      }),
-    )
+    await forkedFlow.addNodes(Array.from(clonedFlow.nodes.values()), true)
+    await forkedFlow.addEdges(Array.from(clonedFlow.edges.values()), true)
 
     // Configure fork metadata
     forkedFlow.metadata.name = input.name || `${originalFlow.metadata.name} (Fork)`
diff --git a/packages/chaingraph-trpc/server/procedures/flow/paste-nodes.ts b/packages/chaingraph-trpc/server/procedures/flow/paste-nodes.ts
@@ -8,10 +8,16 @@
 
 import type {
   Flow,
+  IEdge,
   INode,
   IPort,
   SerializedEdge,
 } from '@badaitech/chaingraph-types'
+import {
+  Edge,
+  EdgeStatus,
+
+} from '@badaitech/chaingraph-types'
 import {
   SerializedNodeSchemaV1Legacy,
 } from '@badaitech/chaingraph-types'
@@ -20,9 +26,15 @@ import {
 } from '@badaitech/chaingraph-types'
 import { NodeStatus } from '@badaitech/chaingraph-types'
 import { SerializedEdgeSchema, SerializedNodeSchema } from '@badaitech/chaingraph-types'
+import { customAlphabet } from 'nanoid'
+import { nolookalikes } from 'nanoid-dictionary'
 import { z } from 'zod'
 import { flowContextProcedure } from '../../trpc'
 
+function generateEdgeID(): string {
+  return `ED${customAlphabet(nolookalikes, 18)()}`
+}
+
 // Input schema for position
 const PositionSchema = z.object({
   x: z.number(),
@@ -110,9 +122,7 @@ export const pasteNodes = flowContextProcedure
               y: pastePosition.y + relativePosition.y,
             }
 
-            clonedNode.setPosition(newPosition, true)
-          } else {
-            console.debug(`[FLOW] Child node ${originalNode.id} keeping relative position:`, originalNode.metadata.ui?.position)
+            clonedNode.setPosition(newPosition, false)
           }
           // For child nodes, keep the original position (relative to parent)
 
@@ -185,82 +195,57 @@ export const pasteNodes = flowContextProcedure
         }
       }
 
-      const addedNodes = await flow.addNodes(createdNodes)
+      const addedNodes = await flow.addNodes(createdNodes, false)
 
       // Step 2: Recreate edges using new IDs
       const createdEdges: SerializedEdge[] = []
+      const skippedEdges: string[] = []
 
       // Prepare edge connection data with validated IDs
-      const edgeConnectionTasks = clipboardData.edges
+      const edgesToAdd = clipboardData.edges
         .map((edgeData) => {
           const newSourceNodeId = nodeIdMapping.get(edgeData.sourceNodeId)
           const newTargetNodeId = nodeIdMapping.get(edgeData.targetNodeId)
           const newSourcePortId = portIdMapping.get(edgeData.sourcePortId)
           const newTargetPortId = portIdMapping.get(edgeData.targetPortId)
 
           if (!newSourceNodeId || !newTargetNodeId || !newSourcePortId || !newTargetPortId) {
+            skippedEdges.push(edgeData.id)
             return null
           }
 
-          return {
-            originalEdgeData: edgeData,
-            newSourceNodeId,
-            newTargetNodeId,
-            newSourcePortId,
-            newTargetPortId,
-          }
-        })
-        .filter((task): task is NonNullable<typeof task> => task !== null)
-
-      // Execute all edge connections in parallel
-      const edgeConnectionResults = await Promise.allSettled(
-        edgeConnectionTasks.map(async task => ({
-          task,
-          edge: await flow.connectPorts(
-            task.newSourceNodeId,
-            task.newSourcePortId,
-            task.newTargetNodeId,
-            task.newTargetPortId,
-          ),
-        })),
-      )
-
-      // Process results and update metadata
-      for (const result of edgeConnectionResults) {
-        if (result.status === 'rejected') {
-          console.error('[FLOW] Failed to create edge:', result.reason)
-          continue
-        }
-
-        const { task, edge } = result.value
+          const sourceNode = addedNodes.find(n => n.id === newSourceNodeId)
+          const targetNode = addedNodes.find(n => n.id === newTargetNodeId)
 
-        if (!edge) {
-          console.warn(`[FLOW] Failed to connect ports for edge ${task.originalEdgeData.id}`)
-          continue
-        }
-
-        // Set edge metadata if provided
-        if (task.originalEdgeData.metadata) {
-          if (task.originalEdgeData.metadata.label && typeof task.originalEdgeData.metadata.label === 'string') {
-            edge.metadata.label = task.originalEdgeData.metadata.label
+          if (!sourceNode || !targetNode) {
+            skippedEdges.push(edgeData.id)
+            console.warn(`[FLOW] Skipping edge ${edgeData.id} because new source or target node was not found (source: ${newSourceNodeId}, target: ${newTargetNodeId})`)
+            return null
           }
-          if (task.originalEdgeData.metadata.description) {
-            edge.metadata.description = task.originalEdgeData.metadata.description
+
+          const sourcePort = sourceNode.ports.get(newSourcePortId)
+          const targetPort = targetNode.ports.get(newTargetPortId)
+          if (!targetPort || !sourcePort) {
+            skippedEdges.push(edgeData.id)
+            console.warn(`[FLOW] Skipping edge ${edgeData.id} because new source or target port was not found (source: ${newSourcePortId}, target: ${newTargetPortId})`)
+            return null
           }
-        }
 
-        createdEdges.push({
-          id: edge.id,
-          metadata: edge.metadata,
-          status: edge.status,
-          sourceNodeId: task.newSourceNodeId,
-          sourcePortId: task.newSourcePortId,
-          targetNodeId: task.newTargetNodeId,
-          targetPortId: task.newTargetPortId,
+          const edge = new Edge(
+            generateEdgeID(),
+            sourceNode,
+            sourcePort,
+            targetNode,
+            targetPort,
+            edgeData.metadata || {},
+          )
+          edge.status = edgeData.status || EdgeStatus.Active
+
+          return edge as IEdge
         })
+        .filter((edge): edge is NonNullable<typeof edge> => edge !== null)
 
-        console.debug(`[FLOW] Created edge ${edge.id}`)
-      }
+      await flow.addEdges(edgesToAdd, false)
 
       // Save the updated flow
       await ctx.flowStore.updateFlow(flow as Flow)
@@ -276,6 +261,7 @@ export const pasteNodes = flowContextProcedure
           position: node.metadata.ui?.position,
         })),
         createdEdges,
+        skippedEdges,
         nodeIdMapping: Object.fromEntries(nodeIdMapping),
         portIdMapping: Object.fromEntries(portIdMapping),
       }
diff --git a/packages/chaingraph-types/CHANGELOG.md b/packages/chaingraph-types/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @badaitech/chaingraph-types
 
+## 0.5.5-dev.30
+
+### Patch Changes
+
+- feat: optimize flow node and edge addition with batch processing for improved performance
+
 ## 0.5.5-dev.29
 
 ### Patch Changes
diff --git a/packages/chaingraph-types/package.json b/packages/chaingraph-types/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-types",
   "type": "module",
-  "version": "0.5.5-dev.29",
+  "version": "0.5.5-dev.30",
   "private": false,
   "description": "Core type definitions for the Chaingraph project",
   "license": "BUSL-1.1",
diff --git a/packages/chaingraph-types/src/flow/flow.ts b/packages/chaingraph-types/src/flow/flow.ts
diff --git a/packages/chaingraph-types/src/flow/interface.ts b/packages/chaingraph-types/src/flow/interface.ts
diff --git a/packages/typescript-config/CHANGELOG.md b/packages/typescript-config/CHANGELOG.md
diff --git a/packages/typescript-config/package.json b/packages/typescript-config/package.json

Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,7 @@`
`45`	`45`	`"tall-bags-follow",`
`46`	`46`	`"tiny-things-report",`
`47`	`47`	`"tired-days-sleep",`
`48`		`- "tough-rice-spend"`
	`48`	`+ "tough-rice-spend",`
	`49`	`+ "vast-dryers-beg"`
`49`	`50`	`]`
`50`	`51`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@badaitech/chaingraph-backend",`
`3`	`3`	`"type": "module",`
`4`		`- "version": "0.5.5-dev.29",`
	`4`	`+ "version": "0.5.5-dev.30",`
`5`	`5`	`"private": false,`
`6`	`6`	`"description": "Backend server for the Chaingraph project",`
`7`	`7`	`"license": "BUSL-1.1",`
Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@badaitech/chaingraph-execution-api",`
`3`	`3`	`"type": "module",`
`4`		`- "version": "0.5.5-dev.29",`
	`4`	`+ "version": "0.5.5-dev.30",`
`5`	`5`	`"private": false,`
`6`	`6`	`"description": "Chaingraph tRPC Server - Scalable API server for Chaingraph execution management",`
`7`	`7`	`"license": "BUSL-1.1",`
Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@badaitech/chaingraph-frontend",`
`3`	`3`	`"type": "module",`
`4`		`- "version": "0.5.5-dev.29",`
	`4`	`+ "version": "0.5.5-dev.30",`
`5`	`5`	`"private": false,`
`6`	`6`	`"description": "Frontend application for the Chaingraph project",`
`7`	`7`	`"license": "BUSL-1.1",`
Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@badaitech/badai-api-example",`
`3`	`3`	`"type": "module",`
`4`		`- "version": "0.5.5-dev.29",`
	`4`	`+ "version": "0.5.5-dev.30",`
`5`	`5`	`"private": true,`
`6`	`6`	`"description": "Bad AI GraphQL API client example",`
`7`	`7`	`"license": "BUSL-1.1",`