use std::collections::{BTreeMap, BTreeSet};
use std::sync::LazyLock;
use serde_json::Value;
use crate::pipeline::model::NodeCapability;
use crate::platform::model::MultiNodePackageDefinition;
const NODE_KIND_PREFIX: &str = "n.";
pub fn node_kinds_in_pipeline(value: &Value) -> BTreeSet<String> {
let mut found = BTreeSet::new();
collect_node_kinds(value, &mut found);
found
}
fn collect_node_kinds(value: &Value, out: &mut BTreeSet<String>) {
match value {
Value::Object(map) => {
if let Some(kind) = map
.get("kind")
.and_then(Value::as_str)
.filter(|kind| kind.starts_with(NODE_KIND_PREFIX))
{
out.insert(kind.to_string());
}
for child in map.values() {
collect_node_kinds(child, out);
}
}
Value::Array(items) => {
for item in items {
collect_node_kinds(item, out);
}
}
_ => {}
}
}
pub fn known_node_capabilities() -> &'static BTreeMap<String, BTreeSet<NodeCapability>> {
static KNOWN: LazyLock<BTreeMap<String, BTreeSet<NodeCapability>>> = LazyLock::new(|| {
let mut known = crate::pipeline::nodes::native_node_capabilities().clone();
for definition in
crate::platform::services::NodeRegistryService::embedded_official_definitions()
{
known.insert(
definition.kind,
definition.capabilities.into_iter().collect(),
);
}
known
});
&KNOWN
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct DerivedBundleCapabilities {
pub nodes: BTreeMap<String, BTreeSet<NodeCapability>>,
pub unresolved: BTreeSet<String>,
}
pub fn derive_bundle_capabilities(
package: &MultiNodePackageDefinition,
function_graph: impl Fn(&str) -> Option<Value>,
known: &BTreeMap<String, BTreeSet<NodeCapability>>,
) -> DerivedBundleCapabilities {
let composed_by_function: BTreeMap<String, BTreeSet<String>> = package
.functions
.iter()
.filter_map(|(name, rel_path)| {
let graph = function_graph(rel_path)?;
Some((name.clone(), node_kinds_in_pipeline(&graph)))
})
.collect();
let mut composed: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
for node in &package.nodes {
let mut kinds = BTreeSet::new();
for binding in node.run.iter().chain(
node.lifecycle
.iter()
.flat_map(|lifecycle| lifecycle.on_activate.iter().chain(&lifecycle.on_deactivate)),
) {
if let Some(function) = binding.function.as_deref()
&& let Some(inner) = composed_by_function.get(function)
{
kinds.extend(inner.iter().cloned());
}
}
composed.insert(node.kind.clone(), kinds);
}
let mut derived: BTreeMap<String, BTreeSet<NodeCapability>> = composed
.keys()
.map(|kind| (kind.clone(), BTreeSet::new()))
.collect();
let mut unresolved = BTreeSet::new();
loop {
let mut changed = false;
for (kind, inner_kinds) in &composed {
let mut union = BTreeSet::new();
for inner in inner_kinds {
if let Some(capabilities) = known.get(inner) {
union.extend(capabilities.iter().copied());
} else if let Some(capabilities) = derived.get(inner) {
union.extend(capabilities.iter().copied());
} else {
unresolved.insert(inner.clone());
}
}
let entry = derived.entry(kind.clone()).or_default();
if !union.is_subset(entry) {
entry.extend(union);
changed = true;
}
}
if !changed {
break;
}
}
DerivedBundleCapabilities {
nodes: derived,
unresolved,
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn package(spec: Value) -> MultiNodePackageDefinition {
serde_json::from_value(spec).expect("package spec")
}
fn graph(kinds: &[&str]) -> Value {
json!({
"apiVersion": "zebflow.com/v1",
"kind": "Pipeline",
"metadata": {"name": "fn"},
"spec": {
"id": "fn",
"nodes": kinds
.iter()
.enumerate()
.map(|(index, kind)| json!({"id": format!("n{index}"), "kind": kind}))
.collect::<Vec<_>>(),
"edges": [],
}
})
}
#[test]
fn only_node_kinds_are_read_out_of_a_pipeline_document() {
let kinds = node_kinds_in_pipeline(&graph(&["n.trigger.function", "n.http.request"]));
assert_eq!(
kinds,
BTreeSet::from([
"n.http.request".to_string(),
"n.trigger.function".to_string()
])
);
}
#[test]
fn a_composite_is_the_union_of_the_nodes_it_composes() {
let spec = package(json!({
"package": "acme",
"version": "1.0.0",
"title": "Acme",
"functions": {"main": "functions/main.zf.json"},
"nodes": [{
"kind": "n.x.acme.sync",
"title": "Sync",
"run": {"function": "main"},
"definition": {}
}]
}));
let known = BTreeMap::from([
(
"n.http.request".to_string(),
BTreeSet::from([NodeCapability::Network]),
),
(
"n.fs.put".to_string(),
BTreeSet::from([NodeCapability::Filesystem]),
),
("n.trigger.function".to_string(), BTreeSet::new()),
]);
let derived = derive_bundle_capabilities(
&spec,
|rel| {
(rel == "functions/main.zf.json")
.then(|| graph(&["n.trigger.function", "n.http.request", "n.fs.put"]))
},
&known,
);
assert_eq!(
derived.nodes["n.x.acme.sync"],
BTreeSet::from([NodeCapability::Network, NodeCapability::Filesystem])
);
assert!(derived.unresolved.is_empty());
}
#[test]
fn a_wasm_node_derives_no_capability_at_all() {
let spec = package(json!({
"package": "acme",
"version": "1.0.0",
"title": "Acme",
"modules": {"core": {"path": "wasm/core.wasm", "abi": "zebflow-wasm-json-v1"}},
"nodes": [{
"kind": "n.x.acme.crunch",
"title": "Crunch",
"run": {"module": "core", "export": "run"},
"definition": {}
}]
}));
let derived = derive_bundle_capabilities(&spec, |_| None, &BTreeMap::new());
assert!(derived.nodes["n.x.acme.crunch"].is_empty());
assert!(derived.unresolved.is_empty());
}
#[test]
fn a_composite_inherits_through_another_node_of_its_own_bundle() {
let spec = package(json!({
"package": "acme",
"version": "1.0.0",
"title": "Acme",
"functions": {
"outer": "functions/outer.zf.json",
"inner": "functions/inner.zf.json"
},
"nodes": [
{"kind": "n.x.acme.outer", "title": "Outer",
"run": {"function": "outer"}, "definition": {}},
{"kind": "n.x.acme.inner", "title": "Inner",
"run": {"function": "inner"}, "definition": {}}
]
}));
let known = BTreeMap::from([(
"n.pg.query".to_string(),
BTreeSet::from([NodeCapability::Database, NodeCapability::Credential]),
)]);
let derived = derive_bundle_capabilities(
&spec,
|rel| match rel {
"functions/outer.zf.json" => Some(graph(&["n.x.acme.inner"])),
"functions/inner.zf.json" => Some(graph(&["n.pg.query", "n.x.acme.outer"])),
_ => None,
},
&known,
);
let expected = BTreeSet::from([NodeCapability::Database, NodeCapability::Credential]);
assert_eq!(derived.nodes["n.x.acme.outer"], expected);
assert_eq!(derived.nodes["n.x.acme.inner"], expected);
}
#[test]
fn a_kind_nothing_provides_is_reported_as_unresolved() {
let spec = package(json!({
"package": "acme",
"version": "1.0.0",
"title": "Acme",
"functions": {"main": "functions/main.zf.json"},
"nodes": [{
"kind": "n.x.acme.sync",
"title": "Sync",
"run": {"function": "main"},
"definition": {}
}]
}));
let derived = derive_bundle_capabilities(
&spec,
|_| Some(graph(&["n.x.elsewhere.thing"])),
&BTreeMap::new(),
);
assert_eq!(
derived.unresolved,
BTreeSet::from(["n.x.elsewhere.thing".to_string()])
);
assert!(derived.nodes["n.x.acme.sync"].is_empty());
}
#[test]
fn an_embedded_composite_reports_what_its_function_pipeline_reaches() {
let embedding = known_node_capabilities()
.get("n.ai.embedding")
.expect("the openai-embedding bundle is embedded in this binary");
assert!(
embedding.contains(&NodeCapability::Network),
"{embedding:?}"
);
assert!(
embedding.contains(&NodeCapability::Credential),
"{embedding:?}"
);
assert!(
embedding.contains(&NodeCapability::Process),
"{embedding:?}"
);
}
#[test]
fn every_native_kind_has_a_capability_answer() {
let known = known_node_capabilities();
for definition in crate::pipeline::nodes::builtin_node_definitions() {
assert!(
known.contains_key(&definition.kind),
"{} has no capability entry",
definition.kind
);
}
}
}