oxirs_shacl/validation/

multi_graph.rs

//! Multi-graph SHACL validation for federated and distributed RDF datasets
//!
//! This module provides validation capabilities across multiple named graphs,
//! supporting federated queries, graph-scoped constraints, and distributed validation.

#[cfg(feature = "async")]
use std::collections::{HashMap, HashSet};
#[cfg(feature = "async")]
use std::sync::{Arc, RwLock};
#[cfg(feature = "async")]
use std::time::{Duration, Instant};

#[cfg(feature = "async")]
use anyhow::Result;
#[cfg(feature = "async")]
use indexmap::IndexMap;
#[cfg(feature = "async")]
use serde::{Deserialize, Serialize};
#[cfg(feature = "async")]
use tokio::sync::Semaphore;

#[cfg(feature = "async")]
use oxirs_core::{
    model::{GraphName, Term, Triple},
    Store,
};

#[cfg(test)]
use oxirs_core::ConcreteStore;

#[cfg(feature = "async")]
use crate::{
    targets::*, validation::ValidationEngine, Shape, ShapeId, ValidationConfig, ValidationReport,
};

/// Configuration for multi-graph validation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MultiGraphValidationConfig {
    /// Base validation configuration
    pub base_config: ValidationConfig,

    /// Graph selection strategy
    pub graph_strategy: GraphSelectionStrategy,

    /// Maximum number of graphs to validate concurrently
    pub max_concurrent_graphs: usize,

    /// Timeout for individual graph validation
    pub graph_timeout: Duration,

    /// Whether to enable cross-graph constraint validation
    pub enable_cross_graph_constraints: bool,

    /// Whether to aggregate results across graphs
    pub aggregate_results: bool,

    /// Graph priority weights for resource allocation
    pub graph_priorities: HashMap<GraphName, f64>,

    /// Remote endpoint configurations for federated validation
    pub remote_endpoints: HashMap<GraphName, RemoteEndpointConfig>,

    /// Cache configuration for cross-graph queries
    pub cross_graph_cache_size: usize,

    /// Whether to enable partial validation on errors
    pub enable_partial_validation: bool,
}

/// Strategy for selecting graphs to validate
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum GraphSelectionStrategy {
    /// Validate all available graphs
    All,
    /// Validate only specified graphs
    Explicit(HashSet<GraphName>),
    /// Validate graphs matching a pattern
    Pattern(String),
    /// Validate based on shape targets
    TargetBased,
    /// Custom selection function
    Custom,
}

/// Configuration for remote SPARQL endpoints
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RemoteEndpointConfig {
    /// Endpoint URL
    pub url: String,

    /// Authentication credentials
    pub auth: Option<EndpointAuth>,

    /// Request timeout
    pub timeout: Duration,

    /// Maximum concurrent requests
    pub max_concurrent_requests: usize,

    /// Retry configuration
    pub retry_config: RetryConfig,

    /// Whether the endpoint supports SHACL-SPARQL
    pub supports_shacl_sparql: bool,
}

/// Authentication configuration for remote endpoints
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EndpointAuth {
    pub username: String,
    pub password: String,
    pub auth_type: AuthType,
}

/// Authentication types
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum AuthType {
    Basic,
    Bearer,
    ApiKey,
}

/// Retry configuration for endpoint requests
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RetryConfig {
    pub max_retries: usize,
    pub base_delay: Duration,
    pub max_delay: Duration,
    pub backoff_factor: f64,
}

/// Result of multi-graph validation
#[derive(Debug, Clone)]
pub struct MultiGraphValidationResult {
    /// Individual graph validation reports
    pub graph_reports: HashMap<GraphName, ValidationReport>,

    /// Aggregated validation report
    pub aggregated_report: Option<ValidationReport>,

    /// Cross-graph constraint violations
    pub cross_graph_violations: Vec<CrossGraphViolation>,

    /// Validation statistics per graph
    pub graph_stats: HashMap<GraphName, GraphValidationStats>,

    /// Overall validation statistics
    pub overall_stats: MultiGraphStats,

    /// Validation errors per graph
    pub graph_errors: HashMap<GraphName, Vec<String>>,

    /// Federated query statistics
    pub federation_stats: Option<FederationStats>,
}

/// Violation that spans multiple graphs
#[derive(Debug, Clone)]
pub struct CrossGraphViolation {
    /// Primary focus node
    pub focus_node: Term,

    /// Constraint that was violated
    pub constraint_component_id: crate::ConstraintComponentId,

    /// Graphs involved in the violation
    pub involved_graphs: HashSet<GraphName>,

    /// Cross-graph evidence
    pub evidence: CrossGraphEvidence,

    /// Violation message
    pub message: String,

    /// Severity level
    pub severity: crate::Severity,
}

/// Evidence for cross-graph violations
#[derive(Debug, Clone)]
pub struct CrossGraphEvidence {
    /// Triples from different graphs that contribute to the violation
    pub contributing_triples: HashMap<GraphName, Vec<Triple>>,

