mecha10-core 0.6.5

//! RPC with Circuit Breaker Integration
//!
//! Combines RPC calls with circuit breaker pattern for improved fault tolerance.
//! Prevents cascading failures when services are down or slow.
//!
//! # Features
//!
//! - Automatic circuit breaking for failing RPC endpoints
//! - Per-endpoint circuit breaker configuration
//! - Shared circuit breaker registry for efficiency
//! - Fail-fast when circuit is open
//! - Compatible with existing RPC API
//!
//! # Example
//!
//! ```rust
//! use mecha10_core::prelude::*;
//! use mecha10_core::rpc_breaker::{RpcBreakerExt, CircuitBreakerConfig};
//!
//! # async fn example(ctx: &Context) -> Result<()> {
//! // Make RPC call with circuit breaker
//! let result: CommandResult = ctx
//!     .request_with_breaker(
//!         "/motor/control",
//!         &command,
//!         CircuitBreakerConfig {
//!             failure_threshold: 5,
//!             success_threshold: 2,
//!             timeout: Duration::from_secs(30),
//!             name: Some("motor_control".to_string()),
//!         }
//!     )
//!     .timeout(Duration::from_secs(5))
//!     .execute()
//!     .await?;
//! # Ok(())
//! # }
//! ```

use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
use crate::context::Context;
use crate::error::Result;
use crate::messages::Message;
use crate::rpc::RpcExt;
use serde::{de::DeserializeOwned, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;

/// Global registry of circuit breakers for RPC endpoints
///
/// Maintains one circuit breaker per endpoint to track health across all calls.
#[derive(Clone)]
pub struct CircuitBreakerRegistry {
    breakers: Arc<RwLock<HashMap<String, CircuitBreaker>>>,
}

impl CircuitBreakerRegistry {
    /// Create a new empty registry
    pub fn new() -> Self {
        Self {
            breakers: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// Get or create a circuit breaker for an endpoint
    ///
    /// # Arguments
    ///
    /// * `endpoint` - Endpoint identifier (typically the topic)
    /// * `config` - Configuration for the circuit breaker
    ///
    /// # Returns
    ///
    /// A circuit breaker for this endpoint
    pub async fn get_or_create(&self, endpoint: &str, config: CircuitBreakerConfig) -> CircuitBreaker {
        let mut breakers = self.breakers.write().await;

        breakers
            .entry(endpoint.to_string())
            .or_insert_with(|| CircuitBreaker::new(config))
            .clone()
    }

    /// Get statistics for all circuit breakers
    pub async fn stats(&self) -> HashMap<String, CircuitBreakerStats> {
        let breakers = self.breakers.read().await;
        let mut stats_map = HashMap::new();

        for (endpoint, breaker) in breakers.iter() {
            let (total_requests, total_successes, total_failures, total_rejected) = breaker.stats().await;
            let success_rate = breaker.success_rate().await;
            let state = if breaker.is_open().await {
                "open".to_string()
            } else if breaker.is_half_open().await {
                "half-open".to_string()
            } else {
                "closed".to_string()
            };

            stats_map.insert(
                endpoint.clone(),
                CircuitBreakerStats {
                    state,
                    total_requests,
                    total_successes,
                    total_failures,
                    total_rejected,
                    success_rate,
                },
            );
        }

        stats_map
    }

    // duplicate: false positive — matches `SecretsManager::clear_cache` and
    // `MultiKeyRateLimiter::clear` on AST shape only (write-lock a map, `.clear()` it); the
    // three types share no relationship that a generic helper would express cleanly.
    /// Reset all circuit breakers
    pub async fn reset_all(&self) {
        let mut breakers = self.breakers.write().await;
        breakers.clear();
    }
}

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

/// Circuit breaker statistics
#[derive(Debug, Clone)]
pub struct CircuitBreakerStats {
    pub state: String,
    pub total_requests: u64,
    pub total_successes: u64,
    pub total_failures: u64,
    pub total_rejected: u64,
    pub success_rate: f64,
}

/// Extension trait for RPC with circuit breaker support
pub trait RpcBreakerExt {
    /// Make an RPC request with circuit breaker protection
    ///
    /// # Arguments
    ///
    /// * `topic` - Base topic for the RPC
    /// * `request` - Request payload
    /// * `config` - Circuit breaker configuration
    ///
    /// # Returns
    ///
    /// A request builder that can be configured and executed
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10_core::prelude::*;
    /// use mecha10_core::rpc_breaker::{RpcBreakerExt, CircuitBreakerConfig};
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let result: Response = ctx
    ///     .request_with_breaker(
    ///         "/motor/control",
    ///         &command,
    ///         CircuitBreakerConfig::default()
    ///     )
    ///     .timeout(Duration::from_secs(5))
    ///     .execute()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    fn request_with_breaker<Req, Resp>(
        &self,
        topic: &str,
        request: &Req,
        config: CircuitBreakerConfig,
    ) -> BreakerRequestBuilder<'_, Req, Resp>
    where
        Req: Message + Serialize + Clone,
        Resp: Message + DeserializeOwned + Send + 'static;
}

impl RpcBreakerExt for Context {
    fn request_with_breaker<Req, Resp>(
        &self,
        topic: &str,
        request: &Req,
        config: CircuitBreakerConfig,
    ) -> BreakerRequestBuilder<'_, Req, Resp>
    where
        Req: Message + Serialize + Clone,
        Resp: Message + DeserializeOwned + Send + 'static,
    {
        BreakerRequestBuilder {
            ctx: self,
            topic: topic.to_string(),
            request: request.clone(),
            config,
            timeout: Duration::from_secs(10),
            _phantom: std::marker::PhantomData,
        }
    }
}

/// Builder for RPC requests with circuit breaker
pub struct BreakerRequestBuilder<'a, Req, Resp> {
    ctx: &'a Context,
    topic: String,
    request: Req,
    config: CircuitBreakerConfig,
    timeout: Duration,
    _phantom: std::marker::PhantomData<Resp>,
}

impl<'a, Req, Resp> BreakerRequestBuilder<'a, Req, Resp>
where
    Req: Message + Serialize + Clone,
    Resp: Message + DeserializeOwned + Send + 'static,
{
    /// Set the request timeout
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10_core::prelude::*;
    /// # use mecha10_core::rpc_breaker::RpcBreakerExt;
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let result = ctx
    ///     .request_with_breaker("/topic", &request, Default::default())
    ///     .timeout(Duration::from_secs(5))
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }

    /// Execute the RPC request with circuit breaker protection
    ///
    /// # Returns
    ///
    /// - `Ok(Resp)` if request succeeded
    /// - `Err(CircuitBreakerOpen)` if circuit is open
    /// - `Err(...)` if request failed or timed out
    pub async fn execute(self) -> Result<Resp> {
        let registry = get_global_registry();
        let breaker = registry.get_or_create(&self.topic, self.config).await;

        let ctx = self.ctx;
        let topic = self.topic.clone();
        let request = self.request.clone();
        let timeout = self.timeout;

        breaker
            .call(|| async move { ctx.request(&topic, &request).timeout(timeout).execute().await })
            .await
    }
}

