aimdb_core/

graph.rs

//! Dependency graph for AimDB record topology
//!
//! This module provides the `DependencyGraph` and related types that represent
//! the entire database topology — sources, links, transforms, taps, and their
//! relationships.
//!
//! The graph is constructed once during `build()` and is immutable thereafter.
//! It enables:
//! - Build-time validation (cycle detection, missing inputs)
//! - Spawn ordering (topological sort)
//! - Runtime introspection (AimX protocol, MCP tools, CLI)

use alloc::{
    string::{String, ToString},
    vec::Vec,
};

// ============================================================================
// Record Origin
// ============================================================================

/// How a record gets its values — part of the dependency graph.
#[derive(Clone, Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[cfg_attr(feature = "serde", serde(rename_all = "snake_case"))]
pub enum RecordOrigin {
    /// Autonomous producer via `.source()`
    Source,
    /// Inbound connector via `.link_from()`
    Link { protocol: String, address: String },
    /// Single-input reactive derivation via `.transform()`
    Transform { input: String },
    /// Multi-input reactive join via `.transform_join()`
    TransformJoin { inputs: Vec<String> },
    /// No registered producer (writable via `record.set` / `db.produce()`)
    Passive,
}

// ============================================================================
// Graph Node
// ============================================================================

/// Metadata for one node in the dependency graph.
#[derive(Clone, Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct GraphNode {
    /// Record key (e.g. "temp.vienna")
    pub key: String,
    /// How this record gets its values
    pub origin: RecordOrigin,
    /// Buffer type ("spmc_ring", "single_latest", "mailbox", "none")
    pub buffer_type: String,
    /// Buffer capacity (None for unbounded or non-ring buffers)
    pub buffer_capacity: Option<usize>,
    /// Number of taps attached
    pub tap_count: usize,
    /// Whether an outbound link is configured
    pub has_outbound_link: bool,
}

// ============================================================================
// Graph Edge
// ============================================================================

/// One directed edge in the dependency graph.
#[derive(Clone, Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct GraphEdge {
    /// Source record key (None for external origins like source/link)
    pub from: Option<String>,
    /// Target record key (None for side-effects like taps/link_out)
    pub to: Option<String>,
    /// Classification of this edge
    pub edge_type: EdgeType,
}

/// Classification of a dependency graph edge.
#[derive(Clone, Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
#[cfg_attr(feature = "serde", serde(rename_all = "snake_case"))]
pub enum EdgeType {
    Source,
    Link { protocol: String },
    Transform,
    TransformJoin,
    Tap { index: usize },
    LinkOut { protocol: String },
}

// ============================================================================
// Dependency Graph
// ============================================================================

/// The dependency graph, constructed once during `build()` and immutable thereafter.
#[derive(Clone, Debug)]
pub struct DependencyGraph {
    /// All nodes indexed by record key.
    pub nodes: Vec<GraphNode>,
    /// All edges (both internal and external).
    pub edges: Vec<GraphEdge>,
    /// Topological order of record keys (transforms come after their inputs).
    pub topo_order: Vec<String>,
}

