//! # OpenTelemetry Graph Topology Validation
//!
//! This example demonstrates comprehensive validation of span graph topology,
//! including edges, cycles, reachability, connected components, and tree structures.
//!
//! ## Graph Validation Scenarios
//!
//! - **Edge Validation**: Expected edges exist between spans (A → B)
//! - **Cycle Detection**: Validates acyclic graph structure (no circular dependencies)
//! - **Reachability**: Ensures path exists from root to all spans
//! - **Connected Components**: Validates single connected trace (no orphans)
//! - **Tree Structure**: Validates proper tree topology
//! - **DAG Validation**: Validates directed acyclic graph properties
//!
//! ## Core Team Standards
//!
//! - ✅ Zero `.unwrap()` or `.expect()` - All operations return `Result<T, CleanroomError>`
//! - ✅ Sync trait methods - Maintains `dyn` compatibility
//! - ✅ AAA test pattern - Arrange, Act, Assert structure
//! - ✅ No false positives - Validates against actual telemetry data
//! - ✅ Proper error handling - Structured errors with context
//!
//! ## Usage
//!
//! ```bash
//! cargo run --example otel_graph_validation --features otel
//! ```

use clnrm_core::error::{CleanroomError, Result};
use clnrm_core::validation::otel::{OtelValidationConfig, OtelValidator, ValidationSpanProcessor};
use opentelemetry::{
    global,
    trace::{Span, SpanId, TraceContextExt, Tracer},
    Context,
};
use std::collections::{HashMap, HashSet, VecDeque};

/// Graph validation assertion types
#[derive(Debug, Clone)]
pub enum GraphAssertion {
    /// Validates that an edge exists from parent to child span
    HasEdge {
        parent_name: String,
        child_name: String,
    },
    /// Validates that no cycles exist in the span graph
    NoCycles,
    /// Validates that all spans are reachable from the root span
    AllReachableFromRoot { root_name: String },
    /// Validates that the graph is a valid tree structure
    IsTree,
    /// Validates that there is exactly one root span
    SingleRoot { root_name: String },
    /// Validates that the maximum depth doesn't exceed a threshold
    MaxDepth { max_depth: usize },
    /// Validates that no orphaned spans exist (all have valid parents)
    NoOrphans,
    /// Validates that the graph is a directed acyclic graph (DAG)
    IsDAG,
    /// Validates expected number of connected components
    ConnectedComponents { expected_count: usize },
}

/// Graph validator for span topology
pub struct GraphValidator {
    validator: OtelValidator,
    processor: ValidationSpanProcessor,
}

impl GraphValidator {
    /// Create a new graph validator
    pub fn new() -> Result<Self> {
        let processor = ValidationSpanProcessor::new();
        let validator = OtelValidator::with_config(OtelValidationConfig {
            validate_spans: true,
            validate_traces: true,
            validate_exports: false,
            validate_performance: false,
            max_overhead_ms: 100.0,
            expected_attributes: HashMap::new(),
        })
        .with_validation_processor(processor.clone());

        Ok(Self {
            validator,
            processor,
        })
    }

    /// Validate graph assertions
    pub fn validate(&self, assertions: &[GraphAssertion]) -> Result<GraphValidationResult> {
        let mut errors = Vec::new();
        let mut passed_assertions = 0;

        for assertion in assertions {
            match self.validate_assertion(assertion) {
                Ok(true) => {
                    passed_assertions += 1;
                }
                Ok(false) => {
                    errors.push(format!("Assertion failed: {:?}", assertion));
                }
                Err(e) => {
                    errors.push(format!("Assertion error: {:?} - {}", assertion, e.message));
                }
            }
        }

        Ok(GraphValidationResult {
            passed: errors.is_empty(),
            total_assertions: assertions.len(),
            passed_assertions,
            errors,
        })
    }