    /// Query results from federated queries
    pub federated_results: Vec<QueryResult>,
}

/// Query result for federated queries
#[derive(Debug, Clone)]
pub struct QueryResult {
    /// Source graph or endpoint
    pub source: GraphName,

    /// SPARQL query that was executed
    pub query: String,

    /// Query results
    pub bindings: Vec<HashMap<String, Term>>,

    /// Query execution time
    pub execution_time: Duration,
}

/// Validation statistics for individual graphs
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GraphValidationStats {
    /// Graph identifier
    pub graph_name: GraphName,

    /// Number of triples validated
    pub triples_validated: usize,

    /// Number of shapes applied
    pub shapes_applied: usize,

    /// Number of constraints evaluated
    pub constraints_evaluated: usize,

    /// Validation duration
    pub validation_duration: Duration,

    /// Memory used for validation
    pub memory_used_bytes: usize,

    /// Number of violations found
    pub violations_found: usize,

    /// Whether validation completed successfully
    pub validation_successful: bool,

    /// Remote endpoint statistics (if applicable)
    pub remote_stats: Option<RemoteGraphStats>,
}

/// Statistics for remote graph validation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RemoteGraphStats {
    /// Endpoint URL
    pub endpoint_url: String,

    /// Number of SPARQL queries executed
    pub queries_executed: usize,

    /// Total network time
    pub network_time: Duration,

    /// Number of retry attempts
    pub retry_attempts: usize,

    /// Data transfer statistics
    pub bytes_transferred: usize,
}

/// Overall multi-graph validation statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MultiGraphStats {
    /// Total number of graphs validated
    pub total_graphs: usize,

    /// Number of successfully validated graphs
    pub successful_graphs: usize,

    /// Number of failed graph validations
    pub failed_graphs: usize,

    /// Total validation time
    pub total_validation_time: Duration,

    /// Maximum concurrent graphs validated
    pub max_concurrent_graphs: usize,

    /// Cross-graph constraint evaluations
    pub cross_graph_evaluations: usize,

    /// Total violations across all graphs
    pub total_violations: usize,

    /// Average validation time per graph
    pub average_validation_time: Duration,

    /// Resource utilization metrics
    pub resource_utilization: ResourceUtilization,
}

/// Resource utilization metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceUtilization {
    /// Peak memory usage
    pub peak_memory_bytes: usize,

    /// Average CPU utilization
    pub average_cpu_percent: f64,

    /// Network bandwidth used
    pub network_bandwidth_mbps: f64,

    /// Disk I/O operations
    pub disk_io_operations: usize,
}

/// Statistics for federated query execution
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FederationStats {
    /// Number of federated queries executed
    pub federated_queries: usize,

    /// Total federation overhead
    pub federation_overhead: Duration,

    /// Network latency statistics
    pub network_latency: NetworkLatencyStats,

    /// Data locality statistics
    pub data_locality: DataLocalityStats,
}

/// Network latency statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NetworkLatencyStats {
    /// Minimum latency observed
    pub min_latency: Duration,

    /// Maximum latency observed
    pub max_latency: Duration,

    /// Average latency
    pub average_latency: Duration,

    /// 95th percentile latency
    pub p95_latency: Duration,
}

/// Data locality statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DataLocalityStats {
    /// Percentage of data accessed locally
    pub local_data_percent: f64,

    /// Percentage of data accessed remotely
    pub remote_data_percent: f64,

    /// Cache hit ratio for remote data
    pub remote_cache_hit_ratio: f64,
}

impl Default for MultiGraphValidationConfig {
    fn default() -> Self {
        Self {
            base_config: ValidationConfig::default(),
            graph_strategy: GraphSelectionStrategy::All,
            max_concurrent_graphs: 4,
            graph_timeout: Duration::from_secs(30),
            enable_cross_graph_constraints: false,
            aggregate_results: true,
            graph_priorities: HashMap::new(),
            remote_endpoints: HashMap::new(),
            cross_graph_cache_size: 1000,
            enable_partial_validation: true,
        }
    }
}

impl Default for RetryConfig {
    fn default() -> Self {
        Self {
            max_retries: 3,
            base_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(5),
            backoff_factor: 2.0,
        }
    }
}

/// Multi-graph SHACL validation engine
pub struct MultiGraphValidationEngine {
    /// Validation configuration
    config: MultiGraphValidationConfig,

    /// SHACL shapes for validation
    shapes: Arc<RwLock<IndexMap<ShapeId, Shape>>>,

    /// Graph-specific shape mappings
    #[allow(dead_code)]
    graph_shape_mappings: HashMap<GraphName, HashSet<ShapeId>>,

    /// Cross-graph constraint cache
    #[allow(dead_code)]
    cross_graph_cache: Arc<RwLock<HashMap<String, CrossGraphCacheEntry>>>,

    /// Remote endpoint clients
    #[allow(dead_code)]
    remote_clients: HashMap<GraphName, RemoteGraphClient>,

