treetop-core 0.0.19

#![cfg_attr(not(feature = "observability"), allow(dead_code))]
//! Vendor-agnostic metrics collection via a pluggable sink.
//!
//! This module provides a trait-based sink pattern that allows consumers to
//! collect evaluation and reload metrics without tying the library to a specific
//! metrics backend (Prometheus, OpenTelemetry, CloudWatch, etc.).
//!
//! **Note:** This module is only available when the `observability` feature is enabled.
//!
//! Doctests in this module use `#[cfg(feature = "observability")]` to conditionally
//! compile: they will pass when the feature is disabled (code compiles away),
//! and execute when the feature is enabled (including with `cargo test --all-features`)
//! or `cargo test --features observability`.
//!
//! ## Usage
//!
//! Implement the [`MetricsSink`] trait to process evaluation and reload events:
//!
//! ```rust
//! # #[cfg(feature = "observability")]
//! # {
//! use treetop_core::metrics::{MetricsSink, EvaluationStats, ReloadStats};
//! use std::sync::atomic::{AtomicU64, Ordering};
//! use std::sync::Arc;
//!
//! struct MyMetricsSink {
//!     evals: Arc<AtomicU64>,
//! }
//!
//! impl MetricsSink for MyMetricsSink {
//!     fn on_evaluation(&self, stats: &EvaluationStats) {
//!         self.evals.fetch_add(1, Ordering::Relaxed);
//!         eprintln!("Eval took {:?}, allowed: {}", stats.duration, stats.allowed);
//!     }
//!
//!     fn on_reload(&self, _stats: &ReloadStats) {
//!         eprintln!("Policy reloaded");
//!     }
//! }
//!
//! // Set the global sink:
//! treetop_core::metrics::set_sink(Arc::new(MyMetricsSink { evals: Arc::new(AtomicU64::new(0)) }));
//! # }
//! ```
//!
//! For **Prometheus**, implement a sink that records to your Prometheus client,
//! or pipe [`tracing`] spans to OpenTelemetry.
//!
//! For **OpenTelemetry**, emit spans in `PolicyEngine::evaluate()` so consumers
//! can add OTel instrumentation via `tracing-opentelemetry`.

use arc_swap::{ArcSwap, Guard};
use serde::Serialize;
use std::sync::{Arc, OnceLock};
use std::time::{Duration, SystemTime};

/// Snapshot of a policy evaluation, passed to [`MetricsSink::on_evaluation`].
///
/// This struct captures the essential metrics from a single evaluation:
/// the total duration and the decision outcome. It is serializable so
/// consumers can easily log or transmit it.
///
/// # Fields
///
/// * `duration` - Total wall-clock time for the evaluation, including
///   label application, entity construction, and Cedar authorization.
/// * `allowed` - `true` if the decision was `Allow`, `false` if `Deny`.
/// * `action_id` - Fully qualified Cedar action identifier for optional,
///   bounded metric dimensions.
/// * `matched_policies` - List of policy IDs that matched during evaluation.
///   For `Allow` decisions, this contains the IDs of permit policies that matched.
///   For `Deny` decisions, this may contain forbid policies or be empty.
///
/// # Example
///
/// ```rust
/// # #[cfg(feature = "observability")]
/// # {
/// use treetop_core::metrics::EvaluationStats;
/// use std::time::Duration;
///
/// let stats = EvaluationStats {
///     duration: Duration::from_micros(500),
///     allowed: true,
///     action_id: r#"Action::"view_host""#.to_string(),
///     matched_policies: vec!["policy0".to_string()],
/// };
/// println!("Evaluation: {:?}ms, allowed: {}, action: {}, policies: {:?}",
///     stats.duration.as_millis(), stats.allowed, stats.action_id, stats.matched_policies);
/// # }
/// ```
#[derive(Debug, Clone, Serialize)]
pub struct EvaluationStats {
    /// Total time spent evaluating the request
    pub duration: Duration,
    /// Whether the decision was Allow (true) or Deny (false)
    pub allowed: bool,
    /// Fully qualified Cedar action identifier (e.g., `Action::"view_host"`).
    ///
    /// Use this as a metric label only when the application's action vocabulary
    /// is bounded and controlled.
    pub action_id: String,
    /// Policy IDs that matched during evaluation
    pub matched_policies: Vec<String>,
}