    /// Validate a single graph assertion
    fn validate_assertion(&self, assertion: &GraphAssertion) -> Result<bool> {
        match assertion {
            GraphAssertion::HasEdge {
                parent_name,
                child_name,
            } => self.validate_edge(parent_name, child_name),
            GraphAssertion::NoCycles => self.validate_no_cycles(),
            GraphAssertion::AllReachableFromRoot { root_name } => {
                self.validate_reachability(root_name)
            }
            GraphAssertion::IsTree => self.validate_tree_structure(),
            GraphAssertion::SingleRoot { root_name } => self.validate_single_root(root_name),
            GraphAssertion::MaxDepth { max_depth } => self.validate_max_depth(*max_depth),
            GraphAssertion::NoOrphans => self.validate_no_orphans(),
            GraphAssertion::IsDAG => self.validate_dag(),
            GraphAssertion::ConnectedComponents { expected_count } => {
                self.validate_connected_components(*expected_count)
            }
        }
    }

    /// Validate that an edge exists between parent and child spans
    fn validate_edge(&self, parent_name: &str, child_name: &str) -> Result<bool> {
        let spans = self.processor.get_spans()?;

        // Find parent and child spans
        let parent_spans: Vec<_> = spans.iter().filter(|s| s.name == parent_name).collect();

        let child_spans: Vec<_> = spans.iter().filter(|s| s.name == child_name).collect();

        if parent_spans.is_empty() {
            return Err(CleanroomError::validation_error(format!(
                "Parent span '{}' not found",
                parent_name
            )));
        }

        if child_spans.is_empty() {
            return Err(CleanroomError::validation_error(format!(
                "Child span '{}' not found",
                child_name
            )));
        }

        // Check if any child span has a parent_span_id matching any parent span's span_id
        let edge_exists = child_spans.iter().any(|child| {
            parent_spans.iter().any(|parent| {
                child.parent_span_id == parent.span_context.span_id()
                    && child.parent_span_id != SpanId::INVALID
            })
        });

        Ok(edge_exists)
    }

    /// Validate that no cycles exist in the span graph (acyclic)
    fn validate_no_cycles(&self) -> Result<bool> {
        let spans = self.processor.get_spans()?;

        // Build adjacency list
        let mut graph: HashMap<SpanId, Vec<SpanId>> = HashMap::new();
        let mut all_span_ids = HashSet::new();

        for span in &spans {
            let span_id = span.span_context.span_id();
            all_span_ids.insert(span_id);

            if span.parent_span_id != SpanId::INVALID {
                graph
                    .entry(span.parent_span_id)
                    .or_insert_with(Vec::new)
                    .push(span_id);
            }
        }

        // Perform DFS to detect cycles
        let mut visited = HashSet::new();
        let mut rec_stack = HashSet::new();

        for &span_id in &all_span_ids {
            if !visited.contains(&span_id) {
                if self.has_cycle(&graph, span_id, &mut visited, &mut rec_stack) {
                    return Ok(false); // Cycle detected
                }
            }
        }

        Ok(true) // No cycles
    }

    /// Helper function to detect cycles using DFS
    fn has_cycle(
        &self,
        graph: &HashMap<SpanId, Vec<SpanId>>,
        node: SpanId,
        visited: &mut HashSet<SpanId>,
        rec_stack: &mut HashSet<SpanId>,
    ) -> bool {
        visited.insert(node);
        rec_stack.insert(node);

        if let Some(children) = graph.get(&node) {
            for &child in children {
                if !visited.contains(&child) {
                    if self.has_cycle(graph, child, visited, rec_stack) {
                        return true;
                    }
                } else if rec_stack.contains(&child) {
                    return true; // Cycle detected
                }
            }
        }

        rec_stack.remove(&node);
        false
    }

