perf-sentinel-core 0.7.8

//! Pipeline: wires all stages together.

use crate::config::Config;
use crate::correlate;
use crate::detect;
use crate::detect::{Confidence, DetectConfig};
use crate::event::SpanEvent;
use crate::normalize;
use crate::report::{Analysis, Report};
use crate::score;

/// Run the full analysis pipeline on a batch of events.
#[must_use]
pub fn analyze(events: Vec<SpanEvent>, config: &Config) -> Report {
    analyze_with_traces(events, config).0
}

/// Run the full analysis pipeline, returning both the report and the correlated traces.
///
/// Use this when you need the intermediate `Trace` structures (e.g., for tree building
/// in the TUI inspect mode) without re-running normalization and correlation.
#[must_use]
pub fn analyze_with_traces(
    events: Vec<SpanEvent>,
    config: &Config,
) -> (Report, Vec<correlate::Trace>) {
    let start = std::time::Instant::now();
    let event_count = events.len();

    let normalized = normalize::normalize_all(events);
    let traces = correlate::correlate(normalized);
    let trace_count = traces.len();

    let detect_config = DetectConfig::from(config);
    let findings = detect::run_full_detection(&traces, &detect_config);

    let (mut findings, green_summary, per_endpoint_io_ops) = if config.green.enabled {
        let carbon_ctx = config.carbon_context();
        score::score_green(&traces, findings, Some(&carbon_ctx))
    } else {
        let total_io_ops = traces.iter().map(|t| t.spans.len()).sum();
        // Green disabled: skip the full scoring pass but still walk
        // the spans once for the per-endpoint counter. `score_green`
        // returns the same data as part of its own iteration when
        // enabled, so we never iterate twice in either branch.
        let per_endpoint_io_ops = crate::report::compute_per_endpoint_io_ops(&traces);
        (
            findings,
            crate::report::GreenSummary::disabled(total_io_ops),
            per_endpoint_io_ops,
        )
    };

    // Sort findings for deterministic output (HashMap iteration order is random)
    detect::sort_findings(&mut findings);

    // Stamp confidence on every finding. `analyze` is the batch path,
    // always CiBatch regardless of the daemon environment config. The
    // real daemon path (daemon::process_traces) stamps Staging or
    // Production from Config::confidence(). Detectors themselves never
    // reason about confidence; they emit Confidence::default() and the
    // pipeline caller overrides it here via the shared helper.
    detect::apply_confidence(&mut findings, Confidence::CiBatch);

    // Stamp the canonical signature so JSON consumers can copy-paste it
    // into `.perf-sentinel-acknowledgments.toml` without having to recompute.
    crate::acknowledgments::enrich_with_signatures(&mut findings);

    let quality_gate = crate::quality_gate::evaluate(&findings, &green_summary, config);

    let report = Report {
        analysis: Analysis {
            duration_ms: start.elapsed().as_millis() as u64,
            events_processed: event_count,
            traces_analyzed: trace_count,
        },
        findings,
        green_summary,
        quality_gate,
        per_endpoint_io_ops,
        // Batch mode does not run the cross-trace correlator, whose
        // rolling window only exists in the daemon. Always empty here.
        correlations: vec![],
        warnings: vec![],
        warning_details: vec![],
        acknowledged_findings: vec![],
        binary_version: env!("CARGO_PKG_VERSION").to_string(),
    };

    (report, traces)
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use super::*;
    use crate::event::SpanEvent;

    #[test]
    fn empty_pipeline_produces_empty_report() {
        let config = Config::default();
        let report = analyze(vec![], &config);
        assert!(report.findings.is_empty());
        assert_eq!(report.analysis.events_processed, 0);
        assert_eq!(report.analysis.traces_analyzed, 0);
        assert!(report.quality_gate.passed);
    }

    #[test]
    fn waste_dedup_no_double_count() {
        use crate::test_helpers::{make_sql_event, make_sql_series_events};
        // 5 different params + 2 duplicates of param 1 = 7 events, same template
        // N+1 sees 7 occurrences with 5 distinct params -> finding (avoidable = 6)
        // Redundant sees 3 occurrences of order_id=1 -> finding (avoidable = 2)
        // Without dedup: 6 + 2 = 8. With dedup: max(6, 2) = 6.
        let mut events: Vec<SpanEvent> = make_sql_series_events(5);
        // Add 2 more with order_id = 1 (duplicates)
        for i in 6..=7 {
            events.push(make_sql_event(
                "trace-1",
                &format!("span-{i}"),
                "SELECT * FROM order_item WHERE order_id = 1",
                &format!("2025-07-10T14:32:01.{:03}Z", i * 40),
            ));
        }

        let config = Config::default();
        let report = analyze(events, &config);
        assert!(!report.findings.is_empty());
        assert_eq!(report.green_summary.avoidable_io_ops, 6);
    }

    #[test]
    fn zero_events_waste_ratio_is_zero() {
        let config = Config::default();
        let report = analyze(vec![], &config);
        assert!((report.green_summary.io_waste_ratio - 0.0).abs() < f64::EPSILON);
        assert_eq!(report.green_summary.total_io_ops, 0);
        assert_eq!(report.green_summary.avoidable_io_ops, 0);
    }

    #[test]
    fn clean_events_zero_waste_ratio() {
        use crate::test_helpers::make_sql_event;
        // 4 events with different templates -> no N+1 (below threshold), no redundant
        let events = vec![
            make_sql_event(
                "trace-1",
                "span-1",
                "SELECT * FROM users WHERE id = 1",
                "2025-07-10T14:32:01.000Z",
            ),
            make_sql_event(
                "trace-1",
                "span-2",
                "SELECT * FROM orders WHERE id = 2",
                "2025-07-10T14:32:01.050Z",
            ),
            make_sql_event(
                "trace-1",
                "span-3",
                "SELECT * FROM products WHERE id = 3",
                "2025-07-10T14:32:01.100Z",
            ),
            make_sql_event(
                "trace-1",
                "span-4",
                "INSERT INTO logs (msg) VALUES ('ok')",
                "2025-07-10T14:32:01.150Z",
            ),
        ];

        let config = Config::default();
        let report = analyze(events, &config);

        assert!(report.findings.is_empty());
        assert_eq!(report.green_summary.total_io_ops, 4);
        assert_eq!(report.green_summary.avoidable_io_ops, 0);
        assert!((report.green_summary.io_waste_ratio - 0.0).abs() < f64::EPSILON);
    }

    #[test]
    fn pipeline_with_findings_computes_green_summary() {
        use crate::test_helpers::make_n_plus_one_events;
        // 6 events with different params -> N+1 finding
        let events = make_n_plus_one_events();

        let config = Config::default();
        let report = analyze(events, &config);

        assert!(!report.findings.is_empty());
        assert_eq!(report.green_summary.avoidable_io_ops, 5);
        assert!((report.green_summary.io_waste_ratio - 5.0_f64 / 6.0).abs() < f64::EPSILON);
        assert_eq!(report.green_summary.total_io_ops, 6);
    }

    #[test]
    fn dedup_across_traces() {
        use crate::test_helpers::make_sql_event;
        // Two traces, each with redundant queries on different templates
        let mut events = Vec::new();
        for i in 1..=3 {
            events.push(make_sql_event(
                "trace-A",
                &format!("span-a{i}"),
                "SELECT * FROM order_item WHERE order_id = 42",
                &format!("2025-07-10T14:32:01.{:03}Z", i * 50),
            ));
        }
        for i in 1..=3 {
            events.push(make_sql_event(
                "trace-B",
                &format!("span-b{i}"),
                "SELECT * FROM orders WHERE user_id = 7",
                &format!("2025-07-10T14:32:02.{:03}Z", i * 50),
            ));
        }

        let config = Config::default();
        let report = analyze(events, &config);

        // Each trace has 3 redundant -> avoidable = 2 each -> total = 4
        assert_eq!(report.green_summary.avoidable_io_ops, 4);
        assert_eq!(report.green_summary.total_io_ops, 6);
    }

    #[test]
    fn pipeline_with_green_default_region_produces_co2() {
        use crate::test_helpers::make_n_plus_one_events;
        let events = make_n_plus_one_events();

        let config = Config {
            green: crate::config::GreenConfig {
                default_region: Some("eu-west-3".to_string()),
                ..crate::config::GreenConfig::default()
            },
            ..Config::default()
        };
        let report = analyze(events, &config);

        let co2 = report
            .green_summary
            .co2
            .as_ref()
            .expect("co2 should be Some when default_region is configured");
        assert!(co2.total.mid > 0.0);
        assert!(co2.avoidable.mid > 0.0);
    }

    #[test]
    fn pipeline_empty_traces_no_co2() {
        // With 0 events, compute_carbon_report early-returns
        // (None, vec![]), nothing meaningful to report.
        // Avoids emitting a noisy all-zeros co2 object for empty daemon ticks.
        let config = Config::default();
        let report = analyze(vec![], &config);
        assert!(
            report.green_summary.co2.is_none(),
            "co2 should be None for empty traces"
        );
        assert!(report.green_summary.regions.is_empty());
    }

    #[test]
    fn green_disabled_skips_scoring() {
        use crate::test_helpers::make_n_plus_one_events;
        // 6 events -> N+1 finding, but green scoring disabled
        let events = make_n_plus_one_events();

        let config = Config {
            green: crate::config::GreenConfig {
                enabled: false,
                ..crate::config::GreenConfig::default()
            },
            ..Config::default()
        };
        let report = analyze(events, &config);

        // Findings are still detected
        assert!(!report.findings.is_empty());
        // But green scoring is bypassed
        assert_eq!(report.green_summary.avoidable_io_ops, 0);
        assert!((report.green_summary.io_waste_ratio - 0.0).abs() < f64::EPSILON);
        assert!(report.green_summary.top_offenders.is_empty());
        assert!(report.green_summary.co2.is_none());
        assert!(report.green_summary.regions.is_empty());
        // total_io_ops still counted
        assert_eq!(report.green_summary.total_io_ops, 6);
        // green_impact on findings should be None
        for f in &report.findings {
            assert!(f.green_impact.is_none());
        }
    }

    #[test]
    fn green_disabled_with_region_still_no_co2() {
        let config = Config {
            green: crate::config::GreenConfig {
                enabled: false,
                default_region: Some("eu-west-3".to_string()),
                ..crate::config::GreenConfig::default()
            },
            ..Config::default()
        };
        let report = analyze(vec![], &config);
        assert!(report.green_summary.co2.is_none());
    }

    // --- batch mode always stamps CiBatch ---

    #[test]
    fn batch_analyze_stamps_ci_batch_confidence() {
        use crate::test_helpers::make_n_plus_one_events;
        let events = make_n_plus_one_events();
        // Even with a production environment in config, batch analyze
        // must stamp CiBatch, confidence is mode-driven, not config-driven,
        // for `analyze` (the config `daemon.environment` only affects
        // `watch` daemon mode).
        let config = Config {
            daemon: crate::config::DaemonConfig {
                environment: crate::config::DaemonEnvironment::Production,
                ..crate::config::DaemonConfig::default()
            },
            ..Config::default()
        };
        let report = analyze(events, &config);
        assert!(!report.findings.is_empty());
        for f in &report.findings {
            assert_eq!(f.confidence, Confidence::CiBatch);
        }
    }

    // ---------------------------------------------------------------
    // Sharded trace routing correctness
    // ---------------------------------------------------------------

    /// Simulate sticky `trace_id` sharding across N instances.
    /// Split events by FNV-1a hash of `trace_id` (same algorithm as
    /// daemon sampling), run the pipeline on each shard independently,
    /// then verify that every per-trace finding from the baseline
    /// (non-sharded) run also appears in the sharded results.
    ///
    /// This validates the claim that horizontal scaling via an `OTel`
    /// Collector `loadbalancingexporter` produces the same per-trace
    /// detection results as a single daemon instance.
    #[test]
    fn sharded_detection_matches_single_instance() {
        use std::collections::{HashMap, HashSet};
        const NUM_SHARDS: u64 = 2;

        // Build a dataset with 4 distinct traces, each containing an
        // N+1 SQL pattern (6 similar queries). Use different services
        // so the shards are non-trivial.
        let traces_data = [
            ("trace-A", "svc-alpha"),
            ("trace-B", "svc-beta"),
            ("trace-C", "svc-gamma"),
            ("trace-D", "svc-delta"),
        ];
        let mut all_events: Vec<SpanEvent> = Vec::new();
        for (trace_id, service) in &traces_data {
            for i in 0..6 {
                let ts = format!("2025-07-10T14:32:01.{i:03}Z");
                let mut ev = crate::test_helpers::make_sql_event(
                    trace_id,
                    &format!("span-{trace_id}-{i}"),
                    &format!("SELECT * FROM orders WHERE id = {}", 100 + i),
                    &ts,
                );
                ev.service = Arc::from(*service);
                all_events.push(ev);
            }
        }

        let config = Config::default();

        // Baseline: all events in a single instance
        let baseline = analyze(all_events.clone(), &config);
        assert!(
            baseline.findings.len() >= 4,
            "expected at least 4 findings (one N+1 per trace), got {}",
            baseline.findings.len()
        );

        // Shard into 2 buckets by FNV-1a hash of trace_id
        let mut shards: Vec<Vec<SpanEvent>> = vec![vec![]; NUM_SHARDS as usize];
        for event in &all_events {
            let hash = fnv1a_hash(event.trace_id.as_bytes());
            let bucket = (hash % NUM_SHARDS) as usize;
            shards[bucket].push(event.clone());
        }

        // Verify each shard got at least one trace (hash distribution)
        for (i, shard) in shards.iter().enumerate() {
            assert!(
                !shard.is_empty(),
                "shard {i} is empty, hash distribution failed"
            );
        }

        // Verify no trace is split across shards
        let mut trace_to_shard: HashMap<String, usize> = HashMap::new();
        for (i, shard) in shards.iter().enumerate() {
            for ev in shard {
                if let Some(&prev) = trace_to_shard.get(&ev.trace_id) {
                    assert_eq!(
                        prev, i,
                        "trace {} split across shards {prev} and {i}",
                        ev.trace_id
                    );
                }
                trace_to_shard.insert(ev.trace_id.clone(), i);
            }
        }

        // Run pipeline on each shard independently
        let mut sharded_findings = Vec::new();
        for shard in shards {
            let report = analyze(shard, &config);
            sharded_findings.extend(report.findings);
        }

        // Build a set of (trace_id, finding_type) for comparison.
        // Cross-trace findings (slow percentiles) are excluded because
        // they depend on seeing all traces, which sharding splits.
        let baseline_set: HashSet<(String, String)> = baseline
            .findings
            .iter()
            .filter(|f| {
                !matches!(
                    f.finding_type,
                    detect::FindingType::SlowSql | detect::FindingType::SlowHttp
                )
            })
            .map(|f| (f.trace_id.clone(), f.finding_type.as_str().to_string()))
            .collect();
        let sharded_set: HashSet<(String, String)> = sharded_findings
            .iter()
            .filter(|f| {
                !matches!(
                    f.finding_type,
                    detect::FindingType::SlowSql | detect::FindingType::SlowHttp
                )
            })
            .map(|f| (f.trace_id.clone(), f.finding_type.as_str().to_string()))
            .collect();

        assert_eq!(
            baseline_set, sharded_set,
            "sharded findings differ from baseline.\n\
             baseline: {baseline_set:?}\n\
             sharded:  {sharded_set:?}"
        );
    }

    /// FNV-1a hash (same algorithm as `daemon::hash_trace_id`).
    fn fnv1a_hash(bytes: &[u8]) -> u64 {
        let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
        for &b in bytes {
            hash ^= u64::from(b);
            hash = hash.wrapping_mul(0x0100_0000_01b3);
        }
        hash
    }
}