fix: for the paste node trpc procedure emit the migrated nodes instead of original one. Add support for updating multiple nodes and introduce NodesUpdated event

Author: yakud

apps/chaingraph-frontend/src/store/flow/stores.ts

@@ -51,6 +51,7 @@ import {
   removeNode,
   setNodeVersion,
   updateNode,
+  updateNodes,
   updateNodeUILocal,
 } from '../nodes/stores'
 import {
@@ -432,14 +433,11 @@ function createEventHandlers(flowId: string, nodes: Record<string, INode>): Flow
 
     [FlowEventType.NodeUpdated]: (data) => {
       const node = nodes[data.node.id]
-      if (!node) {
-        console.error(`[NodeUpdated] Node ${data.node.id} not found`)
-        return
-      }
-
-      if (data.node.getVersion() && data.node.getVersion() < node.getVersion()) {
-        // console.warn(`[NodeUpdated] Received outdated node update event for node ${data.node.id}`)
-        return
+      if (node) {
+        if (data.node.getVersion() && data.node.getVersion() < node.getVersion()) {
+          // console.warn(`[NodeUpdated] Received outdated node update event for node ${data.node.id}`)
+          return
+        }
       }
 
       // console.log(`[NodeUpdated] Updating node ${data.node.id} to version ${data.node.getVersion()}`)
@@ -448,6 +446,27 @@ function createEventHandlers(flowId: string, nodes: Record<string, INode>): Flow
       nodeUpdated(data.node.id)
     },
 
+    [FlowEventType.NodesUpdated]: (data) => {
+      const validNodes = data.nodes.filter((node) => {
+        const existingNode = nodes[node.id]
+        if (existingNode) {
+          if (node.getVersion() && node.getVersion() < existingNode.getVersion()) {
+            // console.warn(`[NodesUpdated] Skipping outdated node update event for node ${node.id}`)
+            return false
+          }
+        }
+        return true
+      })
+
+      if (validNodes.length === 0) {
+        return
+      }
+
+      // console.log(`[NodesUpdated] Updating nodes:`, validNodes.map(n => n.id).join(', '))
+      updateNodes(validNodes)
+      validNodes.forEach(node => nodeUpdated(node.id))
+    },
+
     [FlowEventType.NodeRemoved]: (data) => {
       removeNode(data.nodeId)
     },

apps/chaingraph-frontend/src/store/nodes/stores.ts

@@ -42,6 +42,7 @@ import { positionInterpolator } from './position-interpolation-advanced'
 export const addNode = nodesDomain.createEvent<INode>()
 export const addNodes = nodesDomain.createEvent<INode[]>()
 export const updateNode = nodesDomain.createEvent<INode>()
+export const updateNodes = nodesDomain.createEvent<INode[]>()
 export const removeNode = nodesDomain.createEvent<string>()
 export const setNodeMetadata = nodesDomain.createEvent<{ nodeId: string, metadata: NodeState['metadata'] }>()
 export const setNodeVersion = nodesDomain.createEvent<{ nodeId: string, version: number }>()
@@ -214,6 +215,16 @@ export const $nodes = nodesDomain.createStore<Record<string, INode>>({})
     return newState
   })
 
+  // Update nodes operation
+  .on(updateNodes, (state, nodes) => {
+    const newState = { ...state }
+    nodes.forEach((node) => {
+      newState[node.id] = node
+    })
+
+    return newState
+  })
+
   .on(updateNode, (state, node) => {
     return { ...state, [node.id]: node }
   })

packages/chaingraph-trpc/server/procedures/flow/paste-nodes.ts

@@ -7,12 +7,14 @@
  */
 
 import type {
-  Flow,
   IEdge,
   INode,
   IPort,
   SerializedEdge,
 } from '@badaitech/chaingraph-types'
+import {
+  Flow,
+} from '@badaitech/chaingraph-types'
 import {
   FlowMigration,
 } from '@badaitech/chaingraph-types'
@@ -81,10 +83,12 @@ export const pasteNodes = flowContextProcedure
       throw new Error(`Flow ${flowId} not found`)
     }
 
+    const emptyFlow = new Flow()
+
     await ctx.flowStore.lockFlow(flowId)
 
     try {
-      flow.setIsDisabledPropagationEvents(true)
+      emptyFlow.setIsDisabledPropagationEvents(true)
 
       // Step 1: Clone nodes with new IDs using cloneWithNewId()
       const nodeIdMapping = new Map<string, string>()
@@ -192,7 +196,7 @@ export const pasteNodes = flowContextProcedure
         }
       }
 