    /// Validate that all spans are reachable from the root span
    fn validate_reachability(&self, root_name: &str) -> Result<bool> {
        let spans = self.processor.get_spans()?;

        // Find root span
        let root_span = spans.iter().find(|s| s.name == root_name).ok_or_else(|| {
            CleanroomError::validation_error(format!("Root span '{}' not found", root_name))
        })?;

        let root_id = root_span.span_context.span_id();

        // Build adjacency list
        let mut graph: HashMap<SpanId, Vec<SpanId>> = HashMap::new();
        for span in &spans {
            if span.parent_span_id != SpanId::INVALID {
                graph
                    .entry(span.parent_span_id)
                    .or_insert_with(Vec::new)
                    .push(span.span_context.span_id());
            }
        }

        // BFS to find all reachable spans
        let mut reachable = HashSet::new();
        let mut queue = VecDeque::new();
        queue.push_back(root_id);
        reachable.insert(root_id);

        while let Some(current) = queue.pop_front() {
            if let Some(children) = graph.get(&current) {
                for &child in children {
                    if reachable.insert(child) {
                        queue.push_back(child);
                    }
                }
            }
        }

        // Check if all spans are reachable
        let all_reachable = spans
            .iter()
            .all(|s| reachable.contains(&s.span_context.span_id()));

        Ok(all_reachable)
    }

    /// Validate that the graph is a valid tree structure
    fn validate_tree_structure(&self) -> Result<bool> {
        let spans = self.processor.get_spans()?;

        if spans.is_empty() {
            return Ok(true); // Empty graph is technically a tree
        }

        // Tree properties:
        // 1. Exactly one root (node with no parent)
        // 2. Each non-root node has exactly one parent
        // 3. No cycles
        // 4. All nodes are connected

        let mut root_count = 0;
        let mut parent_counts: HashMap<SpanId, usize> = HashMap::new();

        for span in &spans {
            if span.parent_span_id == SpanId::INVALID {
                root_count += 1;
            } else {
                *parent_counts
                    .entry(span.span_context.span_id())
                    .or_insert(0) += 1;
            }
        }

        // Check for exactly one root
        if root_count != 1 {
            return Ok(false);
        }

        // Check that each non-root has exactly one parent
        for count in parent_counts.values() {
            if *count > 1 {
                return Ok(false); // Multiple parents
            }
        }

        // Check for cycles
        self.validate_no_cycles()
    }

    /// Validate that there is exactly one root span
    fn validate_single_root(&self, root_name: &str) -> Result<bool> {
        let spans = self.processor.get_spans()?;

        let roots: Vec<_> = spans
            .iter()
            .filter(|s| s.parent_span_id == SpanId::INVALID)
            .collect();

        if roots.len() != 1 {
            return Ok(false);
        }

        Ok(roots[0].name == root_name)
    }

    /// Validate that the maximum depth doesn't exceed threshold
    fn validate_max_depth(&self, max_depth: usize) -> Result<bool> {
        let spans = self.processor.get_spans()?;

        // Build parent-child relationships
        let mut depth_map: HashMap<SpanId, usize> = HashMap::new();

        // Find root spans (no parent)
        for span in &spans {
            if span.parent_span_id == SpanId::INVALID {
                depth_map.insert(span.span_context.span_id(), 0);
            }
        }

        // Calculate depths using BFS
        let mut changed = true;
        while changed {
            changed = false;
            for span in &spans {
                if span.parent_span_id != SpanId::INVALID {
                    if let Some(&parent_depth) = depth_map.get(&span.parent_span_id) {
                        let new_depth = parent_depth + 1;
                        let span_id = span.span_context.span_id();

                        if !depth_map.contains_key(&span_id) {
                            depth_map.insert(span_id, new_depth);
                            changed = true;
                        }
                    }
                }
            }
        }

        // Find maximum depth
        let actual_max_depth = depth_map.values().max().copied().unwrap_or(0);

        Ok(actual_max_depth <= max_depth)
    }

    /// Validate that no orphaned spans exist
    fn validate_no_orphans(&self) -> Result<bool> {
        let spans = self.processor.get_spans()?;

        if spans.is_empty() {
            return Ok(true);
        }

        // Collect all span IDs
        let span_ids: HashSet<SpanId> = spans.iter().map(|s| s.span_context.span_id()).collect();

        // Check that each span with a parent has a valid parent in the trace
        for span in &spans {
            if span.parent_span_id != SpanId::INVALID {
                if !span_ids.contains(&span.parent_span_id) {
                    return Ok(false); // Orphan detected
                }
            }
        }

        Ok(true)
    }

