af-workflow 0.7.0

//! Node-type registry. Port of `platform/registry.py`.
//!
//! Maps a node type string (`transform.state_append`) to a factory that builds
//! a [`StepNode`] from its config. Products extend the registry with their own
//! business node types without touching this crate.

use std::collections::{BTreeMap, HashMap};

use serde_json::Value;

use crate::node::StepNode;
use crate::{CapabilityManifest, CapabilityPin, GuardKind};

/// Node construction or registry failure.
#[derive(Debug, thiserror::Error)]
pub enum NodeError {
    /// Unknown node type '' (not registered).
    #[error("unknown node type '{0}' (not registered)")]
    UnknownType(String),
    /// Node '`node_type`' has invalid config: `reason`.
    #[error("node '{node_type}' has invalid config: {reason}")]
    InvalidConfig {
        /// Node type being built.
        node_type: String,
        /// What is wrong with the config.
        reason: String,
    },
    /// Invalid capability manifest.
    #[error("invalid capability manifest: {0}")]
    InvalidCapability(String),
    /// Capability '' is already registered.
    #[error("capability '{0}' is already registered")]
    DuplicateCapability(String),
}

/// Builds a step node from its (raw, unresolved) config value.
pub type StepFactory = fn(config: &Value) -> Result<Box<dyn StepNode>, NodeError>;

/// Config field type in a node schema.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum FieldType {
    /// String.
    String,
    /// Number.
    Number,
    /// Boolean.
    Bool,
    /// Array.
    Array,
    /// Object.
    Object,
    /// Any JSON value.
    Any,
}

/// One config field.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct FieldSpec {
    /// Field name.
    pub key: String,
    /// Expected type.
    pub ty: FieldType,
    /// Whether it must be present.
    pub required: bool,
}

impl NodeSchema {
    /// JSON Schema projection used by forms and the same validator rule.
    pub fn to_json_schema(&self) -> Value {
        let mut properties = serde_json::Map::new();
        let mut required = Vec::new();
        for field in &self.fields {
            let schema = match field.ty {
                FieldType::String => serde_json::json!({"type":"string"}),
                FieldType::Number => serde_json::json!({"type":"number"}),
                FieldType::Bool => serde_json::json!({"type":"boolean"}),
                FieldType::Array => serde_json::json!({"type":"array"}),
                FieldType::Object => serde_json::json!({"type":"object"}),
                FieldType::Any => serde_json::json!({}),
            };
            properties.insert(field.key.clone(), schema);
            if field.required {
                required.push(field.key.clone());
            }
        }
        serde_json::json!({"type":"object","properties":properties,"required":required,"additionalProperties":false})
    }
}

impl FieldSpec {
    /// A required field.
    pub fn required(key: impl Into<String>, ty: FieldType) -> Self {
        Self {
            key: key.into(),
            ty,
            required: true,
        }
    }

    /// An optional field.
    pub fn optional(key: impl Into<String>, ty: FieldType) -> Self {
        Self {
            key: key.into(),
            ty,
            required: false,
        }
    }
}

/// Config schema of a node type.
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct NodeSchema {
    /// Fields in declaration order.
    pub fields: Vec<FieldSpec>,
}

/// Registry of step-node factories. Ingress types are tracked by name only —
/// the runner owns their execution.
#[derive(Debug, Clone)]
pub struct NodeRegistry {
    steps: HashMap<String, StepFactory>,
    ingress: std::collections::HashSet<String>,
    fan_out: std::collections::HashSet<String>,
    side_effect_guards: std::collections::HashSet<String>,
    guard_kinds: HashMap<String, GuardKind>,
    capabilities: BTreeMap<CapabilityPin, CapabilityManifest>,
    capability_steps: BTreeMap<CapabilityPin, StepFactory>,
    capability_schemas: BTreeMap<CapabilityPin, NodeSchema>,
    capability_guards: BTreeMap<CapabilityPin, GuardKind>,
    capability_fan_out: std::collections::BTreeSet<CapabilityPin>,
    schemas: HashMap<String, NodeSchema>,
}

