//! Circuit breaker e2e: provider that fails N times in a row trips
//! the breaker, the executor skips it without invoking on the next
//! call, and the chain falls through to the next provider in line.
//!
//! Phase voice-plugin-extract Step 59 — re-enables once breakers
//! land in fallback::execute.

use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use nexo_plugin_voice::breaker::{BreakerRegistry, BreakerSettings};
use nexo_plugin_voice::error::{ReasonCode, Result, VoiceError};
use nexo_plugin_voice::fallback::{execute, FallbackConfig, FallbackStrategyKind, RetryPolicy};
use nexo_plugin_voice::providers::{
    AudioFormat, CostClass, SynthesizeOutput, SynthesizeRequest, TtsProvider,
};

struct ConfigurableProvider {
    id_str: &'static str,
    fail_n_times: AtomicU32,
    total_calls: Arc<AtomicU32>,
    success_bytes: &'static [u8],
}

impl ConfigurableProvider {
    fn new(id: &'static str, fail_count: u32, success_bytes: &'static [u8]) -> Self {
        Self {
            id_str: id,
            fail_n_times: AtomicU32::new(fail_count),
            total_calls: Arc::new(AtomicU32::new(0)),
            success_bytes,
        }
    }

    fn call_count(&self) -> u32 {
        self.total_calls.load(Ordering::SeqCst)
    }
}

#[async_trait]
impl TtsProvider for ConfigurableProvider {
    fn id(&self) -> &'static str {
        self.id_str
    }
    fn label(&self) -> &str {
        self.id_str
    }
    fn cost_class(&self) -> CostClass {
        CostClass::Free
    }
    fn is_configured(&self) -> bool {
        true
    }
    async fn synthesize(&self, _req: SynthesizeRequest<'_>) -> Result<SynthesizeOutput> {
        self.total_calls.fetch_add(1, Ordering::SeqCst);
        let remaining = self.fail_n_times.load(Ordering::SeqCst);
        if remaining > 0 {
            self.fail_n_times.fetch_sub(1, Ordering::SeqCst);
            return Err(VoiceError::Provider {
                provider: self.id_str,
                reason: "configured to fail".into(),
                code: ReasonCode::EmptyAudio,
            });
        }
        Ok(SynthesizeOutput {
            audio_bytes: self.success_bytes.to_vec(),
            raw_mime: "audio/mpeg",
            latency_ms: 50,
            reason_code: ReasonCode::Success,
        })
    }
}

fn req<'a>() -> SynthesizeRequest<'a> {
    SynthesizeRequest {
        body_ssml: "hola mundo",
        voice_id: "es-MX-DaliaNeural",
        speed: None,
        pitch_pct: None,
        style: None,
        model_id: None,
        format: AudioFormat::Mp3_24k,
    }
}

#[tokio::test]
async fn breaker_trips_after_failure_threshold_then_skips_provider() {
    // Edge fails 5 times → trip (failure_threshold=2 for test speed)
    // → 6th attempt must NOT invoke edge, falls through to backup.
    let edge = Arc::new(ConfigurableProvider::new("edge", 100, b""));
    let backup = Arc::new(ConfigurableProvider::new("elevenlabs", 0, b"backup_audio"));
    let edge_h = edge.clone();
    let backup_h = backup.clone();
    let providers: Vec<Arc<dyn TtsProvider>> = vec![edge, backup];

    // Tight threshold so test runs fast.
    let breakers = BreakerRegistry::new(BreakerSettings {
        failure_threshold: 2,
        success_threshold: 1,
        initial_backoff_seconds: 60, // long enough we don't re-enter half-open mid-test
        max_backoff_seconds: 120,
    });

    // max_attempts = 1 so the circuit-breaker count assertions
    // observe exactly one call per execute() invocation.
    let mut cfg = FallbackConfig::default();
    cfg.retry = RetryPolicy {
        max_attempts: 1,
        ..RetryPolicy::default()
    };
    cfg.strategy = FallbackStrategyKind::Chain;
    cfg.chain = vec!["edge".into(), "elevenlabs".into()];

    // Call 1: edge fails (first failure), backup succeeds.
    let o1 = execute(&providers, &cfg, &breakers, &req()).await.unwrap();
    assert_eq!(o1.winner, "elevenlabs");
    assert_eq!(edge_h.call_count(), 1);

    // Call 2: edge fails (second failure → trip), backup succeeds.
    let o2 = execute(&providers, &cfg, &breakers, &req()).await.unwrap();
    assert_eq!(o2.winner, "elevenlabs");
    assert_eq!(edge_h.call_count(), 2);
    assert!(
        breakers.is_open("edge"),
        "after 2 failures with threshold=2, breaker MUST be open"
    );

    // Call 3: edge breaker open → skipped without invocation,
    // backup serves. edge call_count must NOT increment.
    let o3 = execute(&providers, &cfg, &breakers, &req()).await.unwrap();
    assert_eq!(o3.winner, "elevenlabs");
    assert_eq!(
        edge_h.call_count(),
        2,
        "edge must NOT be invoked while breaker is open"
    );
    assert!(o3.chain_tried.contains(&"elevenlabs"));
    assert_eq!(backup_h.call_count(), 3);
}