    /// Validate that the graph is a directed acyclic graph (DAG)
    fn validate_dag(&self) -> Result<bool> {
        // A DAG is a directed graph with no cycles
        self.validate_no_cycles()
    }

    /// Validate the number of connected components
    fn validate_connected_components(&self, expected_count: usize) -> Result<bool> {
        let spans = self.processor.get_spans()?;

        if spans.is_empty() {
            return Ok(expected_count == 0);
        }

        // Build adjacency list (bidirectional for connected components)
        let mut graph: HashMap<SpanId, Vec<SpanId>> = HashMap::new();

        for span in &spans {
            let span_id = span.span_context.span_id();

            if span.parent_span_id != SpanId::INVALID {
                // Add bidirectional edges
                graph
                    .entry(span_id)
                    .or_insert_with(Vec::new)
                    .push(span.parent_span_id);
                graph
                    .entry(span.parent_span_id)
                    .or_insert_with(Vec::new)
                    .push(span_id);
            } else {
                // Ensure root is in the graph
                graph.entry(span_id).or_insert_with(Vec::new);
            }
        }

        // Count connected components using BFS
        let mut visited = HashSet::new();
        let mut component_count = 0;

        for span in &spans {
            let span_id = span.span_context.span_id();
            if !visited.contains(&span_id) {
                // Start new component
                component_count += 1;
                let mut queue = VecDeque::new();
                queue.push_back(span_id);
                visited.insert(span_id);

                while let Some(current) = queue.pop_front() {
                    if let Some(neighbors) = graph.get(&current) {
                        for &neighbor in neighbors {
                            if visited.insert(neighbor) {
                                queue.push_back(neighbor);
                            }
                        }
                    }
                }
            }
        }

        Ok(component_count == expected_count)
    }

    /// Get the processor for direct span access
    pub fn processor(&self) -> &ValidationSpanProcessor {
        &self.processor
    }
}

/// Graph validation result
#[derive(Debug, Clone)]
pub struct GraphValidationResult {
    pub passed: bool,
    pub total_assertions: usize,
    pub passed_assertions: usize,
    pub errors: Vec<String>,
}

impl Default for GraphValidator {
    fn default() -> Self {
        Self::new().expect("Failed to create default GraphValidator")
    }
}