    /// Validation statistics
    #[allow(dead_code)]
    stats: Arc<RwLock<MultiGraphStats>>,
}

/// Cache entry for cross-graph queries
#[derive(Debug, Clone)]
#[allow(dead_code)]
struct CrossGraphCacheEntry {
    /// Cached query results
    results: Vec<QueryResult>,

    /// Cache timestamp
    timestamp: Instant,

    /// Cache expiry time
    expiry: Duration,
}

/// Client for remote graph validation
#[allow(dead_code)]
struct RemoteGraphClient {
    /// Endpoint configuration
    config: RemoteEndpointConfig,

    /// HTTP client
    client: reqwest::Client,

    /// Semaphore for concurrent request limiting
    semaphore: Arc<Semaphore>,
}

impl MultiGraphValidationEngine {
    /// Create a new multi-graph validation engine
    pub fn new(
        shapes: IndexMap<ShapeId, Shape>,
        config: MultiGraphValidationConfig,
    ) -> Result<Self> {
        let shapes = Arc::new(RwLock::new(shapes));
        let cross_graph_cache = Arc::new(RwLock::new(HashMap::new()));
        let stats = Arc::new(RwLock::new(MultiGraphStats::new()));

        // Initialize remote clients
        let mut remote_clients = HashMap::new();
        for (graph_name, endpoint_config) in &config.remote_endpoints {
            let client = RemoteGraphClient::new(endpoint_config.clone())?;
            remote_clients.insert(graph_name.clone(), client);
        }

        Ok(Self {
            config,
            shapes,
            graph_shape_mappings: HashMap::new(),
            cross_graph_cache,
            remote_clients,
            stats,
        })
    }

    /// Validate multiple graphs with the configured strategy
    pub async fn validate_multi_graph(
        &self,
        stores: &HashMap<GraphName, Arc<dyn Store>>,
    ) -> Result<MultiGraphValidationResult> {
        let start_time = Instant::now();

        // Select graphs to validate based on strategy
        let selected_graphs = self.select_graphs_for_validation(stores).await?;

        // Validate graphs concurrently
        let graph_results = self
            .validate_graphs_concurrent(&selected_graphs, stores)
            .await?;

        // Evaluate cross-graph constraints if enabled
        let cross_graph_violations = if self.config.enable_cross_graph_constraints {
            self.evaluate_cross_graph_constraints(stores, &graph_results)
                .await?
        } else {
            Vec::new()
        };

        // Aggregate results if requested
        let aggregated_report = if self.config.aggregate_results {
            Some(self.aggregate_validation_reports(&graph_results)?)
        } else {
            None
        };

        // Calculate statistics
        let overall_stats = self.calculate_overall_stats(&graph_results, start_time)?;

        Ok(MultiGraphValidationResult {
            graph_reports: graph_results
                .iter()
                .map(|(g, r)| (g.clone(), r.report.clone()))
                .collect(),
            aggregated_report,
            cross_graph_violations,
            graph_stats: graph_results
                .iter()
                .map(|(g, r)| (g.clone(), r.stats.clone()))
                .collect(),
            overall_stats,
            graph_errors: self.collect_graph_errors(&graph_results),
            federation_stats: self.calculate_federation_stats(&graph_results).await,
        })
    }

    /// Select graphs for validation based on the configured strategy
    async fn select_graphs_for_validation(
        &self,
        stores: &HashMap<GraphName, Arc<dyn Store>>,
    ) -> Result<HashSet<GraphName>> {
        match &self.config.graph_strategy {
            GraphSelectionStrategy::All => Ok(stores.keys().cloned().collect()),
            GraphSelectionStrategy::Explicit(graphs) => Ok(graphs
                .intersection(&stores.keys().cloned().collect())
                .cloned()
                .collect()),
            GraphSelectionStrategy::Pattern(pattern) => {
                // Simple pattern matching on graph names
                let regex = regex::Regex::new(pattern)?;
                Ok(stores
                    .keys()
                    .filter(|graph_name| {
                        if let GraphName::NamedNode(named) = graph_name {
                            regex.is_match(named.as_str())
                        } else {
                            false
                        }
                    })
                    .cloned()
                    .collect())
            }
            GraphSelectionStrategy::TargetBased => {
                // Select graphs based on shape targets
                self.select_target_based_graphs(stores).await
            }
            GraphSelectionStrategy::Custom => {
                // For now, select all graphs (custom logic would be pluggable)
                Ok(stores.keys().cloned().collect())
            }
        }
    }

