af_workflow/

registry.rs

//! 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::HashMap;

use serde_json::Value;

use crate::node::StepNode;
use crate::{CapabilityManifest, 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)]
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)]
pub struct FieldSpec {
    /// Field name.
    pub key: String,
    /// Expected type.
    pub ty: FieldType,
    /// Whether it must be present.
    pub required: bool,
}

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)]
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)]
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: HashMap<String, CapabilityManifest>,
    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(),
            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)
    }

    /// 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)
    }

    /// 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()))?;
        if self.capabilities.contains_key(&manifest.id) {
            return Err(NodeError::DuplicateCapability(manifest.id));
        }
        self.capabilities.insert(manifest.id.clone(), manifest);
        Ok(())
    }

    /// Registered manifest by capability id.
    pub fn capability(&self, id: &str) -> Option<&CapabilityManifest> {
        self.capabilities.get(id)
    }

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