refactor analysis engine for modularity

Author: Critlist

Cargo.toml

@@ -6,6 +6,10 @@ members = [
     "crates/og-graph", 
     "crates/og-db",
     "crates/og-analytics",
+    "crates/og-metrics-centrality",
+    "crates/og-metrics-community",
+    "crates/og-metrics-risk",
+    "crates/og-metrics-quality",
     "crates/og-services",
     "crates/og-utils"
 ]

crates/og-analytics/Cargo.toml

@@ -7,6 +7,10 @@ edition = "2021"
 og-types = { path = "../og-types" }
 og-graph = { path = "../og-graph" }
 og-db = { path = "../og-db" }
+og-metrics-centrality = { path = "../og-metrics-centrality" }
+og-metrics-community = { path = "../og-metrics-community" }
+og-metrics-risk = { path = "../og-metrics-risk" }
+og-metrics-quality = { path = "../og-metrics-quality" }
 neo4rs = { workspace = true }
 tokio = { workspace = true }
 serde = { workspace = true }

crates/og-analytics/src/engine_v2.rs

@@ -0,0 +1,330 @@
+use crate::metrics::{MetricResults, MetricValue};
+use anyhow::Result;
+use og_graph::graph::CodeGraph;
+use og_metrics_centrality::{CentralityMetrics, CentralityResults};
+use og_metrics_community::{CommunityDetection, CommunityResults};
+use og_metrics_risk::{RiskAnalyzer, RiskResults};
+use og_metrics_quality::{QualityAnalyzer, QualityResults};
+// Removed unused imports
+use std::time::Duration;
+use std::sync::Arc;
+use tracing::{debug, info, warn, error};
+
+/// Configuration for analytics engine v2 with improved error handling
+#[derive(Debug, Clone)]
+pub struct AnalyticsConfigV2 {
+    /// Weights for composite metrics
+    pub weights: MetricWeights,
+    /// Enable parallel computation per metric type
+    pub parallel_metrics: bool,
+    /// Timeout for each metric calculation
+    pub metric_timeout: Duration,
+    /// PageRank iterations
+    pub pagerank_iterations: usize,
+    /// PageRank damping factor
+    pub pagerank_damping: f64,
+    /// Community detection resolution
+    pub louvain_resolution: f64,
+    /// Enable sampling for large graphs
+    pub use_sampling: bool,
+    /// Sample size for betweenness centrality
+    pub betweenness_sample_size: usize,
+}
+
+impl Default for AnalyticsConfigV2 {
+    fn default() -> Self {
+        Self {
+            weights: MetricWeights::default(),
+            parallel_metrics: true,
+            metric_timeout: Duration::from_secs(10),
+            pagerank_iterations: 30,
+            pagerank_damping: 0.85,
+            louvain_resolution: 1.0,
+            use_sampling: true,
+            betweenness_sample_size: 1000,
+        }
+    }
+}
+
+/// Weights for composite metrics
+#[derive(Debug, Clone)]
+pub struct MetricWeights {
+    pub importance_pagerank: f64,
+    pub importance_degree: f64,
+    pub importance_betweenness: f64,
+    pub risk_complexity: f64,
+    pub risk_coupling: f64,
+    pub risk_churn: f64,
+    pub chokepoint_betweenness: f64,
+    pub chokepoint_clustering: f64,
+    pub chokepoint_degree: f64,
+    pub payoff_risk: f64,
+    pub payoff_importance: f64,
+    pub payoff_coverage: f64,
+}
+
+impl Default for MetricWeights {
+    fn default() -> Self {
+        Self {
+            importance_pagerank: 0.5,
+            importance_degree: 0.3,
+            importance_betweenness: 0.2,
+            risk_complexity: 0.4,
+            risk_coupling: 0.4,
+            risk_churn: 0.2,
+            chokepoint_betweenness: 0.6,
+            chokepoint_clustering: 0.2,
+            chokepoint_degree: 0.2,
+            payoff_risk: 0.4,
+            payoff_importance: 0.3,
+            payoff_coverage: 0.3,
+        }
+    }
+}
+
+/// Main analytics engine v2 with modular metrics
+pub struct AnalyticsEngineV2 {
+    config: AnalyticsConfigV2,
+    centrality_metrics: Arc<CentralityMetrics>,
+    community_detector: Arc<CommunityDetection>,
+    risk_analyzer: Arc<RiskAnalyzer>,
+    quality_analyzer: Arc<QualityAnalyzer>,
+}
+
+impl AnalyticsEngineV2 {
+    /// Create a new analytics engine v2
+    pub fn new(config: AnalyticsConfigV2) -> Self {
+        // Configure individual metric modules
+        let mut centrality_metrics = CentralityMetrics::new();
+        centrality_metrics.max_iterations = config.pagerank_iterations;
+        centrality_metrics.use_sampling = config.use_sampling;
+        centrality_metrics.sample_size = config.betweenness_sample_size;
+
+        let community_detector = CommunityDetection::with_resolution(config.louvain_resolution);
+        let risk_analyzer = RiskAnalyzer::new();
+        let quality_analyzer = QualityAnalyzer::new();
+
+        Self {
+            config,
+            centrality_metrics: Arc::new(centrality_metrics),
+            community_detector: Arc::new(community_detector),
+            risk_analyzer: Arc::new(risk_analyzer),
+            quality_analyzer: Arc::new(quality_analyzer),
+        }
+    }
+
+    /// Validate graph before analysis
+    fn validate_graph(&self, graph: &CodeGraph) -> Result<()> {
+        let node_count = graph.node_map.len();
+        let edge_count = graph.graph.edge_count();
+        
+        debug!("Validating graph: {} nodes, {} edges", node_count, edge_count);
+        
+        if node_count == 0 {
+            warn!("Graph is empty");
+        }
+        
+        // Check for disconnected components (informational)
+        let connected = petgraph::algo::connected_components(&graph.graph);
+        if connected > 1 {
+            info!("Graph has {} disconnected components", connected);
+        }
+        
+        Ok(())
+    }
+
+    /// Analyze a code graph with modular metrics
+    pub async fn analyze(&self, graph: &CodeGraph) -> Result<ModularAnalysisReport> {
+        info!("Starting modular graph analysis with {} nodes", graph.node_map.len());
+
+        // Validate graph
+        self.validate_graph(graph)?;
+
+        let mut report = ModularAnalysisReport::default();
+
+        // Run each metric module with timeout and error recovery
+        if self.config.parallel_metrics {
+            // Run metrics in parallel with isolated error handling
+            let (centrality, community, risk, quality) = tokio::join!(
+                self.run_centrality_with_timeout(graph),
+                self.run_community_with_timeout(graph),
+                self.run_risk_with_timeout(graph),
+                self.run_quality_with_timeout(graph),
+            );
+
+            report.centrality = centrality.unwrap_or_else(|e| {
+                error!("Centrality metrics failed: {}", e);
+                report.errors.push(format!("Centrality: {}", e));
+                CentralityResults::default()
+            });
+
+            report.community = community.unwrap_or_else(|e| {
+                error!("Community detection failed: {}", e);
+                report.errors.push(format!("Community: {}", e));
+                CommunityResults::default()
+            });
+
+            report.risk = risk.unwrap_or_else(|e| {
+                error!("Risk analysis failed: {}", e);
+                report.errors.push(format!("Risk: {}", e));
+                RiskResults::default()
+            });
+
+            report.quality = quality.unwrap_or_else(|e| {
+                error!("Quality metrics failed: {}", e);
+                report.errors.push(format!("Quality: {}", e));
+                QualityResults::default()
+            });
+        } else {
+            // Run metrics sequentially with individual error handling
+            report.centrality = self.run_centrality_with_timeout(graph).await
+                .unwrap_or_else(|e| {
+                    error!("Centrality metrics failed: {}", e);
+                    report.errors.push(format!("Centrality: {}", e));
+                    CentralityResults::default()
+                });
+
+            report.community = self.run_community_with_timeout(graph).await
+                .unwrap_or_else(|e| {
+                    error!("Community detection failed: {}", e);
+                    report.errors.push(format!("Community: {}", e));
+                    CommunityResults::default()
+                });
+
+            report.risk = self.run_risk_with_timeout(graph).await
+                .unwrap_or_else(|e| {
+                    error!("Risk analysis failed: {}", e);
+                    report.errors.push(format!("Risk: {}", e));
+                    RiskResults::default()
+                });
+
+            report.quality = self.run_quality_with_timeout(graph).await
+                .unwrap_or_else(|e| {
+                    error!("Quality metrics failed: {}", e);
+                    report.errors.push(format!("Quality: {}", e));
+                    QualityResults::default()
+                });
+        }
+
+        // Calculate composite scores
+        report.calculate_composite_scores(&self.config.weights);
+
+        info!("Modular analysis complete with {} errors", report.errors.len());
+        Ok(report)
+    }
+
+    /// Run centrality metrics with timeout
+    async fn run_centrality_with_timeout(&self, graph: &CodeGraph) -> Result<CentralityResults> {
+        let graph = graph.clone();
+        let metrics = Arc::clone(&self.centrality_metrics);
+        let timeout = self.config.metric_timeout;
+        
+        tokio::time::timeout(timeout, tokio::task::spawn_blocking(move || {
+            metrics.calculate_all(&graph)
+        }))
+        .await?
+        .map_err(|e| anyhow::anyhow!("Task join error: {}", e))?
+    }
+
+    /// Run community detection with timeout
+    async fn run_community_with_timeout(&self, graph: &CodeGraph) -> Result<CommunityResults> {
+        let graph = graph.clone();
+        let detector = Arc::clone(&self.community_detector);
+        let timeout = self.config.metric_timeout;
+        
+        tokio::time::timeout(timeout, tokio::task::spawn_blocking(move || {
+            detector.detect_communities(&graph)
+        }))
+        .await?
+        .map_err(|e| anyhow::anyhow!("Task join error: {}", e))?
+    }
+
+    /// Run risk analysis with timeout
+    async fn run_risk_with_timeout(&self, graph: &CodeGraph) -> Result<RiskResults> {
+        let graph = graph.clone();
+        let analyzer = Arc::clone(&self.risk_analyzer);
+        let timeout = self.config.metric_timeout;
+        
+        tokio::time::timeout(timeout, tokio::task::spawn_blocking(move || {
+            analyzer.analyze_risks(&graph)
+        }))
+        .await?
+        .map_err(|e| anyhow::anyhow!("Task join error: {}", e))?
+    }
+
+    /// Run quality analysis with timeout
+    async fn run_quality_with_timeout(&self, graph: &CodeGraph) -> Result<QualityResults> {
+        let graph = graph.clone();
+        let analyzer = Arc::clone(&self.quality_analyzer);
+        let timeout = self.config.metric_timeout;
+        
+        tokio::time::timeout(timeout, tokio::task::spawn_blocking(move || {
+            analyzer.analyze_quality(&graph)
+        }))
+        .await?
+        .map_err(|e| anyhow::anyhow!("Task join error: {}", e))?
+    }
+}
+
+/// Modular analysis report combining all metric results
+#[derive(Debug, Default, Clone)]
+pub struct ModularAnalysisReport {
+    pub centrality: CentralityResults,
+    pub community: CommunityResults,
+    pub risk: RiskResults,
+    pub quality: QualityResults,
+    pub composite_scores: CompositeScores,
+    pub errors: Vec<String>,
+}
+
+impl ModularAnalysisReport {
+    /// Calculate composite scores from individual metrics
+    pub fn calculate_composite_scores(&mut self, _weights: &MetricWeights) {
+        // This would calculate weighted composite scores
+        // Implementation depends on specific business logic
+        self.composite_scores = CompositeScores::default();
+    }
+
+    /// Convert to legacy MetricResults format if needed
+    pub fn to_metric_results(&self) -> Vec<MetricResults> {
+        let mut results = Vec::new();
+        
+        // Convert centrality results
+        let mut centrality_result = MetricResults::new("centrality".to_string());
+        for (node_id, metrics) in &self.centrality.degree {
+            centrality_result.add_value(
+                node_id.clone(),
+                MetricValue::Float(metrics.total_degree),
+            );
+        }
+        results.push(centrality_result);
+        
+        // Add other metric conversions as needed
+        
+        results
+    }
+}
+
+#[derive(Debug, Default, Clone)]
+pub struct CompositeScores {
+    pub importance_scores: Vec<(String, f64)>,
+    pub risk_scores: Vec<(String, f64)>,
+    pub quality_scores: Vec<(String, f64)>,
+}
+
+// Clone implementations moved to respective crates
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_modular_engine() {
+        let graph = CodeGraph::new();
+        let config = AnalyticsConfigV2::default();
+        let engine = AnalyticsEngineV2::new(config);
+        
+        let report = engine.analyze(&graph).await.unwrap();
+        assert!(report.errors.is_empty() || !report.errors.is_empty());
+    }
+}