    /// Select graphs based on shape targets
    async fn select_target_based_graphs(
        &self,
        stores: &HashMap<GraphName, Arc<dyn Store>>,
    ) -> Result<HashSet<GraphName>> {
        let shapes = self
            .shapes
            .read()
            .expect("read lock should not be poisoned");
        let mut selected_graphs = HashSet::new();

        for (graph_name, store) in stores {
            for shape in shapes.values() {
                // Check if any targets in this shape have focus nodes in this graph
                let mut target_selector = TargetSelector::new();
                let mut has_focus_nodes = false;

                for target in &shape.targets {
                    let focus_nodes = target_selector.select_targets(
                        store.as_ref(),
                        target,
                        Some(graph_name.as_ref()),
                    )?;
                    if !focus_nodes.is_empty() {
                        has_focus_nodes = true;
                        break;
                    }
                }

                if has_focus_nodes {
                    selected_graphs.insert(graph_name.clone());
                    break;
                }
            }
        }

        Ok(selected_graphs)
    }

    /// Validate multiple graphs concurrently
    async fn validate_graphs_concurrent(
        &self,
        selected_graphs: &HashSet<GraphName>,
        stores: &HashMap<GraphName, Arc<dyn Store>>,
    ) -> Result<HashMap<GraphName, GraphValidationResult>> {
        let semaphore = Arc::new(Semaphore::new(self.config.max_concurrent_graphs));
        let mut tasks = Vec::new();

        for graph_name in selected_graphs {
            if let Some(store) = stores.get(graph_name) {
                let semaphore = Arc::clone(&semaphore);
                let graph_name = graph_name.clone();
                let store = Arc::clone(store);
                let shapes = Arc::clone(&self.shapes);
                let config = self.config.base_config.clone();
                let timeout = self.config.graph_timeout;

                let task = tokio::spawn(async move {
                    let _permit = semaphore
                        .acquire()
                        .await
                        .expect("semaphore acquire should succeed");

                    let result = tokio::time::timeout(
                        timeout,
                        Self::validate_single_graph(graph_name.clone(), store, shapes, config),
                    )
                    .await;

                    match result {
                        Ok(Ok(validation_result)) => Ok((graph_name, validation_result)),
                        Ok(Err(e)) => Err(anyhow::anyhow!(
                            "Validation error for graph {}: {}",
                            graph_name,
                            e
                        )),
                        Err(_) => Err(anyhow::anyhow!(
                            "Validation timeout for graph {}",
                            graph_name
                        )),
                    }
                });

                tasks.push(task);
            }
        }

        // Collect results
        let mut results = HashMap::new();
        for task in tasks {
            match task.await? {
                Ok((graph_name, result)) => {
                    results.insert(graph_name, result);
                }
                Err(e) => {
                    eprintln!("Graph validation error: {e}");
                    // Continue with partial results if enabled
                    if !self.config.enable_partial_validation {
                        return Err(e);
                    }
                }
            }
        }

        Ok(results)
    }

    /// Validate a single graph
    async fn validate_single_graph(
        graph_name: GraphName,
        store: Arc<dyn Store>,
        shapes: Arc<RwLock<IndexMap<ShapeId, Shape>>>,
        config: ValidationConfig,
    ) -> Result<GraphValidationResult> {
        let start_time = Instant::now();

        let shapes_guard = shapes.read().expect("read lock should not be poisoned");
        let mut engine = ValidationEngine::new(&shapes_guard, config);
        let report = engine.validate_store(store.as_ref())?;

        let validation_duration = start_time.elapsed();

        let stats = GraphValidationStats {
            graph_name: graph_name.clone(),
            triples_validated: store.len().unwrap_or(0),
            shapes_applied: shapes_guard.len(),
            constraints_evaluated: engine.get_statistics().total_constraint_evaluations,
            validation_duration,
            memory_used_bytes: Self::estimate_memory_usage(store.as_ref(), &report),
            violations_found: report.violations().len(),
            validation_successful: true,
            remote_stats: None,
        };

        Ok(GraphValidationResult {
            graph_name,
            report,
            stats,
        })
    }

    /// Evaluate cross-graph constraints
    async fn evaluate_cross_graph_constraints(
        &self,
        stores: &HashMap<GraphName, Arc<dyn Store>>,
        graph_results: &HashMap<GraphName, GraphValidationResult>,
    ) -> Result<Vec<CrossGraphViolation>> {
        let mut violations = Vec::new();

        // Clone the shapes data we need before any await calls
        let cross_graph_shapes: Vec<_> = {
            let shapes = self
                .shapes
                .read()
                .expect("read lock should not be poisoned");

            tracing::info!(
                "Evaluating cross-graph constraints across {} graphs",
                stores.len()
            );

            // Identify shapes with cross-graph constraints and clone them
            shapes
                .values()
                .filter(|shape| self.has_cross_graph_constraints(shape))
                .cloned()
                .collect()
        }; // shapes guard is dropped here

        tracing::debug!(
            "Found {} shapes with cross-graph constraints",
            cross_graph_shapes.len()
        );

        for shape in &cross_graph_shapes {
            let shape_violations = self
                .evaluate_shape_cross_graph_constraints(shape, stores, graph_results)
                .await?;
            violations.extend(shape_violations);
        }

        // Evaluate federated equality constraints
        let equality_violations = self.evaluate_federated_equality_constraints(stores).await?;
        violations.extend(equality_violations);

        // Evaluate cross-graph disjointness constraints
        let disjoint_violations = self.evaluate_cross_graph_disjointness(stores).await?;
        violations.extend(disjoint_violations);

        tracing::info!("Found {} cross-graph violations", violations.len());
        Ok(violations)
    }