/// Detailed evaluation metrics broken down by phase.
///
/// This struct is useful for performance profiling and understanding where
/// time is spent during policy evaluation. All times are in milliseconds.
///
/// # Fields
///
/// * `apply_labels_ms` - Time spent applying label registry augmentation
/// * `construct_entities_ms` - Time spent converting atoms to Cedar EntityUids
/// * `resolve_groups_ms` - Time spent resolving group membership
/// * `authorize_ms` - Time spent in Cedar authorization engine
/// * `total_ms` - Total evaluation time (sum of phases + overhead)
///
#[derive(Debug, Clone, Serialize)]
pub struct EvaluationPhases {
    /// Time spent in label application phase (milliseconds)
    pub apply_labels_ms: f64,
    /// Time spent in entity construction phase (milliseconds)
    pub construct_entities_ms: f64,
    /// Time spent in group resolution phase (milliseconds)
    pub resolve_groups_ms: f64,
    /// Time spent in Cedar authorization phase (milliseconds)
    pub authorize_ms: f64,
    /// Total evaluation time (milliseconds)
    pub total_ms: f64,
}

impl EvaluationPhases {
    /// Calculate overhead time (time not accounted for in measured phases).
    ///
    /// Returns the difference between total time and the sum of individual phases.
    /// Uses `max(0.0, ...)` to ensure the result is never negative, which can
    /// occur due to timing precision or measurement overhead.
    pub fn overhead_ms(&self) -> f64 {
        (self.total_ms
            - (self.apply_labels_ms
                + self.construct_entities_ms
                + self.resolve_groups_ms
                + self.authorize_ms))
            .max(0.0)
    }
}

/// Snapshot of a policy reload event, passed to [`MetricsSink::on_reload`].
///
/// This struct captures the timestamp when a policy reload completed.
/// Consumers can use this to track when policy updates take effect.
///
/// # Fields
///
/// * `reload_time` - System time when the reload operation finished.
///
/// # Example
///
/// ```rust
/// # #[cfg(feature = "observability")]
/// # {
/// use treetop_core::metrics::ReloadStats;
///
/// let stats = ReloadStats {
///     reload_time: std::time::SystemTime::now(),
/// };
/// println!("Policy reloaded at: {:?}", stats.reload_time);
/// # }
/// ```
#[derive(Debug, Clone, Serialize)]
pub struct ReloadStats {
    /// Time when the reload completed
    pub reload_time: SystemTime,
}

/// Trait for consuming evaluation and reload metrics.
///
/// Implement this trait to collect policy engine metrics and send them
/// to any backend (Prometheus, OpenTelemetry, CloudWatch, Datadog, etc.).
///
/// The trait is automatically invoked by `PolicyEngine` after each evaluation
/// and policy reload. Implementations should be thread-safe (hence the
/// `Send + Sync` bounds) and should not block, as they are called in the
/// hot path.
///
/// # Default Implementation
///
/// If no sink is explicitly set via [`set_sink`], a built-in no-op sink
/// is used. The no-op sink disables metric payload allocation.
///
/// # Thread Safety
///
/// Implementations must be thread-safe because `PolicyEngine::evaluate` is
/// thread-safe and may be called concurrently from multiple threads.
/// Use atomic types, mutexes, or channels as appropriate.
/// Panics raised by a sink are caught at the library boundary and do not alter
/// authorization or reload results.
///
/// # Example: Simple Counter Sink
///
/// ```rust
/// # #[cfg(feature = "observability")]
/// # {
/// use treetop_core::metrics::{MetricsSink, EvaluationStats, ReloadStats};
/// use std::sync::atomic::{AtomicU64, Ordering};
/// use std::sync::Arc;
///
/// struct CounterSink {
///     evaluations: AtomicU64,
///     allows: AtomicU64,
///     denies: AtomicU64,
///     reloads: AtomicU64,
/// }
///
/// impl MetricsSink for CounterSink {
///     fn on_evaluation(&self, stats: &EvaluationStats) {
///         self.evaluations.fetch_add(1, Ordering::Relaxed);
///         if stats.allowed {
///             self.allows.fetch_add(1, Ordering::Relaxed);
///         } else {
///             self.denies.fetch_add(1, Ordering::Relaxed);
///         }
///     }
///
///     fn on_reload(&self, _stats: &ReloadStats) {
///         self.reloads.fetch_add(1, Ordering::Relaxed);
///     }
/// }
/// # }
/// ```
pub trait MetricsSink: Send + Sync {
    /// Whether this sink currently wants evaluation metrics.
    ///
    /// Returning `false` avoids metric-dimension allocation on the hot path.
    fn enabled(&self) -> bool {
        true
    }