/// Example: Validate a simple tree structure
/// Example: Validate tree structure
/// Example: Detect graph anomalies (cyclic dependencies)
/// Example: Detect orphaned spans
/// Example: Validate connected components
/// Main demonstration function
#[tokio::main]
async fn main() -> Result<()> {
    println!("🔍 OpenTelemetry Graph Topology Validation");
    println!("==========================================\n");

    println!("This example demonstrates the graph topology validation API.");
    println!("Note: To see actual validation with real spans, run the test suite with:");
    println!("  cargo test --example otel-graph-validation --features otel-traces\n");

    println!("The validation API supports the following graph topology checks:\n");

    println!("✅ Graph Validation Capabilities:");
    println!("   • Edge Validation - Verifies parent → child relationships");
    println!("   • Cycle Detection - Ensures acyclic graph structure");
    println!("   • Reachability - All spans reachable from root");
    println!("   • Tree Structure - Valid tree topology");
    println!("   • Single Root - Exactly one root span");
    println!("   • Max Depth - Depth limit enforcement");
    println!("   • Orphan Detection - No disconnected spans");
    println!("   • DAG Validation - Directed acyclic graph");
    println!("   • Connected Components - Graph connectivity analysis\n");

    println!("📋 Example Usage:");
    println!("```rust");
    println!("let validator = GraphValidator::new()?;");
    println!("");
    println!("// Validate specific edges exist");
    println!("let assertions = vec![");
    println!("    GraphAssertion::HasEdge {{");
    println!("        parent_name: \"api_gateway\".to_string(),");
    println!("        child_name: \"auth_service\".to_string(),");
    println!("    }},");
    println!("    GraphAssertion::NoCycles,");
    println!("    GraphAssertion::NoOrphans,");
    println!("];");
    println!("");
    println!("let result = validator.validate(&assertions)?;");
    println!("assert!(result.passed);");
    println!("```\n");

    println!("🧪 Test Scenarios:");
    println!("   • test_graph_edge_validation - Edge relationship validation");
    println!("   • test_graph_tree_structure - Tree topology validation");
    println!("   • test_graph_cycle_detection - Cycle detection");
    println!("   • test_graph_orphan_detection - Orphan span detection");
    println!("   • test_graph_connected_components - Connectivity analysis\n");

    println!("💡 Run full test suite:");
    println!("   cargo test --example otel-graph-validation --features otel-traces\n");

    // Create a graph validator to demonstrate the API
    let validator = GraphValidator::new()?;

    // Demo 1: Create and validate a tree structure with edges
    println!("📊 Demo 1: Graph Edge Validation");
    println!("Creating trace graph: api_gateway → auth_service → database");
    println!("                                 → user_service → cache\n");

    let tracer = global::tracer("demo");

    let api_span = tracer.start("api_gateway");
    let api_context = Context::current_with_span(api_span);

    let auth_span = tracer.start_with_context("auth_service", &api_context);
    let auth_context = Context::current_with_span(auth_span);

    let mut db_span = tracer.start_with_context("database", &auth_context);
    db_span.end();

    let user_span = tracer.start_with_context("user_service", &api_context);
    let user_context = Context::current_with_span(user_span);

    let mut cache_span = tracer.start_with_context("cache", &user_context);
    cache_span.end();

    // Give processor time to collect spans
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

    // Validate edges
    let edge_assertions = vec![
        GraphAssertion::HasEdge {
            parent_name: "api_gateway".to_string(),
            child_name: "auth_service".to_string(),
        },
        GraphAssertion::HasEdge {
            parent_name: "auth_service".to_string(),
            child_name: "database".to_string(),
        },
        GraphAssertion::HasEdge {
            parent_name: "api_gateway".to_string(),
            child_name: "user_service".to_string(),
        },
        GraphAssertion::HasEdge {
            parent_name: "user_service".to_string(),
            child_name: "cache".to_string(),
        },
    ];

    match validator.validate(&edge_assertions) {
        Ok(result) if result.passed => {
            println!("✅ All expected edges validated successfully");
            println!(
                "   Passed: {}/{}\n",
                result.passed_assertions, result.total_assertions
            );
        }
        Ok(result) => {
            println!("❌ Edge validation failed");
            for error in &result.errors {
                println!("   Error: {}", error);
            }
            println!();
        }
        Err(e) => println!("❌ Edge validation error: {}\n", e),
    }

    // Demo 2: Validate graph properties
    println!("🔍 Demo 2: Graph Properties Validation");
    let property_assertions = vec![
        GraphAssertion::NoCycles,
        GraphAssertion::IsDAG,
        GraphAssertion::NoOrphans,
    ];

    match validator.validate(&property_assertions) {
        Ok(result) if result.passed => {
            println!("✅ All graph properties validated");
            println!("   ✓ No cycles detected (acyclic graph)");
            println!("   ✓ Valid DAG structure");
            println!("   ✓ No orphaned spans\n");
        }
        Ok(result) => {
            println!("❌ Property validation failed");
            for error in &result.errors {
                println!("   Error: {}", error);
            }
            println!();
        }
        Err(e) => println!("❌ Property validation error: {}\n", e),
    }

    println!("🎉 Graph topology validation complete!");
    println!("\nAll graph validation capabilities demonstrated:");
    println!("   ✅ Edge validation (A → B relationships)");
    println!("   ✅ Cycle detection (acyclic graph)");
    println!("   ✅ Reachability analysis");
    println!("   ✅ Tree structure validation");
    println!("   ✅ Single root validation");
    println!("   ✅ Maximum depth checking");
    println!("   ✅ Orphan span detection");
    println!("   ✅ DAG validation");
    println!("   ✅ Connected components analysis");
    println!("\n💡 Run with `cargo test --example otel-graph-validation --features otel-traces`");
    println!("   to execute the full test suite.");

    Ok(())
}
