//! Circuit breaker pattern implementation
//!
//! Provides automatic failure detection and recovery mechanisms to prevent
//! cascading failures in distributed systems.
//!
//! # States
//!
//! - **Closed**: Normal operation, requests pass through
//! - **Open**: Too many failures detected, requests are rejected
//! - **HalfOpen**: Testing if service recovered, limited requests allowed
//!
//! # Example
//!
//! ```rust
//! use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
//! use std::time::Duration;
//!
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! let config = CircuitBreakerConfig {
//!     failure_threshold: 5,
//!     timeout: Duration::from_secs(30),
//!     success_threshold: 2,
//! };
//!
//! let cb = CircuitBreaker::new("api_service", config);
//!
//! if cb.can_execute().await {
//!     match make_api_call().await {
//!         Ok(result) => {
//!             cb.record_success().await;
//!             Ok(result)
//!         }
//!         Err(e) => {
//!             cb.record_failure().await;
//!             Err(e)
//!         }
//!     }
//! } else {
//!     Err("Circuit breaker open".into())
//! }
//! # }
//! # async fn make_api_call() -> Result<(), Box<dyn std::error::Error>> { Ok(()) }
//! ```

use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;

/// Core circuit breaker trait for the state machine pattern.
///
/// Implementors manage the Closed -> Open -> HalfOpen state transitions.
/// This trait provides a common interface so different circuit breaker
/// implementations (e.g., tokio-based, parking_lot-based) can be used
/// polymorphically.
#[async_trait::async_trait]
pub trait CircuitBreakerTrait: Send + Sync {
    /// Check if a request is allowed through.
    async fn can_execute(&self) -> bool;

    /// Record a successful operation.
    async fn record_success(&self);

    /// Record a failed operation.
    async fn record_failure(&self);

    /// Get current state.
    async fn state(&self) -> CircuitBreakerState;

    /// Reset to closed state.
    async fn reset(&self);
}

/// Circuit breaker states
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum CircuitBreakerState {
    /// Normal operation - requests pass through
    Closed,
    /// Too many failures - requests rejected
    Open,
    /// Testing recovery - limited requests allowed
    HalfOpen,
}

/// Circuit breaker configuration
#[derive(Debug, Clone)]
pub struct CircuitBreakerConfig {
    /// Number of consecutive failures before opening circuit
    pub failure_threshold: u32,
    /// Time to wait before transitioning from Open to HalfOpen
    pub timeout: Duration,
    /// Number of consecutive successes required to close circuit from HalfOpen
    pub success_threshold: u32,
}

impl Default for CircuitBreakerConfig {
    fn default() -> Self {
        Self {
            failure_threshold: 5,
            timeout: Duration::from_secs(30),
            success_threshold: 2,
        }
    }
}

/// Internal circuit breaker state
#[derive(Debug)]
struct CircuitBreakerInner {
    /// Current state of the circuit breaker
    state: CircuitBreakerState,
    /// Number of consecutive failures
    consecutive_failures: u32,
    /// Number of consecutive successes in HalfOpen state
    consecutive_successes: u32,
    /// Time when circuit was opened
    last_failure_time: Option<Instant>,
    /// Total number of successful requests
    total_successes: u64,
    /// Total number of failed requests
    total_failures: u64,
    /// Total number of requests
    total_requests: u64,
    /// Whether a probe request is currently in flight during HalfOpen state.
    /// Only one probe is allowed at a time to avoid overwhelming a recovering service.
    probe_in_flight: bool,
}

impl Default for CircuitBreakerInner {
    fn default() -> Self {
        Self {
            state: CircuitBreakerState::Closed,
            consecutive_failures: 0,
            consecutive_successes: 0,
            last_failure_time: None,
            total_successes: 0,
            total_failures: 0,
            total_requests: 0,
            probe_in_flight: false,
        }
    }
}

/// Circuit breaker metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CircuitBreakerMetrics {
    /// Current state
    pub state: CircuitBreakerState,
    /// Total successful requests
    pub total_successes: u64,
    /// Total failed requests
    pub total_failures: u64,
    /// Total requests
    pub total_requests: u64,
    /// Current consecutive failures
    pub consecutive_failures: u32,
    /// Time since last failure (milliseconds)
    pub time_since_last_failure_ms: Option<u64>,
}