impl DependencyGraph {
    /// Construct and validate the dependency graph from registered records.
    ///
    /// This builds a complete `DependencyGraph` with nodes, edges, and topological order.
    /// The graph is immutable after construction and can be used for introspection.
    ///
    /// # Arguments
    /// * `nodes` - One [`GraphNode`] per registered record (origin, buffer, etc.)
    ///
    /// # Returns
    /// * `Ok(DependencyGraph)` - The complete, validated graph
    /// * `Err(DbError::TransformInputNotFound)` - If a transform references a missing record
    /// * `Err(DbError::CyclicDependency)` - If the transform edges form a cycle
    pub fn build_and_validate(nodes: Vec<GraphNode>) -> crate::DbResult<Self> {
        use alloc::collections::VecDeque;

        // Everything in this block borrows `nodes` (&str keys); scoping it
        // lets `nodes` move into the returned graph afterwards.
        let (topo_order, edges) = {
            // Build set of all keys for validation
            let all_keys: hashbrown::HashSet<&str> =
                nodes.iter().map(|info| info.key.as_str()).collect();

            // Extract transform inputs for validation and edge construction
            let mut transform_inputs: Vec<(&str, Vec<&str>)> = Vec::new();
            for info in &nodes {
                match &info.origin {
                    RecordOrigin::Transform { input } => {
                        transform_inputs.push((info.key.as_str(), alloc::vec![input.as_str()]));
                    }
                    RecordOrigin::TransformJoin { inputs } => {
                        let input_refs: Vec<&str> = inputs.iter().map(|s| s.as_str()).collect();
                        transform_inputs.push((info.key.as_str(), input_refs));
                    }
                    _ => {}
                }
            }

            // Validate: check all transform input keys exist
            for (output_key, input_keys) in &transform_inputs {
                for input_key in input_keys {
                    if !all_keys.contains(input_key) {
                        return Err(crate::DbError::TransformInputNotFound {
                            output_key: output_key.to_string(),
                            input_key: input_key.to_string(),
                        });
                    }
                }
            }

            // Build adjacency list and in-degree map for Kahn's algorithm
            let mut in_degree: hashbrown::HashMap<&str, usize> = hashbrown::HashMap::new();
            let mut adjacency: hashbrown::HashMap<&str, Vec<&str>> = hashbrown::HashMap::new();

            // Initialize all keys with in-degree 0
            for key in &all_keys {
                in_degree.entry(*key).or_insert(0);
                adjacency.entry(*key).or_default();
            }

            // Add transform edges: input → output
            for (output_key, input_keys) in &transform_inputs {
                for input_key in input_keys {
                    adjacency.entry(*input_key).or_default().push(*output_key);
                    *in_degree.entry(*output_key).or_insert(0) += 1;
                }
            }

            // Kahn's algorithm — topological sort
            let mut queue: VecDeque<&str> = in_degree
                .iter()
                .filter(|(_, &deg)| deg == 0)
                .map(|(&node, _)| node)
                .collect();
            let mut topo_order = Vec::new();

            while let Some(node) = queue.pop_front() {
                topo_order.push(node.to_string());
                if let Some(neighbors) = adjacency.get(node) {
                    for &neighbor in neighbors {
                        if let Some(deg) = in_degree.get_mut(neighbor) {
                            *deg -= 1;
                            if *deg == 0 {
                                queue.push_back(neighbor);
                            }
                        }
                    }
                }
            }

            if topo_order.len() != all_keys.len() {
                // Find the cycle participants for a helpful error message
                let cycle_records: Vec<String> = in_degree
                    .iter()
                    .filter(|(_, &deg)| deg > 0)
                    .map(|(&k, _)| k.to_string())
                    .collect();

                return Err(crate::DbError::CyclicDependency {
                    records: cycle_records,
                });
            }

            // Build edges from origins
            let mut edges: Vec<GraphEdge> = Vec::new();
            for info in &nodes {
                match &info.origin {
                    RecordOrigin::Source => {
                        edges.push(GraphEdge {
                            from: None,
                            to: Some(info.key.clone()),
                            edge_type: EdgeType::Source,
                        });
                    }
                    RecordOrigin::Link { protocol, .. } => {
                        edges.push(GraphEdge {
                            from: None,
                            to: Some(info.key.clone()),
                            edge_type: EdgeType::Link {
                                protocol: protocol.clone(),
                            },
                        });
                    }
                    RecordOrigin::Transform { input } => {
                        edges.push(GraphEdge {
                            from: Some(input.clone()),
                            to: Some(info.key.clone()),
                            edge_type: EdgeType::Transform,
                        });
                    }
                    RecordOrigin::TransformJoin { inputs } => {
                        for input in inputs {
                            edges.push(GraphEdge {
                                from: Some(input.clone()),
                                to: Some(info.key.clone()),
                                edge_type: EdgeType::TransformJoin,
                            });
                        }
                    }
                    RecordOrigin::Passive => {
                        // No inbound edge for passive records
                    }
                }

                // Add outbound link edges if present
                if info.has_outbound_link {
                    edges.push(GraphEdge {
                        from: Some(info.key.clone()),
                        to: None,
                        edge_type: EdgeType::LinkOut {
                            protocol: "unknown".to_string(), // Protocol info not available here
                        },
                    });
                }

                // Add tap edges
                for tap_idx in 0..info.tap_count {
                    edges.push(GraphEdge {
                        from: Some(info.key.clone()),
                        to: None,
                        edge_type: EdgeType::Tap { index: tap_idx },
                    });
                }
            }

            (topo_order, edges)
        };

        Ok(DependencyGraph {
            nodes,
            edges,
            topo_order,
        })
    }

    /// Get a node by key
    pub fn node(&self, key: &str) -> Option<&GraphNode> {
        self.nodes.iter().find(|n| n.key == key)
    }

    /// Get all nodes
    pub fn nodes(&self) -> &[GraphNode] {
        &self.nodes
    }

    /// Get all edges
    pub fn edges(&self) -> &[GraphEdge] {
        &self.edges
    }

    /// Get the topological order
    pub fn topo_order(&self) -> &[String] {
        &self.topo_order
    }
}