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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

use crate::rules::{ConnectPlan, InsertNodeSpec};
use jellyflow_core::core::{Graph, NodeId, Port, PortDirection, PortId, PortKind};

pub(in crate::rules::connection) struct ValidatedInsertNodeSpec {
    pub input: PortId,
    pub output: PortId,
}

pub(in crate::rules::connection) fn validate_insert_node_spec(
    graph: &Graph,
    inserted: &InsertNodeSpec,
    source_node: NodeId,
    target_node: NodeId,
    expected_port_kind: PortKind,
) -> Result<ValidatedInsertNodeSpec, ConnectPlan> {
    InsertNodeSpecValidator {
        graph,
        inserted,
        source_node,
        target_node,
        expected_port_kind,
    }
    .validate()
}

struct InsertNodeSpecValidator<'a> {
    graph: &'a Graph,
    inserted: &'a InsertNodeSpec,
    source_node: NodeId,
    target_node: NodeId,
    expected_port_kind: PortKind,
}

struct InsertedPortRoles<'a> {
    input: &'a Port,
    output: &'a Port,
}

impl InsertNodeSpecValidator<'_> {
    fn validate(&self) -> Result<ValidatedInsertNodeSpec, ConnectPlan> {
        self.ensure_inserted_node_is_distinct()?;
        self.ensure_inserted_ids_are_available()?;
        self.ensure_roles_are_distinct()?;

        let roles = self.validate_ports()?;
        self.ensure_role_directions(&roles)?;

        Ok(ValidatedInsertNodeSpec {
            input: self.inserted.input,
            output: self.inserted.output,
        })
    }

    fn ensure_inserted_node_is_distinct(&self) -> Result<(), ConnectPlan> {
        if self.inserted.node_id == self.source_node || self.inserted.node_id == self.target_node {
            return Err(ConnectPlan::reject(
                "inserted node id must be distinct from endpoints",
            ));
        }
        Ok(())
    }

    fn ensure_inserted_ids_are_available(&self) -> Result<(), ConnectPlan> {
        if self.graph.nodes.contains_key(&self.inserted.node_id) {
            return Err(ConnectPlan::reject(format!(
                "node already exists: {:?}",
                self.inserted.node_id
            )));
        }
        for (port_id, _) in &self.inserted.ports {
            if self.graph.ports.contains_key(port_id) {
                return Err(ConnectPlan::reject(format!(
                    "port already exists: {port_id:?}"
                )));
            }
        }
        Ok(())
    }

    fn ensure_roles_are_distinct(&self) -> Result<(), ConnectPlan> {
        if self.inserted.input == self.inserted.output {
            return Err(ConnectPlan::reject(
                "inserted input/output ports must be distinct",
            ));
        }
        Ok(())
    }

    fn validate_ports(&self) -> Result<InsertedPortRoles<'_>, ConnectPlan> {
        let mut inserted_in: Option<&Port> = None;
        let mut inserted_out: Option<&Port> = None;
        for (port_id, port) in &self.inserted.ports {
            self.validate_port_owner(*port_id, port)?;
            self.validate_port_kind(*port_id, port)?;

            if *port_id == self.inserted.input {
                inserted_in = Some(port);
            }
            if *port_id == self.inserted.output {
                inserted_out = Some(port);
            }
        }

        let Some(input) = inserted_in else {
            return Err(ConnectPlan::reject(
                "inserted input port is missing from spec",
            ));
        };
        let Some(output) = inserted_out else {
            return Err(ConnectPlan::reject(
                "inserted output port is missing from spec",
            ));
        };

        Ok(InsertedPortRoles { input, output })
    }

    fn validate_port_owner(&self, port_id: PortId, port: &Port) -> Result<(), ConnectPlan> {
        if port.node != self.inserted.node_id {
            return Err(ConnectPlan::reject(format!(
                "inserted port has wrong node: port={port_id:?} expected={:?} got={:?}",
                self.inserted.node_id, port.node
            )));
        }
        Ok(())
    }

    fn validate_port_kind(&self, port_id: PortId, port: &Port) -> Result<(), ConnectPlan> {
        if port.kind != self.expected_port_kind {
            return Err(ConnectPlan::reject(format!(
                "inserted port kind is incompatible: port={port_id:?} kind={:?} expected={:?}",
                port.kind, self.expected_port_kind
            )));
        }
        Ok(())
    }

    fn ensure_role_directions(&self, roles: &InsertedPortRoles<'_>) -> Result<(), ConnectPlan> {
        if roles.input.dir != PortDirection::In || roles.output.dir != PortDirection::Out {
            return Err(ConnectPlan::reject("inserted ports must be in -> out"));
        }
        Ok(())
    }
}