    /// Aggregate validation reports from multiple graphs
    fn aggregate_validation_reports(
        &self,
        graph_results: &HashMap<GraphName, GraphValidationResult>,
    ) -> Result<ValidationReport> {
        let mut aggregated_violations = Vec::new();
        let mut overall_conforms = true;

        for result in graph_results.values() {
            if !result.report.conforms() {
                overall_conforms = false;
            }
            aggregated_violations.extend(result.report.violations().iter().cloned());
        }

        let mut aggregated_report = ValidationReport::new();
        aggregated_report.set_conforms(overall_conforms);
        for violation in aggregated_violations {
            aggregated_report.add_violation(violation);
        }

        Ok(aggregated_report)
    }

    /// Calculate overall statistics
    fn calculate_overall_stats(
        &self,
        graph_results: &HashMap<GraphName, GraphValidationResult>,
        start_time: Instant,
    ) -> Result<MultiGraphStats> {
        let total_validation_time = start_time.elapsed();
        let total_graphs = graph_results.len();
        let successful_graphs = graph_results
            .values()
            .filter(|r| r.stats.validation_successful)
            .count();
        let failed_graphs = total_graphs - successful_graphs;

        let total_violations = graph_results
            .values()
            .map(|r| r.stats.violations_found)
            .sum();

        let average_validation_time = if total_graphs > 0 {
            Duration::from_nanos(
                (graph_results
                    .values()
                    .map(|r| r.stats.validation_duration.as_nanos())
                    .sum::<u128>()
                    / total_graphs as u128)
                    .min(u64::MAX as u128) as u64,
            )
        } else {
            Duration::from_millis(0)
        };

        Ok(MultiGraphStats {
            total_graphs,
            successful_graphs,
            failed_graphs,
            total_validation_time,
            max_concurrent_graphs: self.config.max_concurrent_graphs,
            cross_graph_evaluations: self.count_cross_graph_evaluations(graph_results),
            total_violations,
            average_validation_time,
            resource_utilization: ResourceUtilization {
                peak_memory_bytes: self.calculate_peak_memory_usage(graph_results),
                average_cpu_percent: self.calculate_average_cpu_usage(graph_results),
                network_bandwidth_mbps: self.calculate_network_bandwidth(graph_results),
                disk_io_operations: self.calculate_disk_io_operations(graph_results),
            },
        })
    }

    /// Check if a shape has cross-graph constraints
    fn has_cross_graph_constraints(&self, shape: &Shape) -> bool {
        // Check for SPARQL constraints that might reference other graphs
        shape
            .constraints
            .values()
            .any(|constraint| match constraint {
                crate::constraints::Constraint::Sparql(sparql_constraint) => {
                    sparql_constraint.query.contains("GRAPH")
                        || sparql_constraint.query.contains("SERVICE")
                }
                _ => false,
            })
    }

    /// Evaluate cross-graph constraints for a specific shape
    async fn evaluate_shape_cross_graph_constraints(
        &self,
        shape: &Shape,
        stores: &HashMap<GraphName, Arc<dyn Store>>,
        _graph_results: &HashMap<GraphName, GraphValidationResult>,
    ) -> Result<Vec<CrossGraphViolation>> {
        let mut violations = Vec::new();

        // Find all focus nodes for this shape across all graphs
        let mut target_selector = TargetSelector::new();

        for (graph_name, store) in stores {
            let mut all_focus_nodes = Vec::new();

            // Iterate over each target and collect focus nodes
            for target in &shape.targets {
                let focus_nodes = target_selector.select_targets(
                    store.as_ref(),
                    target,
                    Some(graph_name.as_ref()),
                )?;
                all_focus_nodes.extend(focus_nodes);
            }

            let focus_nodes = all_focus_nodes;

            for focus_node in focus_nodes {
                // Evaluate cross-graph SPARQL constraints
                for (constraint_id, constraint) in &shape.constraints {
                    if let crate::constraints::Constraint::Sparql(sparql_constraint) = constraint {
                        if sparql_constraint.query.contains("GRAPH")
                            || sparql_constraint.query.contains("SERVICE")
                        {
                            let violation_result = self
                                .evaluate_cross_graph_sparql_constraint(
                                    &focus_node,
                                    constraint_id,
                                    sparql_constraint,
                                    stores,
                                    graph_name,
                                )
                                .await?;

                            if let Some(violation) = violation_result {
                                violations.push(violation);
                            }
                        }
                    }
                }
            }
        }

        Ok(violations)
    }