/// Circuit breaker implementation
///
/// Thread-safe circuit breaker that prevents cascading failures by
/// automatically stopping requests to failing services.
#[derive(Clone)]
pub struct CircuitBreaker {
    /// Name for identification and logging
    name: String,
    /// Configuration
    config: CircuitBreakerConfig,
    /// Internal state (thread-safe)
    inner: Arc<Mutex<CircuitBreakerInner>>,
}

impl CircuitBreaker {
    /// Create a new circuit breaker
    ///
    /// # Arguments
    ///
    /// * `name` - Identifier for this circuit breaker (e.g., "database", "api_service")
    /// * `config` - Circuit breaker configuration
    ///
    /// # Example
    ///
    /// ```rust
    /// use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
    /// use std::time::Duration;
    ///
    /// let config = CircuitBreakerConfig {
    ///     failure_threshold: 5,
    ///     timeout: Duration::from_secs(30),
    ///     success_threshold: 2,
    /// };
    ///
    /// let cb = CircuitBreaker::new("my_service", config);
    /// ```
    pub fn new(name: impl Into<String>, config: CircuitBreakerConfig) -> Self {
        Self {
            name: name.into(),
            config,
            inner: Arc::new(Mutex::new(CircuitBreakerInner::default())),
        }
    }

    /// Check if request can be executed
    ///
    /// Returns `true` if the circuit breaker allows the request, `false` otherwise.
    /// Automatically transitions from Open to HalfOpen after timeout expires.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
    /// # async fn example(cb: CircuitBreaker) {
    /// if cb.can_execute().await {
    ///     // Execute request
    /// } else {
    ///     // Request rejected - circuit is open
    /// }
    /// # }
    /// ```
    pub async fn can_execute(&self) -> bool {
        let mut inner = self.inner.lock().await;

        match inner.state {
            CircuitBreakerState::Closed => true,
            CircuitBreakerState::Open => {
                // Check if timeout expired
                if let Some(last_failure) = inner.last_failure_time {
                    if last_failure.elapsed() >= self.config.timeout {
                        // Transition to HalfOpen
                        inner.state = CircuitBreakerState::HalfOpen;
                        inner.consecutive_successes = 0;
                        inner.probe_in_flight = true;
                        tracing::info!(
                            circuit_breaker = %self.name,
                            "Circuit breaker transitioning to HalfOpen"
                        );
                        true
                    } else {
                        false
                    }
                } else {
                    false
                }
            }
            CircuitBreakerState::HalfOpen => {
                if inner.probe_in_flight {
                    false
                } else {
                    inner.probe_in_flight = true;
                    true
                }
            }
        }
    }

    /// Record a successful request
    ///
    /// Updates circuit breaker state based on the success. In HalfOpen state,
    /// consecutive successes will close the circuit.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
    /// # async fn example(cb: CircuitBreaker) {
    /// // After successful operation
    /// cb.record_success().await;
    /// # }
    /// ```
    pub async fn record_success(&self) {
        let mut inner = self.inner.lock().await;

        inner.total_successes += 1;
        inner.total_requests += 1;
        inner.consecutive_failures = 0;

        match inner.state {
            CircuitBreakerState::HalfOpen => {
                inner.probe_in_flight = false;
                inner.consecutive_successes += 1;
                if inner.consecutive_successes >= self.config.success_threshold {
                    inner.state = CircuitBreakerState::Closed;
                    inner.consecutive_successes = 0;
                    tracing::info!(
                        circuit_breaker = %self.name,
                        "Circuit breaker closed after successful recovery"
                    );
                }
            }
            CircuitBreakerState::Closed => {
                // Already closed, just reset failure counter
            }
            CircuitBreakerState::Open => {
                // Should not happen - open circuit rejects requests
                tracing::warn!(
                    circuit_breaker = %self.name,
                    "Received success while circuit is open"
                );
            }
        }
    }