#[tokio::test]
async fn breaker_half_open_probes_recovery() {
    // Threshold=1, short backoff so we can observe the
    // Open → HalfOpen → Closed transition in a single test.
    let edge = Arc::new(ConfigurableProvider::new("edge", 1, b"edge_recovered"));
    let backup = Arc::new(ConfigurableProvider::new("elevenlabs", 0, b"backup_audio"));
    let edge_h = edge.clone();
    let providers: Vec<Arc<dyn TtsProvider>> = vec![edge, backup];

    let breakers = BreakerRegistry::new(BreakerSettings {
        failure_threshold: 1,
        success_threshold: 1,
        initial_backoff_seconds: 1, // short for test
        max_backoff_seconds: 1,
    });

    // max_attempts = 1 so the circuit-breaker count assertions
    // observe exactly one call per execute() invocation.
    let mut cfg = FallbackConfig::default();
    cfg.retry = RetryPolicy {
        max_attempts: 1,
        ..RetryPolicy::default()
    };
    cfg.strategy = FallbackStrategyKind::Chain;
    cfg.chain = vec!["edge".into(), "elevenlabs".into()];

    // Call 1: edge fails → breaker trips → backup wins.
    let _ = execute(&providers, &cfg, &breakers, &req()).await.unwrap();
    assert!(breakers.is_open("edge"));
    assert_eq!(edge_h.call_count(), 1);

    // Wait > backoff so the next allow() transitions Open → HalfOpen.
    tokio::time::sleep(Duration::from_millis(1100)).await;

    // Call 2: edge breaker half-open, edge now succeeds (configured
    // to fail only once) → breaker closes again.
    let o = execute(&providers, &cfg, &breakers, &req()).await.unwrap();
    assert_eq!(o.winner, "edge");
    assert_eq!(edge_h.call_count(), 2);
    assert!(
        !breakers.is_open("edge"),
        "successful probe in HalfOpen must close the breaker"
    );
}

#[tokio::test]
async fn non_retryable_errors_do_not_trip_breaker() {
    // Provider returns ReasonCode::Unauthorized (non-retryable).
    // The chain executor short-circuits the whole call without
    // marking the breaker as failed — Unauthorized is a config
    // bug, not a transient flake.
    struct AuthFailProvider;

    #[async_trait]
    impl TtsProvider for AuthFailProvider {
        fn id(&self) -> &'static str {
            "edge"
        }
        fn label(&self) -> &str {
            "edge"
        }
        fn cost_class(&self) -> CostClass {
            CostClass::Free
        }
        fn is_configured(&self) -> bool {
            true
        }
        async fn synthesize(&self, _req: SynthesizeRequest<'_>) -> Result<SynthesizeOutput> {
            Err(VoiceError::Provider {
                provider: "edge",
                reason: "401 unauthorized".into(),
                code: ReasonCode::Unauthorized,
            })
        }
    }

    let providers: Vec<Arc<dyn TtsProvider>> = vec![Arc::new(AuthFailProvider)];
    let breakers = BreakerRegistry::new(BreakerSettings {
        failure_threshold: 1,
        success_threshold: 1,
        initial_backoff_seconds: 60,
        max_backoff_seconds: 60,
    });

    let mut cfg = FallbackConfig::default();
    cfg.retry = RetryPolicy {
        max_attempts: 1,
        ..RetryPolicy::default()
    };
    cfg.chain = vec!["edge".into()];

    let _ = execute(&providers, &cfg, &breakers, &req()).await.unwrap_err();
    assert!(
        !breakers.is_open("edge"),
        "Unauthorized must NOT trip the breaker (non-retryable, breaker is for transient flakes)"
    );
}