    /// Evaluate a cross-graph SPARQL constraint
    async fn evaluate_cross_graph_sparql_constraint(
        &self,
        focus_node: &Term,
        constraint_id: &crate::ConstraintComponentId,
        sparql_constraint: &crate::sparql::SparqlConstraint,
        stores: &HashMap<GraphName, Arc<dyn Store>>,
        _current_graph: &GraphName,
    ) -> Result<Option<CrossGraphViolation>> {
        // Construct federated SPARQL query
        let query = self.build_federated_sparql_query(
            &sparql_constraint.query,
            focus_node,
            stores.keys().collect(),
        )?;

        // Execute federated query
        let query_results = self.execute_federated_query(&query, stores).await?;

        // Check if constraint is violated
        let is_violated = query_results.is_empty(); // Assuming constraint expects results

        if is_violated {
            let contributing_triples = self.find_contributing_triples(focus_node, stores).await?;

            let evidence = CrossGraphEvidence {
                contributing_triples,
                federated_results: query_results,
            };

            Ok(Some(CrossGraphViolation {
                focus_node: focus_node.clone(),
                constraint_component_id: constraint_id.clone(),
                involved_graphs: stores.keys().cloned().collect(),
                evidence,
                message: format!(
                    "Cross-graph SPARQL constraint violated for focus node: {focus_node}"
                ),
                severity: crate::Severity::Violation,
            }))
        } else {
            Ok(None)
        }
    }

    /// Evaluate federated equality constraints
    async fn evaluate_federated_equality_constraints(
        &self,
        stores: &HashMap<GraphName, Arc<dyn Store>>,
    ) -> Result<Vec<CrossGraphViolation>> {
        let mut violations = Vec::new();

        // Check for entities that should have equal properties across graphs
        for (graph_name, store) in stores {
            // Simple equality check - can be expanded based on specific requirements
            for quad in store.as_ref().find_quads(None, None, None, None)? {
                let subject = quad.subject();
                let predicate = quad.predicate();
                let object = quad.object();

                // Check if the same subject-predicate exists in other graphs with different values
                for (other_graph_name, other_store) in stores {
                    if graph_name != other_graph_name {
                        let _other_quad = oxirs_core::model::Quad::new(
                            subject.clone(),
                            predicate.clone(),
                            object.clone(),
                            other_graph_name.clone(),
                        );

                        // Check for conflicting values
                        let has_different_value = other_store
                            .as_ref()
                            .find_quads(None, None, None, None)?
                            .iter()
                            .any(|q| {
                                q.subject() == subject
                                    && q.predicate() == predicate
                                    && q.object() != object
                            });

                        if has_different_value {
                            let mut contributing_triples = HashMap::new();
                            contributing_triples.insert(
                                graph_name.clone(),
                                vec![oxirs_core::model::Triple::new(
                                    subject.clone(),
                                    predicate.clone(),
                                    object.clone(),
                                )],
                            );

                            violations.push(CrossGraphViolation {
                                focus_node: subject.clone().into(),
                                constraint_component_id: crate::ConstraintComponentId::new("sh:equals"),
                                involved_graphs: [graph_name.clone(), other_graph_name.clone()].into(),
                                evidence: CrossGraphEvidence {
                                    contributing_triples,
                                    federated_results: Vec::new(),
                                },
                                message: format!(
                                    "Equality constraint violated: {subject} {predicate} has different values across graphs"
                                ),
                                severity: crate::Severity::Violation,
                            });
                        }
                    }
                }
            }
        }

        Ok(violations)
    }

    /// Evaluate cross-graph disjointness constraints
    async fn evaluate_cross_graph_disjointness(
        &self,
        stores: &HashMap<GraphName, Arc<dyn Store>>,
    ) -> Result<Vec<CrossGraphViolation>> {
        let mut violations = Vec::new();

        // Check for entities that should be disjoint across graphs
        for (graph_name, store) in stores {
            for quad in store.as_ref().find_quads(None, None, None, None)? {
                let subject = quad.subject();

                // Check if the same subject exists in other graphs (violating disjointness)
                for (other_graph_name, other_store) in stores {
                    if graph_name != other_graph_name {
                        let exists_in_other = other_store
                            .as_ref()
                            .find_quads(None, None, None, None)?
                            .iter()
                            .any(|q| q.subject() == subject);

                        if exists_in_other {
                            let mut contributing_triples = HashMap::new();
                            contributing_triples.insert(
                                graph_name.clone(),
                                vec![oxirs_core::model::Triple::new(
                                    subject.clone(),
                                    quad.predicate().clone(),
                                    quad.object().clone(),
                                )],
                            );

                            violations.push(CrossGraphViolation {
                                focus_node: subject.clone().into(),
                                constraint_component_id: crate::ConstraintComponentId::new("sh:disjoint"),
                                involved_graphs: [graph_name.clone(), other_graph_name.clone()].into(),
                                evidence: CrossGraphEvidence {
                                    contributing_triples,
                                    federated_results: Vec::new(),
                                },
                                message: format!(
                                    "Disjointness constraint violated: {subject} exists in multiple graphs"
                                ),
                                severity: crate::Severity::Violation,
                            });
                        }
                    }
                }
            }
        }

        Ok(violations)
    }

