//! Error Handling and Recovery for Broker Gateway Service
//!
//! Implements comprehensive error recovery strategies including:
//! - Exponential backoff with retry limits
//! - Circuit breaker pattern for fault isolation
//! - Dead letter queue for unrecoverable orders
//! - Error classification and routing

pub use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitBreakerState};
use std::collections::VecDeque;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tracing::{error, info, warn};

/// Maximum number of retry attempts before giving up
const MAX_RETRY_ATTEMPTS: u32 = 3;

/// Base delay for exponential backoff (100ms)
const BASE_BACKOFF_MS: u64 = 100;

/// Circuit breaker failure threshold (5 failures triggers OPEN state)
const CIRCUIT_BREAKER_THRESHOLD: usize = 5;

/// Circuit breaker timeout duration (60 seconds in OPEN state before HALF_OPEN)
#[cfg(not(test))]
const CIRCUIT_BREAKER_TIMEOUT: Duration = Duration::from_secs(60);
#[cfg(test)]
const CIRCUIT_BREAKER_TIMEOUT: Duration = Duration::from_millis(100);

/// Maximum dead letter queue size
const MAX_DLQ_SIZE: usize = 10_000;

/// Error recovery strategy determines how to handle failures
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ErrorRecoveryStrategy {
    /// Retry with exponential backoff (transient network errors)
    Retry,
    /// Activate circuit breaker (cascading failures)
    CircuitBreak,
    /// Fail immediately without retry (validation errors)
    FailFast,
    /// Use fallback mechanism (degraded mode)
    Fallback,
}

/// Dead letter queue entry for unrecoverable orders
#[derive(Debug, Clone)]
pub struct DeadLetterEntry {
    /// Client order ID
    pub client_order_id: String,
    /// Error message
    pub error: String,
    /// Timestamp when added to DLQ
    pub timestamp: Instant,
    /// Number of retry attempts before giving up
    pub retry_attempts: u32,
}

/// Dead letter queue for unrecoverable orders
pub struct DeadLetterQueue {
    /// Queue of failed orders
    entries: Arc<RwLock<VecDeque<DeadLetterEntry>>>,
}

impl DeadLetterQueue {
    /// Create a new dead letter queue
    pub fn new() -> Self {
        Self {
            entries: Arc::new(RwLock::new(VecDeque::new())),
        }
    }

    /// Add an entry to the dead letter queue
    pub async fn add(&self, entry: DeadLetterEntry) {
        let mut entries = self.entries.write().await;

        if entries.len() >= MAX_DLQ_SIZE {
            warn!(
                "Dead letter queue full ({} entries), removing oldest entry",
                MAX_DLQ_SIZE
            );
            entries.pop_front();
        }

        error!(
            "Order {} added to DLQ after {} retries: {}",
            entry.client_order_id, entry.retry_attempts, entry.error
        );

        entries.push_back(entry);
    }

    /// Get all entries in the dead letter queue
    pub async fn get_all(&self) -> Vec<DeadLetterEntry> {
        let entries = self.entries.read().await;
        entries.iter().cloned().collect()
    }

    /// Get the number of entries in the dead letter queue
    pub async fn len(&self) -> usize {
        let entries = self.entries.read().await;
        entries.len()
    }

    /// Check if the dead letter queue is empty
    pub async fn is_empty(&self) -> bool {
        let entries = self.entries.read().await;
        entries.is_empty()
    }

    /// Remove an entry by client_order_id (returns true if found)
    pub async fn remove(&self, client_order_id: &str) -> bool {
        let mut entries = self.entries.write().await;
        if let Some(pos) = entries.iter().position(|e| e.client_order_id == client_order_id) {
            entries.remove(pos);
            info!("Removed order {} from DLQ", client_order_id);
            true
        } else {
            false
        }
    }

    /// Clear all entries from the dead letter queue
    pub async fn clear(&self) {
        let mut entries = self.entries.write().await;
        let count = entries.len();
        entries.clear();
        info!("Cleared {} entries from DLQ", count);
    }
}

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

/// Error handler with retry logic and circuit breaker
pub struct ErrorHandler {
    /// Circuit breaker for fault isolation
    circuit_breaker: CircuitBreaker,
    /// Dead letter queue for unrecoverable orders
    dead_letter_queue: DeadLetterQueue,
}