impl NodeRegistry {
    /// Empty registry — no node types known.
    pub fn empty() -> Self {
        Self {
            steps: HashMap::new(),
            ingress: Default::default(),
            fan_out: Default::default(),
            side_effect_guards: Default::default(),
            guard_kinds: Default::default(),
            capabilities: Default::default(),
            capability_steps: Default::default(),
            capability_schemas: Default::default(),
            capability_guards: Default::default(),
            capability_fan_out: Default::default(),
            schemas: HashMap::new(),
        }
    }

    /// Registry pre-loaded with the generic (non-business) node types.
    pub fn with_builtins() -> Self {
        let mut r = Self::empty();
        crate::builtins::register_builtins(&mut r);
        r
    }

    /// Register a step-node factory under `node_type`.
    pub fn register_step(&mut self, node_type: impl Into<String>, factory: StepFactory) {
        self.steps.insert(node_type.into(), factory);
    }

    /// Mark a node type as an ingress source.
    pub fn register_ingress(&mut self, node_type: impl Into<String>) {
        self.ingress.insert(node_type.into());
    }

    /// Mark a node type as fan-out-capable (its `process` may return `FanOut`).
    /// The validator (R4') bans these upstream of `execute.*`.
    pub fn register_fan_out(&mut self, node_type: impl Into<String>) {
        self.fan_out.insert(node_type.into());
    }

    /// Whether the node type may return `FanOut`.
    pub fn is_fan_out_capable(&self, node_type: &str) -> bool {
        self.fan_out.contains(node_type)
    }

    /// Declare a product node as an authorization guard for `execute.*` side effects.
    pub fn register_side_effect_guard(&mut self, node_type: impl Into<String>) {
        self.register_guard(node_type, GuardKind::Authorization);
    }

    /// Declare the role a guard plays on an action's dominating path.
    pub fn register_guard(&mut self, node_type: impl Into<String>, kind: GuardKind) {
        let node_type = node_type.into();
        self.side_effect_guards.insert(node_type.clone());
        self.guard_kinds.insert(node_type, kind);
    }

    /// Guard role of a node type, if registered as a guard.
    pub fn guard_kind(&self, node_type: &str) -> Option<GuardKind> {
        self.guard_kinds.get(node_type).copied()
    }

    /// Whether the node type is a registered guard.
    pub fn is_side_effect_guard(&self, node_type: &str) -> bool {
        self.side_effect_guards.contains(node_type)
    }

    /// Whether the node type is an ingress source.
    pub fn is_ingress(&self, node_type: &str) -> bool {
        self.ingress.contains(node_type) || node_type.starts_with("ingress.")
    }

    /// Whether the node type can be compiled as a step.
    pub fn is_step(&self, node_type: &str) -> bool {
        self.steps.contains_key(node_type)
            || self
                .capabilities
                .values()
                .any(|manifest| manifest.id == node_type)
    }

    /// Build a step node instance from a spec node's type + config.
    pub fn build_step(
        &self,
        node_type: &str,
        config: &Value,
    ) -> Result<Box<dyn StepNode>, NodeError> {
        let factory = self
            .steps
            .get(node_type)
            .ok_or_else(|| NodeError::UnknownType(node_type.to_string()))?;
        factory(config)
    }

    /// Registered step node types.
    pub fn known_step_types(&self) -> impl Iterator<Item = &str> {
        self.steps.keys().map(|s| s.as_str())
    }

    /// Attach a config schema to a node type.
    pub fn register_schema(&mut self, node_type: impl Into<String>, schema: NodeSchema) {
        self.schemas.insert(node_type.into(), schema);
    }

    /// Config schema of a node type.
    pub fn schema(&self, node_type: &str) -> Option<&NodeSchema> {
        self.schemas.get(node_type)
    }

    /// Stable node configuration schemas for authoring catalogs.
    pub fn authoring_schemas(&self) -> Vec<(String, Value)> {
        let mut entries = self
            .schemas
            .iter()
            .map(|(id, schema)| (id.clone(), schema.to_json_schema()))
            .collect::<Vec<_>>();
        entries.sort_by(|left, right| left.0.cmp(&right.0));
        entries
    }

    /// Register the immutable contract for a trigger, expression, guard or action.
    pub fn register_capability(&mut self, manifest: CapabilityManifest) -> Result<(), NodeError> {
        manifest
            .validate()
            .map_err(|error| NodeError::InvalidCapability(error.to_string()))?;
        let pin = CapabilityPin {
            id: manifest.id.clone(),
            contract_version: manifest.contract_version.clone(),
            content_digest: manifest.content_digest.clone(),
        };
        if self.capabilities.contains_key(&pin) {
            return Err(NodeError::DuplicateCapability(manifest.id));
        }
        self.capabilities.insert(pin, manifest);
        Ok(())
    }

