//! Same-provider retry-with-backoff e2e (SDK-parity gap 3).
//!
//! Asserts that `execute_chain` retries the SAME provider up to
//! `max_attempts` with exponential backoff before advancing to
//! the next provider in the chain. Mirrors the SDK predecessor's
//! triple retry against Edge (0/250/600 ms) before falling back.

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

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 FailNTimesThenSucceed {
    id_str: &'static str,
    fail_remaining: AtomicU32,
    total_calls: Arc<AtomicU32>,
    success_bytes: &'static [u8],
}

impl FailNTimesThenSucceed {
    fn new(id: &'static str, fail_n: u32, bytes: &'static [u8]) -> Self {
        Self {
            id_str: id,
            fail_remaining: AtomicU32::new(fail_n),
            total_calls: Arc::new(AtomicU32::new(0)),
            success_bytes: bytes,
        }
    }
    fn call_count(&self) -> u32 {
        self.total_calls.load(Ordering::SeqCst)
    }
}

#[async_trait]
impl TtsProvider for FailNTimesThenSucceed {
    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_remaining.load(Ordering::SeqCst);
        if remaining > 0 {
            self.fail_remaining.fetch_sub(1, Ordering::SeqCst);
            return Err(VoiceError::Provider {
                provider: self.id_str,
                reason: "test fail (retryable)".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 retry_same_provider_until_success() {
    // Edge fails twice then succeeds. With max_attempts=3 the
    // executor should keep trying edge (not advance to next
    // provider) and ultimately succeed.
    let edge = Arc::new(FailNTimesThenSucceed::new("edge", 2, b"edge_audio"));
    let backup = Arc::new(FailNTimesThenSucceed::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];

    let mut cfg = FallbackConfig::default();
    cfg.strategy = FallbackStrategyKind::Chain;
    cfg.chain = vec!["edge".into(), "elevenlabs".into()];
    cfg.retry = RetryPolicy {
        max_attempts: 3,
        backoff_ms_initial: 10, // fast for tests
        backoff_ms_max: 30,
        backoff_multiplier: 2.0,
    };

    // High breaker thresholds so retries don't trip it mid-test.
    let breakers = BreakerRegistry::new(BreakerSettings {
        failure_threshold: 100,
        success_threshold: 1,
        initial_backoff_seconds: 60,
        max_backoff_seconds: 60,
    });

    let outcome = execute(&providers, &cfg, &breakers, &req()).await.unwrap();
    assert_eq!(outcome.winner, "edge");
    assert_eq!(edge_h.call_count(), 3, "edge should be called 3 times");
    assert_eq!(
        backup_h.call_count(),
        0,
        "backup should NOT be tried — edge recovered within retry policy"
    );
}

#[tokio::test]
async fn retry_advances_to_next_provider_after_exhaustion() {
    // Edge fails 5 times. max_attempts=3 means after 3 attempts
    // executor advances to backup, which succeeds.
    let edge = Arc::new(FailNTimesThenSucceed::new("edge", 5, b""));
    let backup = Arc::new(FailNTimesThenSucceed::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];

    let mut cfg = FallbackConfig::default();
    cfg.strategy = FallbackStrategyKind::Chain;
    cfg.chain = vec!["edge".into(), "elevenlabs".into()];
    cfg.retry = RetryPolicy {
        max_attempts: 3,
        backoff_ms_initial: 10,
        backoff_ms_max: 30,
        backoff_multiplier: 2.0,
    };

    let breakers = BreakerRegistry::new(BreakerSettings {
        failure_threshold: 100,
        success_threshold: 1,
        initial_backoff_seconds: 60,
        max_backoff_seconds: 60,
    });

    let outcome = execute(&providers, &cfg, &breakers, &req()).await.unwrap();
    assert_eq!(outcome.winner, "elevenlabs");
    assert_eq!(edge_h.call_count(), 3, "edge gets exactly max_attempts tries");
    assert_eq!(backup_h.call_count(), 1, "backup serves the 4th call");
}

#[tokio::test]
async fn non_retryable_error_skips_retry_loop() {
    // Provider returns Unauthorized — NOT retryable. Even with
    // max_attempts=3, the executor must short-circuit after 1 call
    // (configuration bug, retrying repeats the same mistake).
    struct AuthFail {
        calls: Arc<AtomicU32>,
    }
    #[async_trait]
    impl TtsProvider for AuthFail {
        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, _r: SynthesizeRequest<'_>) -> Result<SynthesizeOutput> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            Err(VoiceError::Provider {
                provider: "edge",
                reason: "401".into(),
                code: ReasonCode::Unauthorized,
            })
        }
    }
    let calls = Arc::new(AtomicU32::new(0));
    let provider = Arc::new(AuthFail {
        calls: calls.clone(),
    });
    let providers: Vec<Arc<dyn TtsProvider>> = vec![provider];

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

    let breakers = BreakerRegistry::new(BreakerSettings {
        failure_threshold: 100,
        success_threshold: 1,
        initial_backoff_seconds: 60,
        max_backoff_seconds: 60,
    });

    let _ = execute(&providers, &cfg, &breakers, &req()).await.unwrap_err();
    assert_eq!(
        calls.load(Ordering::SeqCst),
        1,
        "non-retryable error must short-circuit retry loop"
    );
}

#[tokio::test]
async fn backoff_durations_grow_exponentially() {
    let p = RetryPolicy {
        max_attempts: 5,
        backoff_ms_initial: 50,
        backoff_ms_max: 500,
        backoff_multiplier: 2.0,
    };
    assert_eq!(p.backoff_for(1).as_millis(), 0);
    assert_eq!(p.backoff_for(2).as_millis(), 50);
    assert_eq!(p.backoff_for(3).as_millis(), 100);
    assert_eq!(p.backoff_for(4).as_millis(), 200);
    assert_eq!(p.backoff_for(5).as_millis(), 400);
    // Cap.
    assert_eq!(p.backoff_for(99).as_millis(), 500);
}