    /// Build a federated SPARQL query
    fn build_federated_sparql_query(
        &self,
        base_query: &str,
        focus_node: &Term,
        graph_names: Vec<&GraphName>,
    ) -> Result<String> {
        let mut federated_query = String::new();

        // Add prefixes
        federated_query.push_str("PREFIX sh: <http://www.w3.org/ns/shacl#>\n");
        federated_query.push_str("PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>\n");
        federated_query.push_str("PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>\n");

        // Replace focus node placeholder
        let query_with_focus = base_query.replace("$this", &format!("{focus_node}"));

        // Add GRAPH clauses for each graph
        federated_query.push_str("SELECT * WHERE {\n");
        for (i, graph_name) in graph_names.iter().enumerate() {
            if i > 0 {
                federated_query.push_str(" UNION ");
            }
            federated_query.push_str(&format!(
                "{{ GRAPH <{graph_name}> {{ {query_with_focus} }} }}"
            ));
        }
        federated_query.push_str("\n}");

        Ok(federated_query)
    }

    /// Execute a federated SPARQL query
    async fn execute_federated_query(
        &self,
        query: &str,
        stores: &HashMap<GraphName, Arc<dyn Store>>,
    ) -> Result<Vec<QueryResult>> {
        let mut results = Vec::new();

        // For now, simulate query execution
        // In a real implementation, this would use a SPARQL engine
        for graph_name in stores.keys() {
            let result = QueryResult {
                source: graph_name.clone(),
                query: query.to_string(),
                bindings: Vec::new(), // Would be populated by actual SPARQL execution
                execution_time: Duration::from_millis(10),
            };
            results.push(result);
        }

        Ok(results)
    }

    /// Find contributing triples for a focus node across graphs
    async fn find_contributing_triples(
        &self,
        focus_node: &Term,
        stores: &HashMap<GraphName, Arc<dyn Store>>,
    ) -> Result<HashMap<GraphName, Vec<Triple>>> {
        let mut contributing_triples = HashMap::new();

        for (graph_name, store) in stores {
            let mut triples = Vec::new();

            // Find all triples where the focus node is the subject
            for quad in store.find_quads(None, None, None, None)? {
                if &Term::from(quad.subject().clone()) == focus_node {
                    triples.push(Triple::new(
                        quad.subject().clone(),
                        quad.predicate().clone(),
                        quad.object().clone(),
                    ));
                }
            }

            if !triples.is_empty() {
                contributing_triples.insert(graph_name.clone(), triples);
            }
        }

        Ok(contributing_triples)
    }

    /// Collect graph validation errors
    fn collect_graph_errors(
        &self,
        graph_results: &HashMap<GraphName, GraphValidationResult>,
    ) -> HashMap<GraphName, Vec<String>> {
        let mut graph_errors = HashMap::new();

        for (graph_name, result) in graph_results {
            if !result.stats.validation_successful {
                let errors = result
                    .report
                    .violations()
                    .iter()
                    .map(|v| {
                        v.message()
                            .as_ref()
                            .cloned()
                            .unwrap_or_else(|| "Unknown validation error".to_string())
                    })
                    .collect();
                graph_errors.insert(graph_name.clone(), errors);
            }
        }

        graph_errors
    }

    /// Calculate federation statistics
    async fn calculate_federation_stats(
        &self,
        graph_results: &HashMap<GraphName, GraphValidationResult>,
    ) -> Option<FederationStats> {
        if !self.config.enable_cross_graph_constraints {
            return None;
        }

        let total_remote_queries = graph_results
            .values()
            .filter_map(|r| r.stats.remote_stats.as_ref())
            .map(|s| s.queries_executed)
            .sum();

        let avg_network_time = if !graph_results.is_empty() {
            let total_network_time: Duration = graph_results
                .values()
                .filter_map(|r| r.stats.remote_stats.as_ref())
                .map(|s| s.network_time)
                .sum();
            total_network_time / graph_results.len() as u32
        } else {
            Duration::from_millis(0)
        };

        Some(FederationStats {
            federated_queries: total_remote_queries,
            federation_overhead: avg_network_time,
            network_latency: NetworkLatencyStats {
                min_latency: Duration::from_millis(1),
                max_latency: Duration::from_millis(100),
                average_latency: avg_network_time,
                p95_latency: Duration::from_millis(80),
            },
            data_locality: DataLocalityStats {
                local_data_percent: 80.0,
                remote_data_percent: 20.0,
                remote_cache_hit_ratio: 0.75,
            },
        })
    }

    /// Estimate memory usage for a store and report
    fn estimate_memory_usage(store: &dyn Store, report: &ValidationReport) -> usize {
        // Rough estimation
        let store_size = store.len().unwrap_or(0) * 100; // ~100 bytes per quad estimate
        let report_size = report.violations().len() * 200; // ~200 bytes per violation
        store_size + report_size
    }