    /// Bind executable behavior and optional authoring/guard metadata to one immutable capability pin.
    pub fn register_capability_implementation(
        &mut self,
        pin: CapabilityPin,
        factory: StepFactory,
        schema: Option<NodeSchema>,
        guard: Option<GuardKind>,
        fan_out: bool,
    ) -> Result<(), NodeError> {
        if !self.capabilities.contains_key(&pin) {
            return Err(NodeError::InvalidCapability(format!(
                "capability '{}' implementation has no registered manifest at {} ({})",
                pin.id, pin.contract_version, pin.content_digest
            )));
        }
        if self.capability_steps.contains_key(&pin) {
            return Err(NodeError::DuplicateCapability(pin.id));
        }
        self.capability_steps.insert(pin.clone(), factory);
        if let Some(schema) = schema {
            self.capability_schemas.insert(pin.clone(), schema);
        }
        if let Some(guard) = guard {
            self.capability_guards.insert(pin.clone(), guard);
        }
        if fan_out {
            self.capability_fan_out.insert(pin);
        }
        Ok(())
    }

    /// Registered manifest by capability id when exactly one version is selected.
    pub fn capability(&self, id: &str) -> Option<&CapabilityManifest> {
        let mut matches = self
            .capabilities
            .values()
            .filter(|manifest| manifest.id == id);
        let manifest = matches.next()?;
        matches.next().is_none().then_some(manifest)
    }

    /// Registered manifest by its immutable pin.
    pub fn capability_by_pin(&self, pin: &CapabilityPin) -> Option<&CapabilityManifest> {
        self.capabilities.get(pin)
    }

    /// Clone this registry with only the exact capability versions pinned by one revision.
    pub fn for_capability_pins(&self, pins: &[CapabilityPin]) -> Result<Self, NodeError> {
        let mut selected = self.clone();
        selected.capabilities.clear();
        selected
            .fan_out
            .retain(|node_type| !pins.iter().any(|pin| pin.id == *node_type));
        for pin in pins {
            let manifest = self.capability_by_pin(pin).ok_or_else(|| {
                NodeError::InvalidCapability(format!(
                    "capability '{}' is unavailable at {} ({})",
                    pin.id, pin.contract_version, pin.content_digest
                ))
            })?;
            selected.capabilities.insert(pin.clone(), manifest.clone());
            let versions = self
                .capabilities
                .keys()
                .filter(|candidate| candidate.id == pin.id)
                .count();
            if let Some(factory) = self.capability_steps.get(pin) {
                selected.steps.insert(pin.id.clone(), *factory);
            } else if versions > 1 && self.steps.contains_key(&pin.id) {
                return Err(NodeError::InvalidCapability(format!(
                    "capability '{}' has multiple versions but no executable implementation for {} ({})",
                    pin.id, pin.contract_version, pin.content_digest
                )));
            }
            if let Some(schema) = self.capability_schemas.get(pin) {
                selected.schemas.insert(pin.id.clone(), schema.clone());
            } else if versions > 1 && self.schemas.contains_key(&pin.id) {
                return Err(NodeError::InvalidCapability(format!(
                    "capability '{}' has multiple versions but no authoring schema for {} ({})",
                    pin.id, pin.contract_version, pin.content_digest
                )));
            }
            if let Some(kind) = self.capability_guards.get(pin).copied() {
                selected.side_effect_guards.insert(pin.id.clone());
                selected.guard_kinds.insert(pin.id.clone(), kind);
            } else if manifest.kind == crate::CapabilityKind::Guard
                && (versions > 1 || !self.guard_kinds.contains_key(&pin.id))
            {
                return Err(NodeError::InvalidCapability(format!(
                    "guard capability '{}' has no guard implementation for {} ({})",
                    pin.id, pin.contract_version, pin.content_digest
                )));
            }
            if self.capability_fan_out.contains(pin) {
                selected.fan_out.insert(pin.id.clone());
            }
        }
        Ok(selected)
    }