    /// Called after each policy evaluation with timing and decision info.
    ///
    /// This method is invoked synchronously after every call to
    /// `PolicyEngine::evaluate`, regardless of the decision outcome.
    /// It should return quickly to avoid blocking evaluation.
    fn on_evaluation(&self, stats: &EvaluationStats);

    /// Called after each policy reload.
    ///
    /// This method is invoked after a successful call to
    /// `PolicyEngine::reload_from_str`. It provides the timestamp
    /// when the reload completed.
    fn on_reload(&self, stats: &ReloadStats);

    /// Called with detailed phase-level timing information (optional).
    ///
    /// This is an optional method for sinks that want to break down
    /// evaluation time by phase. The default implementation does nothing.
    /// Override this in your implementation to collect per-phase metrics.
    fn on_evaluation_phases(&self, _stats: &EvaluationStats, _phases: &EvaluationPhases) {
        // Default: no-op
    }
}

/// No-op sink; metrics are silently dropped.
///
/// This is the default sink used if none is explicitly set via [`set_sink`].
/// It avoids metric payload allocation.
struct NoOpSink;

impl MetricsSink for NoOpSink {
    fn enabled(&self) -> bool {
        false
    }

    fn on_evaluation(&self, _stats: &EvaluationStats) {}
    fn on_reload(&self, _stats: &ReloadStats) {}
}

// ArcSwap requires a sized pointee, so the inner Arc makes the trait object
// sized and the ArcSwap supplies the outer Arc. Keep the load guard alive
// through related callbacks to avoid another trait-object reference-count bump.
static SINK: OnceLock<ArcSwap<Arc<dyn MetricsSink>>> = OnceLock::new();

pub(crate) type SinkGuard = Guard<Arc<Arc<dyn MetricsSink>>>;

fn sink() -> SinkGuard {
    SINK.get_or_init(|| {
        let default: Arc<dyn MetricsSink> = Arc::new(NoOpSink);
        ArcSwap::from(Arc::new(default))
    })
    .load()
}

/// Set the global metrics sink.
///
/// This function allows you to change the metrics sink at runtime, enabling
/// dynamic switching between different metrics backends for testing,
/// configuration changes, or A/B testing different monitoring solutions.
///
/// # Thread Safety
///
/// Uses lock-free atomic operations via `ArcSwap`. Concurrent evaluations
/// will safely see either the old or new sink with no locks or blocking.
/// Metric collection itself never blocks on synchronization primitives.
///
/// # Performance
///
/// Uses `ArcSwap` for **lock-free atomic reads and swaps**. Concurrent evaluations
/// will see either the old or new sink atomically, with zero lock contention.
/// The old sink is dropped after all evaluations using it complete.
///
/// # Usage
///
/// All evaluation and reload events will be routed to this sink.
/// You can call this at any time to change the active sink:
///
/// ```rust,ignore
/// use std::sync::Arc;
/// use tokio;
/// use treetop_core::metrics::{set_sink, MetricsSink};
///
/// #[tokio::main]
/// async fn main() {
///     // Initial sink
///     let my_sink = Arc::new(MyMetricsSink::new());
///     treetop_core::metrics::set_sink(my_sink);
///
///     // ... run for a while ...
///
///     // Swap to a different sink (e.g., for testing or different backend)
///     let test_sink = Arc::new(TestMetricsSink::new());
///     treetop_core::metrics::set_sink(test_sink);
///
///     // All new evaluations will use the test sink
/// }
/// ```
///
/// # Resetting to No-Op
///
/// To disable metrics collection, swap to the no-op sink:
///
/// ```rust
/// # #[cfg(feature = "observability")]
/// # {
/// use std::sync::Arc;
/// use treetop_core::metrics::{MetricsSink, EvaluationStats, ReloadStats, set_sink};
///
/// // Define or import a no-op sink
/// struct NoOpSink;
/// impl MetricsSink for NoOpSink {
///     fn enabled(&self) -> bool { false }
///     fn on_evaluation(&self, _: &EvaluationStats) {}
///     fn on_reload(&self, _: &ReloadStats) {}
/// }
///
/// set_sink(Arc::new(NoOpSink));
/// # }
/// ```
pub fn set_sink(sink: Arc<dyn MetricsSink>) {
    SINK.get_or_init(|| {
        let default: Arc<dyn MetricsSink> = Arc::new(NoOpSink);
        ArcSwap::from(Arc::new(default))
    })
    .store(Arc::new(sink));
}

