jellyflow-runtime 0.2.0

Headless store, rules, schema, profile, and change pipeline for Jellyflow.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

use crate::io::NodeGraphInteractionState;
use crate::rules::{ConnectPlan, InsertNodeSpec};
use jellyflow_core::core::{EdgeId, Graph, PortId};
use jellyflow_core::interaction::NodeGraphConnectionMode;

use super::super::common::{
    ConnectionOpBuilder, edge_between, ensure_edge_id_available, resolve_policy_checked_connection,
    validate_insert_node_spec,
};
use super::batch::InsertNodeMutationBuilder;

/// Plans connecting two ports by inserting a node between them.
///
/// This is intended for "auto-fix" workflows like inserting a conversion node when types mismatch.
pub fn plan_connect_by_inserting_node_with_policy(
    graph: &Graph,
    a: PortId,
    b: PortId,
    first_edge_id: EdgeId,
    second_edge_id: EdgeId,
    inserted: InsertNodeSpec,
    state: &NodeGraphInteractionState,
) -> ConnectPlan {
    let endpoints = match resolve_policy_checked_connection(
        graph,
        a,
        b,
        NodeGraphConnectionMode::Strict,
        state,
    ) {
        Ok(endpoints) => endpoints,
        Err(plan) => return plan,
    };

    if let Err(reject) = ensure_edge_id_available(graph, first_edge_id) {
        return reject;
    }
    if let Err(reject) = ensure_edge_id_available(graph, second_edge_id) {
        return reject;
    }

    let expected_port_kind = endpoints.edge_kind.port_kind();
    let inserted_ports = match validate_insert_node_spec(
        graph,
        &inserted,
        endpoints.from.node,
        endpoints.to.node,
        expected_port_kind,
    ) {
        Ok(inserted_ports) => inserted_ports,
        Err(plan) => return plan,
    };

    let mut ops = ConnectionOpBuilder::with_endpoint_capacity_disconnects(graph, &endpoints, None);

    let mut batch = match InsertNodeMutationBuilder::new(graph, inserted) {
        Ok(batch) => batch,
        Err(plan) => return plan,
    };
    if let Err(plan) = batch.add_edge(
        first_edge_id,
        edge_between(endpoints.edge_kind, endpoints.from_id, inserted_ports.input),
    ) {
        return plan;
    }
    if let Err(plan) = batch.add_edge(
        second_edge_id,
        edge_between(endpoints.edge_kind, inserted_ports.output, endpoints.to_id),
    ) {
        return plan;
    }
    ops.extend(batch.into_ops());

    ConnectPlan::from_ops(ops.into_ops())
}

/// Plans connecting two ports by inserting a node with default interaction policy.
pub fn plan_connect_by_inserting_node(
    graph: &Graph,
    a: PortId,
    b: PortId,
    first_edge_id: EdgeId,
    second_edge_id: EdgeId,
    inserted: InsertNodeSpec,
) -> ConnectPlan {
    plan_connect_by_inserting_node_with_policy(
        graph,
        a,
        b,
        first_edge_id,
        second_edge_id,
        inserted,
        &NodeGraphInteractionState::default(),
    )
}