impl ErrorHandler {
    /// Create a new error handler
    pub fn new() -> Self {
        let cb_config = CircuitBreakerConfig {
            failure_threshold: CIRCUIT_BREAKER_THRESHOLD as u32,
            timeout: CIRCUIT_BREAKER_TIMEOUT,
            success_threshold: 1,
        };
        Self {
            circuit_breaker: CircuitBreaker::new("broker_gateway", cb_config),
            dead_letter_queue: DeadLetterQueue::new(),
        }
    }

    /// Determine error recovery strategy based on error type
    pub fn classify_error(&self, error: &str) -> ErrorRecoveryStrategy {
        let error_lower = error.to_lowercase();

        // Validation errors - fail fast
        if error_lower.contains("invalid")
            || error_lower.contains("validation")
            || error_lower.contains("bad request")
        {
            return ErrorRecoveryStrategy::FailFast;
        }

        // Database errors - circuit break to prevent cascading failures
        // Check before connection errors since "database connection" contains "connection"
        if error_lower.contains("database")
            || error_lower.contains("postgres")
            || error_lower.contains("sql")
        {
            return ErrorRecoveryStrategy::CircuitBreak;
        }

        // Network/timeout errors - retry with backoff
        if error_lower.contains("timeout")
            || error_lower.contains("network")
            || error_lower.contains("connection")
            || error_lower.contains("unavailable")
        {
            return ErrorRecoveryStrategy::Retry;
        }

        // Default to retry for unknown errors
        ErrorRecoveryStrategy::Retry
    }

    /// Calculate exponential backoff delay
    /// Formula: base_delay * 2^attempt (capped at 25.6 seconds)
    pub fn calculate_backoff(&self, attempt: u32) -> Duration {
        let delay_ms = BASE_BACKOFF_MS * 2_u64.pow(attempt.min(8));
        let capped_delay = delay_ms.min(25_600); // 25.6 seconds max
        Duration::from_millis(capped_delay)
    }

    /// Execute operation with retry logic
    pub async fn retry_with_backoff<F, T, E>(
        &self,
        operation: F,
        operation_name: &str,
    ) -> Result<T, E>
    where
        F: Fn() -> Result<T, E>,
        E: std::fmt::Display,
    {
        let mut last_error: Option<E> = None;

        for attempt in 0..MAX_RETRY_ATTEMPTS {
            match operation() {
                Ok(result) => {
                    if attempt > 0 {
                        info!(
                            "Operation '{}' succeeded after {} retries",
                            operation_name, attempt
                        );
                    }
                    return Ok(result);
                }
                Err(e) => {
                    if attempt < MAX_RETRY_ATTEMPTS - 1 {
                        let backoff = self.calculate_backoff(attempt);
                        warn!(
                            "Operation '{}' failed (attempt {}/{}): {}. Retrying in {:?}",
                            operation_name,
                            attempt + 1,
                            MAX_RETRY_ATTEMPTS,
                            e,
                            backoff
                        );
                        tokio::time::sleep(backoff).await;
                    } else {
                        error!(
                            "Operation '{}' failed after {} attempts: {}",
                            operation_name, MAX_RETRY_ATTEMPTS, e
                        );
                    }
                    last_error = Some(e);
                }
            }
        }

        // Handle edge case where MAX_RETRY_ATTEMPTS is 0 or loop completes
        // without an early return (last attempt falls through).
        match last_error {
            Some(e) => Err(e),
            None => {
                // MAX_RETRY_ATTEMPTS == 0: no iterations ran; call once to get an error
                error!(
                    "Operation '{}' had zero retry attempts configured",
                    operation_name
                );
                operation()
            }
        }
    }

    /// Get reference to circuit breaker
    pub fn circuit_breaker(&self) -> &CircuitBreaker {
        &self.circuit_breaker
    }

    /// Get reference to dead letter queue
    pub fn dead_letter_queue(&self) -> &DeadLetterQueue {
        &self.dead_letter_queue
    }