    /// Every registered manifest.
    pub fn capability_manifests(&self) -> impl Iterator<Item = &CapabilityManifest> {
        self.capabilities.values()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{CapabilityKind, Effect, IdempotencyMode};

    struct Noop;

    #[async_trait::async_trait]
    impl StepNode for Noop {
        async fn process(
            &self,
            event: &crate::Event,
            _ctx: &crate::WorkflowContext,
        ) -> crate::StepResult {
            crate::StepResult::Pass(event.clone())
        }
    }

    fn noop(_: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        Ok(Box::new(Noop))
    }

    #[test]
    fn capability_versions_are_indexed_and_selected_by_full_pin() {
        let manifest = |version: &str, digest: &str| {
            CapabilityManifest::action(
                "action.versioned",
                version,
                digest,
                Effect::ExternalWrite,
                IdempotencyMode::Native,
                true,
            )
        };
        let v1 = manifest("1", "digest-v1");
        let v2 = manifest("2", "digest-v2");
        let pin = CapabilityPin {
            id: v1.id.clone(),
            contract_version: v1.contract_version.clone(),
            content_digest: v1.content_digest.clone(),
        };
        let mut registry = NodeRegistry::empty();
        registry.register_capability(v1.clone()).unwrap();
        registry.register_capability(v2).unwrap();

        assert!(registry.capability("action.versioned").is_none());
        assert_eq!(registry.capability_by_pin(&pin), Some(&v1));
        let selected = registry.for_capability_pins(&[pin]).unwrap();
        assert_eq!(selected.capability("action.versioned"), Some(&v1));
    }

    #[test]
    fn versioned_capability_selects_exact_factory_schema_and_guard() {
        let manifest = |version: &str, digest: &str| {
            let mut manifest = CapabilityManifest::action(
                "guard.versioned",
                version,
                digest,
                Effect::Pure,
                IdempotencyMode::Native,
                true,
            );
            manifest.kind = CapabilityKind::Guard;
            manifest
        };
        let v1 = manifest("1", "digest-v1");
        let v2 = manifest("2", "digest-v2");
        let pin = CapabilityPin {
            id: v1.id.clone(),
            contract_version: v1.contract_version.clone(),
            content_digest: v1.content_digest.clone(),
        };
        let schema = NodeSchema {
            fields: vec![FieldSpec::required("approved", FieldType::Bool)],
        };
        let mut registry = NodeRegistry::empty();
        registry.register_capability(v1).unwrap();
        registry.register_capability(v2).unwrap();
        registry
            .register_capability_implementation(
                pin.clone(),
                noop,
                Some(schema.clone()),
                Some(GuardKind::Authorization),
                false,
            )
            .unwrap();

        let selected = registry.for_capability_pins(&[pin]).unwrap();
        assert_eq!(selected.schema("guard.versioned"), Some(&schema));
        assert_eq!(
            selected.guard_kind("guard.versioned"),
            Some(GuardKind::Authorization)
        );
        assert!(selected.build_step("guard.versioned", &Value::Null).is_ok());
    }

    #[test]
    fn versioned_capability_selects_fan_out_by_full_pin() {
        let manifest = |version: &str, digest: &str| {
            CapabilityManifest::action(
                "transform.versioned",
                version,
                digest,
                Effect::Pure,
                IdempotencyMode::Native,
                true,
            )
        };
        let v1 = manifest("1", "digest-v1");
        let v2 = manifest("2", "digest-v2");
        let pin = |manifest: &CapabilityManifest| CapabilityPin {
            id: manifest.id.clone(),
            contract_version: manifest.contract_version.clone(),
            content_digest: manifest.content_digest.clone(),
        };
        let mut registry = NodeRegistry::empty();
        registry.register_capability(v1.clone()).unwrap();
        registry.register_capability(v2.clone()).unwrap();
        registry
            .register_capability_implementation(pin(&v1), noop, None, None, true)
            .unwrap();
        registry
            .register_capability_implementation(pin(&v2), noop, None, None, false)
            .unwrap();

        assert!(registry
            .for_capability_pins(&[pin(&v1)])
            .unwrap()
            .is_fan_out_capable("transform.versioned"));
        assert!(!registry
            .for_capability_pins(&[pin(&v2)])
            .unwrap()
            .is_fan_out_capable("transform.versioned"));
    }
}