    /// Record a failed request
    ///
    /// Updates circuit breaker state based on the failure. Consecutive failures
    /// will open the circuit. A failure in HalfOpen state reopens the circuit.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
    /// # async fn example(cb: CircuitBreaker) {
    /// // After failed operation
    /// cb.record_failure().await;
    /// # }
    /// ```
    pub async fn record_failure(&self) {
        let mut inner = self.inner.lock().await;

        inner.total_failures += 1;
        inner.total_requests += 1;
        inner.consecutive_failures += 1;
        inner.consecutive_successes = 0;
        inner.last_failure_time = Some(Instant::now());

        match inner.state {
            CircuitBreakerState::Closed => {
                if inner.consecutive_failures >= self.config.failure_threshold {
                    inner.state = CircuitBreakerState::Open;
                    tracing::warn!(
                        circuit_breaker = %self.name,
                        failures = inner.consecutive_failures,
                        "Circuit breaker opened due to failures"
                    );
                }
            }
            CircuitBreakerState::HalfOpen => {
                // Any failure in HalfOpen reopens the circuit
                inner.probe_in_flight = false;
                inner.state = CircuitBreakerState::Open;
                tracing::warn!(
                    circuit_breaker = %self.name,
                    "Circuit breaker reopened after failure in HalfOpen state"
                );
            }
            CircuitBreakerState::Open => {
                // Already open, just update metrics
            }
        }
    }

    /// Get current circuit breaker state
    ///
    /// # Example
    ///
    /// ```rust
    /// # use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitBreakerState};
    /// # async fn example(cb: CircuitBreaker) {
    /// let state = cb.state().await;
    /// match state {
    ///     CircuitBreakerState::Closed => println!("Circuit is healthy"),
    ///     CircuitBreakerState::Open => println!("Circuit is open - service unavailable"),
    ///     CircuitBreakerState::HalfOpen => println!("Circuit is testing recovery"),
    /// }
    /// # }
    /// ```
    pub async fn state(&self) -> CircuitBreakerState {
        let inner = self.inner.lock().await;
        inner.state
    }

    /// Get circuit breaker metrics
    ///
    /// Returns current statistics including state, success/failure counts,
    /// and timing information.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
    /// # async fn example(cb: CircuitBreaker) {
    /// let metrics = cb.metrics().await;
    /// println!("Success rate: {:.2}%",
    ///     (metrics.total_successes as f64 / metrics.total_requests as f64) * 100.0);
    /// # }
    /// ```
    pub async fn metrics(&self) -> CircuitBreakerMetrics {
        let inner = self.inner.lock().await;
        CircuitBreakerMetrics {
            state: inner.state,
            total_successes: inner.total_successes,
            total_failures: inner.total_failures,
            total_requests: inner.total_requests,
            consecutive_failures: inner.consecutive_failures,
            time_since_last_failure_ms: inner
                .last_failure_time
                .map(|t| t.elapsed().as_millis() as u64),
        }
    }

    /// Get circuit breaker name
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Reset circuit breaker to initial state
    ///
    /// Useful for testing or manual recovery. Resets all counters and closes the circuit.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
    /// # async fn example(cb: CircuitBreaker) {
    /// // Reset to initial state
    /// cb.reset().await;
    /// # }
    /// ```
    pub async fn reset(&self) {
        let mut inner = self.inner.lock().await;
        *inner = CircuitBreakerInner::default();
        tracing::info!(
            circuit_breaker = %self.name,
            "Circuit breaker manually reset"
        );
    }
}

impl std::fmt::Debug for CircuitBreaker {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CircuitBreaker")
            .field("name", &self.name)
            .field("config", &self.config)
            .finish()
    }
}

#[async_trait::async_trait]
impl CircuitBreakerTrait for CircuitBreaker {
    async fn can_execute(&self) -> bool {
        CircuitBreaker::can_execute(self).await
    }

    async fn record_success(&self) {
        CircuitBreaker::record_success(self).await;
    }

    async fn record_failure(&self) {
        CircuitBreaker::record_failure(self).await;
    }

    async fn state(&self) -> CircuitBreakerState {
        CircuitBreaker::state(self).await
    }

