enhancement: Add function to check broken edges (#2882)

* chore: create new function to check broken edges

* enhancement: add edges check when a new flow is added

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* refactor: update BROKEN_EDGES_WARNING message to use "connections" instead of "edges"

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* update function name

* refactor: improve error handling in detectBrokenEdgesEdges function

* remove console.log

Co-authored-by: Gabriel Luiz Freitas Almeida <gabriel@langflow.org>

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Gabriel Luiz Freitas Almeida <gabriel@langflow.org>

src/frontend/src/constants/constants.ts

@@ -734,6 +734,9 @@ export const AUTHORIZED_DUPLICATE_REQUESTS = [
   "/auto_login",
 ];
 
+export const BROKEN_EDGES_WARNING =
+  "Some connections were removed because they were invalid:";
+
 export const SAVE_DEBOUNCE_TIME = 300;
 
 export const IS_MAC = navigator.userAgent.toUpperCase().includes("MAC");

src/frontend/src/stores/flowStore.ts

@@ -1,3 +1,5 @@
+import { BROKEN_EDGES_WARNING } from "@/constants/constants";
+import { brokenEdgeMessage } from "@/utils/utils";
 import { cloneDeep, zip } from "lodash";
 import {
   Edge,
@@ -29,6 +31,7 @@ import {
   checkChatInput,
   checkOldComponents,
   cleanEdges,
+  detectBrokenEdgesEdges,
   getHandleId,
   getNodeId,
   scapeJSONParse,
@@ -122,6 +125,13 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
   },
   resetFlow: ({ nodes, edges, viewport }) => {
     const currentFlow = useFlowsManagerStore.getState().currentFlow;
+    let brokenEdges = detectBrokenEdgesEdges(nodes, edges);
+    if (brokenEdges.length > 0) {
+      useAlertStore.getState().setErrorData({
+        title: BROKEN_EDGES_WARNING,
+        list: brokenEdges.map((edge) => brokenEdgeMessage(edge)),
+      });
+    }
     let newEdges = cleanEdges(nodes, edges);
     const { inputs, outputs } = getInputsAndOutputs(nodes);
     set({

src/frontend/src/stores/flowsManagerStore.ts

@@ -1,10 +1,14 @@
+import { brokenEdgeMessage } from "@/utils/utils";
 import { AxiosError } from "axios";
 import { cloneDeep } from "lodash";
 import pDebounce from "p-debounce";
 import { useLocation } from "react-router-dom";
 import { Edge, Node, Viewport, XYPosition } from "reactflow";
 import { create } from "zustand";
-import { SAVE_DEBOUNCE_TIME } from "../constants/constants";
+import {
+  BROKEN_EDGES_WARNING,
+  SAVE_DEBOUNCE_TIME,
+} from "../constants/constants";
 import {
   deleteFlowFromDatabase,
   multipleDeleteFlowsComponents,
@@ -22,6 +26,7 @@ import {
   addVersionToDuplicates,
   createFlowComponent,
   createNewFlow,
+  detectBrokenEdgesEdges,
   extractFieldsFromComponenents,
   processDataFromFlow,
   processFlows,
@@ -299,6 +304,16 @@ const useFlowsManagerStore = create<FlowsManagerStoreType>((set, get) => ({
         throw error; // Re-throw the error so the caller can handle it if needed
       }
     } else {
+      let brokenEdges = detectBrokenEdgesEdges(
+        flow!.data!.nodes,
+        flow!.data!.edges,
+      );
+      if (brokenEdges.length > 0) {
+        useAlertStore.getState().setErrorData({
+          title: BROKEN_EDGES_WARNING,
+          list: brokenEdges.map((edge) => brokenEdgeMessage(edge)),
+        });
+      }
       useFlowStore
         .getState()
         .paste(

src/frontend/src/utils/reactflowUtils.ts

@@ -102,6 +102,100 @@ export function cleanEdges(nodes: NodeType[], edges: Edge[]) {
   return newEdges;
 }
 
+export function detectBrokenEdgesEdges(nodes: NodeType[], edges: Edge[]) {
+  function generateAlertObject(sourceNode, targetNode, edge) {
+    const targetHandleObject: targetHandleType = scapeJSONParse(
+      edge.targetHandle,
+    );
+    const sourceHandleObject: sourceHandleType = scapeJSONParse(
+      edge.sourceHandle,
+    );
+    const name = sourceHandleObject.name;
+    const output = sourceNode.data.node!.outputs?.find(
+      (output) => output.name === name,
+    );
+
+    return {
+      source: {
+        nodeDisplayName: sourceNode.data.node!.display_name,
+        outputDisplayName: output?.display_name,
+      },
+      target: {
+        displayName: targetNode.data.node!.display_name,
+        field:
+          targetNode.data.node!.template[targetHandleObject.fieldName]
+            .display_name,
+      },
+    };
+  }
+  let newEdges = cloneDeep(edges);
+  let BrokenEdges: {
+    source: {
+      nodeDisplayName: string;
+      outputDisplayName?: string;
+    };
+    target: {
+      displayName: string;
+      field: string;
+    };
+  }[] = [];
+  edges.forEach((edge) => {
+    // check if the source and target node still exists
+    const sourceNode = nodes.find((node) => node.id === edge.source);
+    const targetNode = nodes.find((node) => node.id === edge.target);
+    if (!sourceNode || !targetNode) {
+      newEdges = newEdges.filter((edg) => edg.id !== edge.id);
+      return;
+    }
+    // check if the source and target handle still exists
+    const sourceHandle = edge.sourceHandle; //right
+    const targetHandle = edge.targetHandle; //left
+    if (targetHandle) {
+      const targetHandleObject: targetHandleType = scapeJSONParse(targetHandle);
+      const field = targetHandleObject.fieldName;
+      const id: targetHandleType = {
+        type: targetNode.data.node!.template[field]?.type,
+        fieldName: field,
+        id: targetNode.data.id,
+        inputTypes: targetNode.data.node!.template[field]?.input_types,
+      };
+      if (targetNode.data.node!.template[field]?.proxy) {
+        id.proxy = targetNode.data.node!.template[field]?.proxy;
+      }
+      if (scapedJSONStringfy(id) !== targetHandle) {
+        newEdges = newEdges.filter((e) => e.id !== edge.id);
+        BrokenEdges.push(generateAlertObject(sourceNode, targetNode, edge));
+      }
+    }
+    if (sourceHandle) {
+      const parsedSourceHandle = scapeJSONParse(sourceHandle);
+      const name = parsedSourceHandle.name;
+      const output = sourceNode.data.node!.outputs?.find(
+        (output) => output.name === name,
+      );
+      if (output) {
+        const outputTypes =
+          output!.types.length === 1 ? output!.types : [output!.selected!];
+
+        const id: sourceHandleType = {
+          id: sourceNode.data.id,
+          name: name,
+          output_types: outputTypes,
+          dataType: sourceNode.data.type,
+        };
+        if (scapedJSONStringfy(id) !== sourceHandle) {
+          newEdges = newEdges.filter((e) => e.id !== edge.id);
+          BrokenEdges.push(generateAlertObject(sourceNode, targetNode, edge));
+        }
+      } else {
+        newEdges = newEdges.filter((e) => e.id !== edge.id);
+        BrokenEdges.push(generateAlertObject(sourceNode, targetNode, edge));
+      }
+    }
+  });
+  return BrokenEdges;
+}
+
 export function unselectAllNodes({ updateNodes, data }: unselectAllNodesType) {
   let newNodes = cloneDeep(data);
   newNodes.forEach((node: Node) => {

src/frontend/src/utils/utils.ts

@@ -473,6 +473,21 @@ export function isEndpointNameValid(name: string, maxLength: number): boolean {
   );
 }
 
+export function brokenEdgeMessage({
+  source,
+  target,
+}: {
+  source: {
+    nodeDisplayName: string;
+    outputDisplayName?: string;
+  };
+  target: {
+    displayName: string;
+    field: string;
+  };
+}) {
+  return `${source.nodeDisplayName}${source.outputDisplayName ? " | " + source.outputDisplayName : ""} -> ${target.displayName}${target.field ? " | " + target.field : ""}`;
+}
 export function FormatColumns(columns: ColumnField[]): ColDef<any>[] {
   if (!columns) return [];
   const basic_types = new Set(["date", "number"]);