greentic_flow/
flow_bundle.rs

1use crate::{
2    error::{FlowError, FlowErrorLocation, Result},
3    loader,
4};
5use blake3::Hasher;
6use greentic_types::Flow;
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9use std::path::Path;
10
11const INLINE_SOURCE_LABEL: &str = "<inline>";
12const EMBEDDED_SCHEMA: &str = include_str!("../schemas/ygtc.flow.schema.json");
13const DEFAULT_SCHEMA_LABEL: &str = "https://raw.githubusercontent.com/greentic-ai/greentic-flow/refs/heads/master/schemas/ygtc.flow.schema.json";
14
15pub type NodeId = String;
16
17#[derive(Clone, Debug, Serialize, Deserialize)]
18pub struct ComponentPin {
19    pub name: String,
20    pub version_req: String,
21}
22
23#[derive(Clone, Debug, Serialize, Deserialize)]
24pub struct NodeRef {
25    pub node_id: String,
26    pub component: ComponentPin,
27    pub schema_id: Option<String>,
28}
29
30#[derive(Clone, Debug, Serialize, Deserialize)]
31pub struct FlowBundle {
32    pub id: String,
33    pub kind: String,
34    pub entry: String,
35    pub yaml: String,
36    pub json: Value,
37    pub hash_blake3: String,
38    pub nodes: Vec<NodeRef>,
39}
40
41/// Canonicalize a JSON value by sorting object keys recursively.
42pub fn canonicalize_json(value: &Value) -> Value {
43    match value {
44        Value::Object(map) => {
45            let mut keys: Vec<_> = map.keys().collect();
46            keys.sort();
47            let mut ordered = serde_json::Map::with_capacity(map.len());
48            for key in keys {
49                ordered.insert(key.clone(), canonicalize_json(&map[key]));
50            }
51            Value::Object(ordered)
52        }
53        Value::Array(items) => Value::Array(items.iter().map(canonicalize_json).collect()),
54        _ => value.clone(),
55    }
56}
57
58/// Compute a lowercase hex-encoded BLAKE3 hash for the provided bytes.
59pub fn blake3_hex(bytes: impl AsRef<[u8]>) -> String {
60    let mut hasher = Hasher::new();
61    hasher.update(bytes.as_ref());
62    let hash = hasher.finalize();
63    hash.to_hex().to_string()
64}
65
66/// Extract component pins from the IR.
67pub fn extract_component_pins(flow: &Flow) -> Vec<(NodeId, ComponentPin)> {
68    flow.nodes
69        .iter()
70        .map(|(node_id, node)| {
71            (
72                node_id.to_string(),
73                ComponentPin {
74                    name: node.component.id.as_str().to_string(),
75                    version_req: "*".to_string(),
76                },
77            )
78        })
79        .collect()
80}
81
82/// Load YAML into a canonical [`FlowBundle`] using the embedded schema.
83pub fn load_and_validate_bundle(yaml: &str, source: Option<&Path>) -> Result<FlowBundle> {
84    load_and_validate_bundle_with_schema_text(
85        yaml,
86        EMBEDDED_SCHEMA,
87        DEFAULT_SCHEMA_LABEL.to_string(),
88        None,
89        source,
90    )
91    .map(|(bundle, _)| bundle)
92}
93
94pub fn load_and_validate_bundle_with_flow(
95    yaml: &str,
96    source: Option<&Path>,
97) -> Result<(FlowBundle, Flow)> {
98    load_and_validate_bundle_with_schema_text(
99        yaml,
100        EMBEDDED_SCHEMA,
101        DEFAULT_SCHEMA_LABEL.to_string(),
102        None,
103        source,
104    )
105}
106
107pub fn load_and_validate_bundle_with_schema_text(
108    yaml: &str,
109    schema_text: &str,
110    schema_label: impl Into<String>,
111    schema_path: Option<&Path>,
112    source: Option<&Path>,
113) -> Result<(FlowBundle, Flow)> {
114    let schema_label = schema_label.into();
115    let source_label = source
116        .map(|p| p.display().to_string())
117        .unwrap_or_else(|| INLINE_SOURCE_LABEL.to_string());
118
119    let flow_doc = loader::load_with_schema_text(
120        yaml,
121        schema_text,
122        schema_label,
123        schema_path,
124        source_label.clone(),
125        source,
126    )?;
127
128    let flow_json = serde_json::to_value(&flow_doc).map_err(|e| FlowError::Internal {
129        message: format!("flow serialization: {e}"),
130        location: FlowErrorLocation::at_path(source_label.clone()).with_source_path(source),
131    })?;
132    let canonical_json = canonicalize_json(&flow_json);
133    let json_bytes = serde_json::to_vec(&canonical_json).map_err(|e| FlowError::Internal {
134        message: format!("canonical json encode: {e}"),
135        location: FlowErrorLocation::at_path(source_label.clone()).with_source_path(source),
136    })?;
137    let hash_blake3 = blake3_hex(&json_bytes);
138
139    let flow = crate::compile_flow(flow_doc.clone())?;
140    let bundle = build_bundle_from_parts(&flow_doc, &flow, yaml, canonical_json, hash_blake3);
141
142    Ok((bundle, flow))
143}
144
145fn build_bundle_from_parts(
146    doc: &crate::model::FlowDoc,
147    flow: &Flow,
148    yaml: &str,
149    canonical_json: Value,
150    hash_blake3: String,
151) -> FlowBundle {
152    let entry = resolve_entry(doc);
153    let nodes = extract_component_pins(flow)
154        .into_iter()
155        .map(|(node_id, component)| NodeRef {
156            node_id,
157            component,
158            schema_id: None,
159        })
160        .collect();
161
162    FlowBundle {
163        id: doc.id.clone(),
164        kind: doc.flow_type.clone(),
165        entry,
166        yaml: yaml.to_string(),
167        json: canonical_json,
168        hash_blake3,
169        nodes,
170    }
171}
172
173fn resolve_entry(doc: &crate::model::FlowDoc) -> String {
174    if let Some(entry) = &doc.start {
175        return entry.clone();
176    }
177    if doc.nodes.contains_key("in") {
178        return "in".to_string();
179    }
180    doc.nodes.keys().next().cloned().unwrap_or_default()
181}
greentic_flow/flow_bundle.rs

greentic_flow/
flow_bundle.rs