    async fn reset(&self) {
        CircuitBreaker::reset(self).await;
    }
}

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

    /// Helper to create a circuit breaker that transitions to HalfOpen quickly.
    fn make_cb(failure_threshold: u32, success_threshold: u32) -> CircuitBreaker {
        CircuitBreaker::new(
            "test_cb",
            CircuitBreakerConfig {
                failure_threshold,
                timeout: Duration::from_millis(10),
                success_threshold,
            },
        )
    }

    /// Helper to force a circuit breaker into Open state by recording enough failures.
    async fn force_open(cb: &CircuitBreaker, failure_threshold: u32) {
        for _ in 0..failure_threshold {
            cb.record_failure().await;
        }
        assert_eq!(cb.state().await, CircuitBreakerState::Open);
    }

    /// Helper to force a circuit breaker into HalfOpen state by opening it, then
    /// waiting for the timeout and calling can_execute to trigger the transition.
    async fn force_half_open(cb: &CircuitBreaker, failure_threshold: u32) {
        force_open(cb, failure_threshold).await;
        // Wait for the timeout to expire so Open -> HalfOpen transition is possible
        tokio::time::sleep(Duration::from_millis(20)).await;
        // The transition happens inside can_execute, which also sets probe_in_flight.
        // We consume that first probe here.
        let allowed = cb.can_execute().await;
        assert!(allowed, "First probe after timeout should be allowed");
        assert_eq!(cb.state().await, CircuitBreakerState::HalfOpen);
    }

    #[tokio::test]
    async fn test_half_open_allows_only_one_probe() {
        let cb = make_cb(3, 2);

        // Force into HalfOpen (first probe consumed by force_half_open)
        force_half_open(&cb, 3).await;

        // Second can_execute while probe is in flight should be rejected
        let second = cb.can_execute().await;
        assert!(!second, "Second probe should be rejected while first is in flight");

        // Third attempt should also be rejected
        let third = cb.can_execute().await;
        assert!(!third, "Third probe should also be rejected");
    }

    #[tokio::test]
    async fn test_probe_resets_on_success() {
        let cb = make_cb(3, 1);

        // Force into HalfOpen (first probe consumed)
        force_half_open(&cb, 3).await;

        // Record success for the in-flight probe
        cb.record_success().await;

        // With success_threshold=1, should transition to Closed
        assert_eq!(cb.state().await, CircuitBreakerState::Closed);

        // Closed state should allow execution
        let allowed = cb.can_execute().await;
        assert!(allowed, "Closed circuit should allow execution");
    }

    #[tokio::test]
    async fn test_probe_resets_on_failure() {
        let cb = make_cb(3, 2);

        // Force into HalfOpen (first probe consumed)
        force_half_open(&cb, 3).await;

        // Record failure for the in-flight probe
        cb.record_failure().await;

        // Should transition back to Open
        assert_eq!(cb.state().await, CircuitBreakerState::Open);

        // Open state should reject execution (timeout hasn't expired again)
        let allowed = cb.can_execute().await;
        assert!(!allowed, "Open circuit should reject execution");
    }

    #[tokio::test]
    async fn test_probe_allows_new_probe_after_success_in_half_open() {
        // success_threshold=2, so one success stays in HalfOpen
        let cb = make_cb(3, 2);

        // Force into HalfOpen (first probe consumed)
        force_half_open(&cb, 3).await;

        // Record success -- still HalfOpen (need 2 successes)
        cb.record_success().await;
        assert_eq!(cb.state().await, CircuitBreakerState::HalfOpen);

        // probe_in_flight was reset, so a new probe should be allowed
        let allowed = cb.can_execute().await;
        assert!(allowed, "New probe should be allowed after previous success");

        // But a concurrent second probe should be blocked
        let blocked = cb.can_execute().await;
        assert!(!blocked, "Concurrent second probe should be blocked");
    }

    #[tokio::test]
    async fn test_open_to_half_open_transition_sets_probe_in_flight() {
        let cb = make_cb(3, 2);

        // Force open
        force_open(&cb, 3).await;

        // Wait for timeout
        tokio::time::sleep(Duration::from_millis(20)).await;

        // First can_execute transitions to HalfOpen AND sets probe_in_flight
        let first = cb.can_execute().await;
        assert!(first, "First call should transition to HalfOpen and allow probe");
        assert_eq!(cb.state().await, CircuitBreakerState::HalfOpen);

        // Second call should be blocked because probe is in flight
        let second = cb.can_execute().await;
        assert!(!second, "Second call should be blocked - probe already in flight");
    }

    #[tokio::test]
    async fn test_reset_clears_probe_in_flight() {
        let cb = make_cb(3, 2);

        // Force into HalfOpen with probe in flight
        force_half_open(&cb, 3).await;

        // Verify probe is in flight (second call blocked)
        assert!(!cb.can_execute().await, "Probe should be in flight");

        // Reset clears everything
        cb.reset().await;
        assert_eq!(cb.state().await, CircuitBreakerState::Closed);

        // After reset, should allow execution (Closed state)
        assert!(cb.can_execute().await, "Reset circuit should allow execution");
    }
}