/// Get a reference to the current global sink.
///
/// This is an internal function used by the engine to dispatch metrics.
/// Most consumers should only call [`set_sink`] and implement [`MetricsSink`].
pub(crate) fn get_sink() -> SinkGuard {
    sink()
}

/// Query a sink without allowing backend panics to escape into engine logic.
pub(crate) fn metrics_enabled(sink: &SinkGuard) -> bool {
    std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| sink.enabled())).unwrap_or(false)
}

/// Dispatch related evaluation callbacks to one consistent sink snapshot.
pub(crate) fn record_evaluation_with_phases(
    sink: &SinkGuard,
    stats: &EvaluationStats,
    phases: &EvaluationPhases,
) {
    let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        sink.on_evaluation(stats);
        sink.on_evaluation_phases(stats, phases);
    }));
}

/// Record a reload event.
///
/// This is called internally by `PolicyEngine::reload_from_str` and should
/// not be called directly by consumers. It dispatches the metrics to the
/// global sink.
pub(crate) fn record_reload() {
    let sink = get_sink();
    let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        sink.on_reload(&ReloadStats {
            reload_time: SystemTime::now(),
        });
    }));
}

#[cfg(test)]
mod tests {
    use super::*;
    use serial_test::serial;
    use std::sync::Mutex;
    use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

    /// A simple test sink that counts evaluations and reloads.
    struct TestSink {
        eval_count: AtomicU64,
        allow_count: AtomicU64,
        deny_count: AtomicU64,
        reload_count: AtomicU64,
        last_eval_duration: Mutex<Option<Duration>>,
        was_called: AtomicBool,
    }

    impl TestSink {
        fn new() -> Self {
            Self {
                eval_count: AtomicU64::new(0),
                allow_count: AtomicU64::new(0),
                deny_count: AtomicU64::new(0),
                reload_count: AtomicU64::new(0),
                last_eval_duration: Mutex::new(None),
                was_called: AtomicBool::new(false),
            }
        }
    }

    impl MetricsSink for TestSink {
        fn on_evaluation(&self, stats: &EvaluationStats) {
            self.was_called.store(true, Ordering::SeqCst);
            self.eval_count.fetch_add(1, Ordering::SeqCst);
            if stats.allowed {
                self.allow_count.fetch_add(1, Ordering::SeqCst);
            } else {
                self.deny_count.fetch_add(1, Ordering::SeqCst);
            }
            if let Ok(mut d) = self.last_eval_duration.lock() {
                *d = Some(stats.duration);
            }
        }

        fn on_reload(&self, _stats: &ReloadStats) {
            self.reload_count.fetch_add(1, Ordering::SeqCst);
        }
    }

    struct ReloadOnlySink {
        reload_count: AtomicU64,
    }

    impl MetricsSink for ReloadOnlySink {
        fn enabled(&self) -> bool {
            false
        }

        fn on_evaluation(&self, _stats: &EvaluationStats) {
            panic!("evaluation callbacks must be disabled");
        }

        fn on_reload(&self, _stats: &ReloadStats) {
            self.reload_count.fetch_add(1, Ordering::SeqCst);
        }
    }