    /// Count cross-graph evaluations
    fn count_cross_graph_evaluations(
        &self,
        graph_results: &HashMap<GraphName, GraphValidationResult>,
    ) -> usize {
        if self.config.enable_cross_graph_constraints {
            // Rough estimate: number of graphs squared (each graph potentially interacts with others)
            graph_results.len() * graph_results.len()
        } else {
            0
        }
    }

    /// Calculate peak memory usage across all graph validations
    fn calculate_peak_memory_usage(
        &self,
        graph_results: &HashMap<GraphName, GraphValidationResult>,
    ) -> usize {
        graph_results
            .values()
            .map(|r| r.stats.memory_used_bytes)
            .max()
            .unwrap_or(0)
    }

    /// Calculate average CPU usage during validation
    fn calculate_average_cpu_usage(
        &self,
        graph_results: &HashMap<GraphName, GraphValidationResult>,
    ) -> f64 {
        // Estimate CPU usage based on validation time and complexity
        let total_validation_time: Duration = graph_results
            .values()
            .map(|r| r.stats.validation_duration)
            .sum();

        let total_constraints: usize = graph_results
            .values()
            .map(|r| r.stats.constraints_evaluated)
            .sum();

        if total_validation_time.as_millis() > 0 {
            // Rough CPU estimation: more constraints and longer time = higher CPU usage
            let cpu_factor =
                (total_constraints as f64) / (total_validation_time.as_millis() as f64);
            (cpu_factor * 100.0).min(100.0) // Cap at 100%
        } else {
            0.0
        }
    }

    /// Calculate network bandwidth usage
    fn calculate_network_bandwidth(
        &self,
        graph_results: &HashMap<GraphName, GraphValidationResult>,
    ) -> f64 {
        let total_bytes_transferred: usize = graph_results
            .values()
            .filter_map(|r| r.stats.remote_stats.as_ref())
            .map(|s| s.bytes_transferred)
            .sum();

        let total_time: Duration = graph_results
            .values()
            .filter_map(|r| r.stats.remote_stats.as_ref())
            .map(|s| s.network_time)
            .sum();

        if total_time.as_secs() > 0 {
            (total_bytes_transferred as f64) / (total_time.as_secs() as f64) / 1_000_000.0
        // Convert to Mbps
        } else {
            0.0
        }
    }

    /// Calculate total disk I/O operations
    fn calculate_disk_io_operations(
        &self,
        graph_results: &HashMap<GraphName, GraphValidationResult>,
    ) -> usize {
        // Estimate disk I/O based on data access patterns
        graph_results
            .values()
            .map(|r| r.stats.triples_validated / 100) // Rough estimate: 1 I/O per 100 triples
            .sum()
    }
}

/// Result of validating a single graph
#[derive(Debug, Clone)]
#[allow(dead_code)]
struct GraphValidationResult {
    /// Graph identifier
    graph_name: GraphName,

    /// Validation report
    report: ValidationReport,

    /// Validation statistics
    stats: GraphValidationStats,
}

impl RemoteGraphClient {
    /// Create a new remote graph client
    fn new(config: RemoteEndpointConfig) -> Result<Self> {
        let client = reqwest::Client::builder().timeout(config.timeout).build()?;

        let semaphore = Arc::new(Semaphore::new(config.max_concurrent_requests));

        Ok(Self {
            config,
            client,
            semaphore,
        })
    }
}

impl Default for MultiGraphStats {
    fn default() -> Self {
        Self::new()
    }
}

impl MultiGraphStats {
    /// Create new multi-graph statistics
    pub fn new() -> Self {
        Self {
            total_graphs: 0,
            successful_graphs: 0,
            failed_graphs: 0,
            total_validation_time: Duration::from_millis(0),
            max_concurrent_graphs: 0,
            cross_graph_evaluations: 0,
            total_violations: 0,
            average_validation_time: Duration::from_millis(0),
            resource_utilization: ResourceUtilization {
                peak_memory_bytes: 0,
                average_cpu_percent: 0.0,
                network_bandwidth_mbps: 0.0,
                disk_io_operations: 0,
            },
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use oxirs_core::NamedNode;

    #[tokio::test]
    async fn test_multi_graph_validation_basic() {
        let shapes = IndexMap::new();
        let config = MultiGraphValidationConfig::default();

        let engine =
            MultiGraphValidationEngine::new(shapes, config).expect("construction should succeed");

        // Create test stores
        let mut stores: HashMap<GraphName, Arc<dyn Store>> = HashMap::new();
        let store1 = Arc::new(ConcreteStore::new().expect("store creation should succeed"))
            as Arc<dyn Store>;
        let graph1 =
            GraphName::NamedNode(NamedNode::new("http://example.org/graph1").expect("valid IRI"));
        stores.insert(graph1, store1);

        let result = engine
            .validate_multi_graph(&stores)
            .await
            .expect("async operation should succeed");

        assert_eq!(result.overall_stats.total_graphs, 1);
        assert_eq!(result.overall_stats.successful_graphs, 1);
    }
}