-      const addedNodes = await flow.addNodes(createdNodes, false)
+      const addedNodes = await emptyFlow.addNodes(createdNodes, true)
 
       // Step 2: Recreate edges using new IDs
       const createdEdges: SerializedEdge[] = []
@@ -241,10 +245,13 @@ export const pasteNodes = flowContextProcedure
         })
         .filter((edge): edge is NonNullable<typeof edge> => edge !== null)
 
-      await flow.addEdges(edgesToAdd, false)
+      await emptyFlow.addEdges(edgesToAdd, true)
 
       // Force migrate flow to v2
-      FlowMigration.migrateFlowFromV1ToV2(flow)
+      const flowWithPastedNodes = FlowMigration.migrateFlowFromV1ToV2(emptyFlow)
+
+      await flow.addNodes(Array.from(flowWithPastedNodes.nodes.values()), false)
+      await flow.addEdges(Array.from(flowWithPastedNodes.edges.values()), false)
 
       // Save the updated flow
       await ctx.flowStore.updateFlow(flow as Flow)

packages/chaingraph-types/src/flow/events.ts

@@ -25,6 +25,7 @@ export enum FlowEventType {
   NodesAdded = 'flow:nodes:added',
   NodeAdded = 'flow:node:added',
   NodeRemoved = 'flow:node:removed',
+  NodesUpdated = 'flow:nodes:updated',
   NodeUpdated = 'flow:node:updated',
   NodeParentUpdated = 'flow:node:parent-updated',
 
@@ -69,6 +70,11 @@ export interface NodeUpdatedEventData {
   node: INode
 }
 
+/** Data for NodesUpdated event */
+export interface NodesUpdatedEventData {
+  nodes: INode[]
+}
+
 /** Data for PortCreated event */
 export interface PortCreatedEventData {
   port: IPort
@@ -183,6 +189,7 @@ export interface EventDataMap {
   [FlowEventType.NodeAdded]: NodeAddedEventData
   [FlowEventType.NodeRemoved]: NodeRemovedEventData
   [FlowEventType.NodeUpdated]: NodeUpdatedEventData
+  [FlowEventType.NodesUpdated]: NodesUpdatedEventData
   [FlowEventType.PortCreated]: PortCreatedEventData
   [FlowEventType.PortRemoved]: PortRemovedEventData
   [FlowEventType.PortUpdated]: PortUpdatedEventData

packages/chaingraph-types/src/flow/flow.ts

@@ -165,10 +165,6 @@ export class Flow implements IFlow {
   }
 
   async updateNode(node: INode): Promise<void> {
-    if (!this.nodes.has(node.id)) {
-      throw new Error(`Node with ID ${node.id} does not exist in the flow.`)
-    }
-
     // Commit any pending batch updates before updating the node
     // This ensures all collected port updates are emitted before the flow update
     await node.commitBatchUpdate({ sourceOfUpdate: 'flow:updateNode' })
@@ -199,6 +195,36 @@ export class Flow implements IFlow {
     ))
   }
 
+  async updateNodes(nodes: INode[]): Promise<void> {
+    if (nodes.length === 0) {
+      return Promise.resolve()
+    }
+
+    // Commit any pending batch updates before updating the nodes
+    await Promise.all(nodes.map(node => node.commitBatchUpdate({ sourceOfUpdate: 'flow:updateNodes' })))
+
+    // cancel the previous event handler if it exists
+    nodes.forEach((node) => {
+      const oldCancel = this.nodeEventHandlersCancel.get(node.id)
+      this.nodes.set(node.id, node)
+      const newCancel = node.onAll(async (event: NodeEvent) => {
+        return this.handleNodeEvent(node, event)
+      })
+      if (oldCancel) {
+        oldCancel()
+      }
+      this.nodeEventHandlersCancel.set(node.id, newCancel)
+    })
+
+    // Emit NodesUpdated event
+    return this.emitEvent(newEvent(
+      this.getNextEventIndex(),
+      this.id,
+      FlowEventType.NodesUpdated,
+      { nodes: nodes.map(n => n.clone()) },
+    ))
+  }
+
   async removeNode(nodeId: string): Promise<void> {
     const node = this.nodes.get(nodeId)
     if (!node) {

packages/chaingraph-types/src/flow/interface.ts

@@ -63,6 +63,12 @@ export interface IFlow {
    */
   updateNode: (node: INode) => Promise<void>
 
+  /**
+   * Updates multiple nodes in the flow and triggers an event.
+   * @param node
+   */
+  updateNodes: (node: INode[]) => Promise<void>
+
   /**
    * Removes a node from the flow.
    * @param nodeId The ID of the node to remove.