| 941 | } |
| 942 | |
| 943 | fn collect_downstream_nodes_from_context( |
| 944 | context: &graphbit_core::types::WorkflowContext, |
| 945 | parent_node_ids: &std::collections::HashSet<String>, |
| 946 | ) -> std::collections::HashSet<String> { |
| 947 | let mut downstream_nodes: std::collections::HashSet<String> = std::collections::HashSet::new(); |
| 948 | let Some(deps_obj) = context |
| 949 | .metadata |
| 950 | .get("node_dependencies") |
| 951 | .and_then(|v| v.as_object()) |
| 952 | else { |
| 953 | return downstream_nodes; |
| 954 | }; |
| 955 | |
| 956 | let mut queue: Vec<String> = parent_node_ids.iter().cloned().collect(); |
| 957 | while let Some(parent_id) = queue.pop() { |
| 958 | for (node_id, parents) in deps_obj { |
| 959 | if downstream_nodes.contains(node_id) || parent_node_ids.contains(node_id) { |
| 960 | continue; |
| 961 | } |
| 962 | let Some(parent_array) = parents.as_array() else { |
| 963 | continue; |
| 964 | }; |
| 965 | if parent_array.iter().any(|p| p.as_str() == Some(parent_id.as_str())) { |
| 966 | downstream_nodes.insert(node_id.clone()); |
| 967 | queue.push(node_id.clone()); |
| 968 | } |
| 969 | } |
| 970 | } |
| 971 | downstream_nodes |
| 972 | } |
| 973 | |
| 974 | fn clear_context_for_node_ids( |
| 975 | context: &mut graphbit_core::types::WorkflowContext, |