    /// Add failed order to dead letter queue
    pub async fn send_to_dlq(
        &self,
        client_order_id: String,
        error: String,
        retry_attempts: u32,
    ) {
        let entry = DeadLetterEntry {
            client_order_id,
            error,
            timestamp: Instant::now(),
            retry_attempts,
        };

        self.dead_letter_queue.add(entry).await;
    }
}

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

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

    fn test_cb_config() -> CircuitBreakerConfig {
        CircuitBreakerConfig {
            failure_threshold: CIRCUIT_BREAKER_THRESHOLD as u32,
            timeout: CIRCUIT_BREAKER_TIMEOUT,
            success_threshold: 1,
        }
    }

    #[tokio::test]
    async fn test_circuit_breaker_closed_to_open() {
        let cb = CircuitBreaker::new("test_broker", test_cb_config());

        // Initially CLOSED
        assert_eq!(cb.state().await, CircuitBreakerState::Closed);
        assert!(cb.can_execute().await);

        // Record failures up to threshold
        for _ in 0..CIRCUIT_BREAKER_THRESHOLD {
            cb.record_failure().await;
        }

        // Should now be OPEN
        assert_eq!(cb.state().await, CircuitBreakerState::Open);
        assert!(!cb.can_execute().await);
    }

    #[tokio::test]
    async fn test_circuit_breaker_half_open_success() {
        let cb = CircuitBreaker::new("test_broker", test_cb_config());

        // Force HALF_OPEN state
        for _ in 0..CIRCUIT_BREAKER_THRESHOLD {
            cb.record_failure().await;
        }

        // Wait for timeout to transition to HALF_OPEN
        tokio::time::sleep(CIRCUIT_BREAKER_TIMEOUT + Duration::from_millis(10)).await;
        assert!(cb.can_execute().await);
        assert_eq!(cb.state().await, CircuitBreakerState::HalfOpen);

        // Success should close circuit
        cb.record_success().await;
        assert_eq!(cb.state().await, CircuitBreakerState::Closed);
    }

    #[tokio::test]
    async fn test_circuit_breaker_half_open_failure() {
        let cb = CircuitBreaker::new("test_broker", test_cb_config());

        // Force HALF_OPEN state
        for _ in 0..CIRCUIT_BREAKER_THRESHOLD {
            cb.record_failure().await;
        }

        tokio::time::sleep(CIRCUIT_BREAKER_TIMEOUT + Duration::from_millis(10)).await;
        assert!(cb.can_execute().await); // Trigger transition to HALF_OPEN
        assert_eq!(cb.state().await, CircuitBreakerState::HalfOpen);

        // Failure should reopen circuit
        cb.record_failure().await;
        assert_eq!(cb.state().await, CircuitBreakerState::Open);
    }

    #[tokio::test]
    async fn test_dead_letter_queue_operations() {
        let dlq = DeadLetterQueue::new();

        // Initially empty
        assert!(dlq.is_empty().await);
        assert_eq!(dlq.len().await, 0);

        // Add entry
        let entry = DeadLetterEntry {
            client_order_id: "test-order-1".to_string(),
            error: "Network timeout".to_string(),
            timestamp: Instant::now(),
            retry_attempts: 3,
        };

        dlq.add(entry).await;
        assert_eq!(dlq.len().await, 1);
        assert!(!dlq.is_empty().await);

        // Get all entries
        let entries = dlq.get_all().await;
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].client_order_id, "test-order-1");

        // Remove entry
        assert!(dlq.remove("test-order-1").await);
        assert!(dlq.is_empty().await);
    }

    #[tokio::test]
    async fn test_error_classification() {
        let handler = ErrorHandler::new();

        // Validation errors
        assert_eq!(
            handler.classify_error("Invalid order quantity"),
            ErrorRecoveryStrategy::FailFast
        );

        // Network errors
        assert_eq!(
            handler.classify_error("Connection timeout"),
            ErrorRecoveryStrategy::Retry
        );

        // Database errors
        assert_eq!(
            handler.classify_error("Database connection failed"),
            ErrorRecoveryStrategy::CircuitBreak
        );
    }

    #[test]
    fn test_exponential_backoff() {
        let handler = ErrorHandler::new();

        // Test exponential growth
        assert_eq!(handler.calculate_backoff(0), Duration::from_millis(100));
        assert_eq!(handler.calculate_backoff(1), Duration::from_millis(200));
        assert_eq!(handler.calculate_backoff(2), Duration::from_millis(400));
        assert_eq!(handler.calculate_backoff(3), Duration::from_millis(800));
        assert_eq!(handler.calculate_backoff(4), Duration::from_millis(1_600));

        // Test capping at 25.6 seconds
        assert_eq!(handler.calculate_backoff(10), Duration::from_millis(25_600));
    }
}