    #[test]
    fn test_evaluation_stats_serialization() {
        let stats = EvaluationStats {
            duration: Duration::from_millis(42),
            allowed: true,
            action_id: r#"Action::"view_host""#.to_string(),
            matched_policies: vec!["policy0".to_string()],
        };
        let json = serde_json::to_string(&stats).unwrap();
        assert!(json.contains("42") || json.contains("0.042")); // millis or seconds in JSON
        assert!(json.contains("true"));
        assert!(json.contains(r#"Action::\"view_host\""#));
        assert!(json.contains("policy0"));
    }

    #[test]
    fn test_reload_stats_serialization() {
        let now = SystemTime::now();
        let stats = ReloadStats { reload_time: now };
        let json = serde_json::to_string(&stats).unwrap();
        // Just ensure it serializes without error
        assert!(!json.is_empty());
    }

    #[test]
    #[serial(metrics)]
    fn test_record_reload_dispatches_to_current_sink() {
        let sink = Arc::new(TestSink::new());
        set_sink(sink.clone());

        record_reload();

        // Reload tests elsewhere in the crate may run concurrently while this
        // process-global sink is installed.
        assert!(sink.reload_count.load(Ordering::SeqCst) >= 1);
    }

    #[test]
    #[serial(metrics)]
    fn disabling_evaluations_does_not_disable_reload_events() {
        let sink = Arc::new(ReloadOnlySink {
            reload_count: AtomicU64::new(0),
        });
        set_sink(sink.clone());

        record_reload();

        assert!(sink.reload_count.load(Ordering::SeqCst) >= 1);
    }

    #[test]
    fn test_noop_sink_impl() {
        let sink = NoOpSink;
        let stats = EvaluationStats {
            duration: Duration::from_micros(1),
            allowed: true,
            action_id: r#"Action::"view_host""#.to_string(),
            matched_policies: vec![],
        };
        // Should not panic
        sink.on_evaluation(&stats);
        let reload_stats = ReloadStats {
            reload_time: SystemTime::now(),
        };
        sink.on_reload(&reload_stats);
    }

    #[test]
    fn test_evaluation_stats_clone() {
        let stats1 = EvaluationStats {
            duration: Duration::from_secs(1),
            allowed: false,
            action_id: r#"Action::"delete_host""#.to_string(),
            matched_policies: vec!["policy1".to_string()],
        };
        let stats2 = stats1.clone();
        assert_eq!(stats1.duration, stats2.duration);
        assert_eq!(stats1.allowed, stats2.allowed);
        assert_eq!(stats1.action_id, stats2.action_id);
        assert_eq!(stats1.matched_policies, stats2.matched_policies);
    }

    #[test]
    fn test_reload_stats_clone() {
        let now = SystemTime::now();
        let stats1 = ReloadStats { reload_time: now };
        let stats2 = stats1.clone();
        assert_eq!(stats1.reload_time, stats2.reload_time);
    }

    #[test]
    fn test_evaluation_stats_debug() {
        let stats = EvaluationStats {
            duration: Duration::from_micros(250),
            allowed: true,
            action_id: r#"Action::"view_host""#.to_string(),
            matched_policies: vec![],
        };
        let debug_str = format!("{:?}", stats);
        assert!(debug_str.contains("EvaluationStats"));
        assert!(debug_str.contains("true"));
        assert!(debug_str.contains("view_host"));
    }

    #[test]
    fn test_reload_stats_debug() {
        let stats = ReloadStats {
            reload_time: SystemTime::now(),
        };
        let debug_str = format!("{:?}", stats);
        assert!(debug_str.contains("ReloadStats"));
    }

    #[test]
    #[serial(metrics)]
    fn test_dynamic_sink_swapping() {
        // Create two different test sinks
        let sink1 = Arc::new(TestSink::new());
        let sink2 = Arc::new(TestSink::new());

        // Set first sink
        set_sink(sink1.clone());

        // Record evaluation with first sink
        let stats1 = EvaluationStats {
            duration: Duration::from_millis(10),
            allowed: true,
            action_id: r#"Action::"view_host""#.to_string(),
            matched_policies: vec![],
        };
        let phases = EvaluationPhases {
            apply_labels_ms: 0.0,
            construct_entities_ms: 0.0,
            resolve_groups_ms: 0.0,
            authorize_ms: 0.0,
            total_ms: 10.0,
        };
        record_evaluation_with_phases(&get_sink(), &stats1, &phases);

        // Swap to second sink
        set_sink(sink2.clone());

        // Record evaluation with second sink
        let stats2 = EvaluationStats {
            duration: Duration::from_millis(20),
            allowed: false,
            action_id: r#"Action::"delete_host""#.to_string(),
            matched_policies: vec![],
        };
        record_evaluation_with_phases(&get_sink(), &stats2, &phases);

        // Other unit tests may evaluate policies concurrently while a global sink is
        // installed. Each explicit dispatch above must reach its selected sink, but
        // those unrelated evaluations can legitimately increase either count.
        assert!(sink1.eval_count.load(Ordering::SeqCst) >= 1);
        assert!(sink2.eval_count.load(Ordering::SeqCst) >= 1);
    }
}