use once_cell::sync::Lazy;

static GLOBAL_REGISTRY: Lazy<CircuitBreakerRegistry> = Lazy::new(CircuitBreakerRegistry::new);

/// Get the global circuit breaker registry
pub fn get_global_registry() -> &'static CircuitBreakerRegistry {
    &GLOBAL_REGISTRY
}

/// Get statistics for all circuit breakers
pub async fn get_all_stats() -> HashMap<String, CircuitBreakerStats> {
    GLOBAL_REGISTRY.stats().await
}

/// Reset all circuit breakers (useful for testing)
pub async fn reset_all_breakers() {
    GLOBAL_REGISTRY.reset_all().await;
}

impl CircuitBreakerConfig {
    /// Shared constructor behind the `critical`/`standard`/`resilient` presets, which only
    /// differ in threshold/timeout values.
    fn preset(name: &str, failure_threshold: u32, success_threshold: u32, timeout: Duration) -> Self {
        Self {
            failure_threshold,
            success_threshold,
            timeout,
            name: Some(name.to_string()),
        }
    }

    // duplicate: false positive — `critical`/`standard`/`resilient` already delegate their
    // shared shape to `preset` above; each is just a distinct literal (threshold/timeout)
    // tuple, which is the whole point of having named presets.
    /// Configuration for critical services (aggressive circuit breaking)
    ///
    /// - Opens after 3 failures
    /// - Requires 3 successes to close
    /// - 60 second timeout
    pub fn critical(name: &str) -> Self {
        Self::preset(name, 3, 3, Duration::from_secs(60))
    }

    /// Configuration for standard services (balanced)
    ///
    /// - Opens after 5 failures
    /// - Requires 2 successes to close
    /// - 30 second timeout
    pub fn standard(name: &str) -> Self {
        Self::preset(name, 5, 2, Duration::from_secs(30))
    }

    /// Configuration for resilient services (lenient circuit breaking)
    ///
    /// - Opens after 10 failures
    /// - Requires 2 successes to close
    /// - 15 second timeout
    pub fn resilient(name: &str) -> Self {
        Self::preset(name, 10, 2, Duration::from_secs(15))
    }
}