DevFlow/crates/df-workflow/src/registry.rs

//! 节点注册表 — 根据 node_type 字符串创建节点实例

use std::collections::HashMap;

use crate::dag::Dag;
use crate::dag_def::DagDef;
use crate::node::Node;

/// 节点工厂函数类型
type NodeFactory = Box<dyn Fn(&serde_json::Value) -> Box<dyn Node> + Send + Sync>;

/// 节点注册表 — 根据 node_type 字符串创建节点实例
pub struct NodeRegistry {
    factories: HashMap<String, NodeFactory>,
}

impl NodeRegistry {
    /// 创建空的注册表
    pub fn new() -> Self {
        Self {
            factories: HashMap::new(),
        }
    }

    /// 注册一个节点工厂
    pub fn register<F>(&mut self, type_name: &str, factory: F)
    where
        F: Fn(&serde_json::Value) -> Box<dyn Node> + Send + Sync + 'static,
    {
        self.factories.insert(type_name.to_string(), Box::new(factory));
    }

    /// 根据 node_type 创建节点实例
    pub fn create(&self, type_name: &str, config: &serde_json::Value) -> anyhow::Result<Box<dyn Node>> {
        self.factories
            .get(type_name)
            .ok_or_else(|| anyhow::anyhow!("未注册的节点类型: {}", type_name))
            .map(|factory| factory(config))
    }

    /// 从 DagDef 构建完整的运行时 Dag
    pub fn build_dag(&self, def: &DagDef) -> anyhow::Result<Dag> {
        let mut dag = Dag::new();

        // 创建所有节点实例，同时收集节点 ID 集合用于边校验
        let mut node_ids: std::collections::HashSet<&String> = std::collections::HashSet::new();
        for (id, node_def) in &def.nodes {
            let node = self.create(&node_def.node_type, &node_def.config)?;
            dag.add_node(id.clone(), node);
            node_ids.insert(id);
        }

        // 添加所有边 — 校验两端节点均存在，野边早返回
        for edge_def in &def.edges {
            if !node_ids.contains(&edge_def.source) || !node_ids.contains(&edge_def.target) {
                anyhow::bail!(
                    "边 source/target 节点不存在: {} -> {}",
                    edge_def.source,
                    edge_def.target
                );
            }
            match &edge_def.condition {
                Some(cond) => dag.add_edge_with_condition(
                    edge_def.source.clone(),
                    edge_def.target.clone(),
                    cond.clone(),
                ),
                None => dag.add_edge(edge_def.source.clone(), edge_def.target.clone()),
            }
        }

        Ok(dag)
    }
}

// 不实现 Default:原 Default 注册了一个会 panic 的 script 工厂(unimplemented!),
// 违反项目铁律「无 panic——所有占位代码返回空/默认值」。所有调用方应显式 new() +
// 手动 register 真实节点(如 state.rs::build_registry 的做法)。