//! Fallback strategies for the multi-provider synth pipeline.
//!
//! Operator configures which strategy to apply when the primary
//! provider fails. Four strategies cover the latency/cost trade-off
//! space:
//!
//! - **Chain** — try primary, then each fallback in order until
//!   one succeeds. Default. Conservative on cost.
//! - **ParallelFirstWins** — race all providers; first audio wins,
//!   cancel the rest. Lowest latency, highest cost.
//! - **ParallelCheapestWins** — race providers whose cost class
//!   ≤ max_cost; pick the cheapest success in the race window.
//! - **TextOnly** — skip synthesis entirely on primary failure;
//!   return plain text with markers stripped.

use std::sync::Arc;

use serde::{Deserialize, Serialize};
use tokio::time::Duration;

use crate::breaker::BreakerRegistry;
use crate::error::{ReasonCode, Result, VoiceError};
use crate::providers::{CostClass, SynthesizeOutput, SynthesizeRequest, TtsProvider};

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum FallbackStrategyKind {
    Chain,
    ParallelFirstWins,
    ParallelCheapestWins,
    TextOnly,
}

impl Default for FallbackStrategyKind {
    fn default() -> Self {
        Self::Chain
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct FallbackConfig {
    pub strategy: FallbackStrategyKind,
    /// Provider id order for [`FallbackStrategyKind::Chain`].
    /// First entry is the primary; remaining are fallbacks tried
    /// on transient failure.
    pub chain: Vec<String>,
    /// Maximum cost class to consider in
    /// [`FallbackStrategyKind::ParallelCheapestWins`]. Ignored by
    /// other strategies.
    pub max_cost: CostClassFilter,
    /// Same-provider retry policy. The Chain strategy retries each
    /// provider up to `max_attempts` with exponential backoff
    /// before advancing to the next provider in the chain.
    /// SDK-parity: predecessor microapp retried Edge 3 times with
    /// 0/250/600ms before falling back to text.
    pub retry: RetryPolicy,
}

impl Default for FallbackConfig {
    fn default() -> Self {
        Self {
            strategy: FallbackStrategyKind::Chain,
            chain: vec!["edge".to_string()],
            max_cost: CostClassFilter::Free,
            retry: RetryPolicy::default(),
        }
    }
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum CostClassFilter {
    Free,
    Cheap,
    Premium,
}

impl CostClassFilter {
    /// Returns `true` when `class` is at or below this filter.
    pub fn admits(self, class: CostClass) -> bool {
        match (self, class) {
            (Self::Premium, _) => true,
            (Self::Cheap, CostClass::Premium) => false,
            (Self::Cheap, _) => true,
            (Self::Free, CostClass::Free) => true,
            (Self::Free, _) => false,
        }
    }
}

/// Successful synth + which providers were tried in order. The
/// pipeline uses this to populate audit events + metrics.
#[derive(Debug, Clone)]
pub struct FallbackOutcome {
    pub output: SynthesizeOutput,
    pub winner: &'static str,
    pub chain_tried: Vec<&'static str>,
}

/// Execute `req` against the configured chain. Borrows the
/// providers vec to avoid cloning them per call.
///
/// `breakers` lets each provider's circuit breaker veto the call
/// without invoking the endpoint. Pass an empty registry for tests
/// that don't care about breaker behaviour — every `allow()` will
/// return `true` against a freshly-created breaker.
pub async fn execute(
    providers: &[Arc<dyn TtsProvider>],
    config: &FallbackConfig,
    breakers: &BreakerRegistry,
    req: &SynthesizeRequest<'_>,
) -> Result<FallbackOutcome> {
    if providers.is_empty() {
        return Err(VoiceError::UnknownProvider("(no providers registered)".into()));
    }

    match config.strategy {
        FallbackStrategyKind::TextOnly => Err(VoiceError::NotSupported("text_only strategy")),
        FallbackStrategyKind::Chain => {
            execute_chain(providers, &config.chain, &config.retry, breakers, req).await
        }
        FallbackStrategyKind::ParallelFirstWins => {
            execute_parallel_first_wins(providers, breakers, req).await
        }
        FallbackStrategyKind::ParallelCheapestWins => {
            execute_parallel_cheapest_wins(providers, config.max_cost, breakers, req).await
        }
    }
}

async fn execute_chain(
    providers: &[Arc<dyn TtsProvider>],
    order: &[String],
    retry: &RetryPolicy,
    breakers: &BreakerRegistry,
    req: &SynthesizeRequest<'_>,
) -> Result<FallbackOutcome> {
    let mut chain_tried: Vec<&'static str> = Vec::new();
    let mut last_err: Option<VoiceError> = None;

    for pid in order {
        let Some(p) = providers.iter().find(|p| p.id() == pid || p.aliases().contains(&pid.as_str()))
        else {
            tracing::warn!(provider = %pid, "voice/fallback: provider not registered, skipping");
            continue;
        };
        if !p.is_configured() {
            chain_tried.push(p.id());
            continue;
        }
        chain_tried.push(p.id());

        // Circuit breaker — atomically transitions Open → HalfOpen
        // when the backoff expires, so a single rejected call won't
        // permanently lock a recovering provider out.
        let breaker = breakers.get(p.id());
        if !breaker.allow() {
            tracing::info!(provider = %p.id(), "voice/fallback: breaker open, skipping");
            last_err = Some(VoiceError::CircuitOpen(p.id().to_string()));
            continue;
        }

        // Same-provider retry loop with exponential backoff before
        // moving to the next provider in the chain. SDK-parity gap 3:
        // microapp predecessor retried Edge 3 times with 0/250/600ms
        // before falling back. Non-retryable errors short-circuit
        // the entire chain (config bugs don't get retried).
        let mut provider_err: Option<VoiceError> = None;
        for attempt in 1..=retry.max_attempts.max(1) {
            let delay = retry.backoff_for(attempt);
            if !delay.is_zero() {
                tokio::time::sleep(delay).await;
            }
            match p.synthesize(borrow_req(req)).await {
                Ok(output) => {
                    breaker.on_success();
                    if attempt > 1 {
                        tracing::info!(
                            provider = %p.id(),
                            attempt,
                            "voice/fallback: synth recovered after retry"
                        );
                    }
                    return Ok(FallbackOutcome {
                        output,
                        winner: p.id(),
                        chain_tried,
                    });
                }
                Err(e) => {
                    let code = e.reason_code();
                    tracing::warn!(
                        provider = %p.id(),
                        attempt,
                        max = retry.max_attempts,
                        reason = ?code,
                        err = %e,
                        "voice/fallback: provider attempt failed"
                    );
                    if !code.is_retryable() {
                        // Config bug — short-circuit whole chain.
                        return Err(e);
                    }
                    breaker.on_failure();
                    provider_err = Some(e);
                    // Continue to next attempt against same provider.
                }
            }
        }
        // All attempts against this provider failed; carry the last
        // error forward and advance to the next provider.
        if let Some(e) = provider_err {
            last_err = Some(e);
        }
    }
    Err(last_err.unwrap_or(VoiceError::UnknownProvider(
        "no configured provider in chain".into(),
    )))
}

async fn execute_parallel_first_wins(
    providers: &[Arc<dyn TtsProvider>],
    breakers: &BreakerRegistry,
    req: &SynthesizeRequest<'_>,
) -> Result<FallbackOutcome> {
    let configured: Vec<Arc<dyn TtsProvider>> = providers
        .iter()
        .filter(|p| p.is_configured() && !breakers.is_open(p.id()))
        .cloned()
        .collect();
    if configured.is_empty() {
        return Err(VoiceError::UnknownProvider("no configured providers".into()));
    }
    let tried: Vec<&'static str> = configured.iter().map(|p| p.id()).collect();

    // Spawn each synth as a tokio task; first Ok wins, drop the rest.
    let mut set = tokio::task::JoinSet::new();
    for p in configured {
        let body = req.body_ssml.to_string();
        let voice = req.voice_id.to_string();
        let style = req.style.map(|s| s.to_string());
        let model_id = req.model_id.map(|s| s.to_string());
        let speed = req.speed;
        let pitch = req.pitch_pct;
        let fmt = req.format;
        set.spawn(async move {
            let r = SynthesizeRequest {
                body_ssml: &body,
                voice_id: &voice,
                speed,
                pitch_pct: pitch,
                style: style.as_deref(),
                model_id: model_id.as_deref(),
                format: fmt,
            };
            let id = p.id();
            let res = p.synthesize(r).await;
            (id, res)
        });
    }

    let mut last_err: Option<VoiceError> = None;
    while let Some(joined) = set.join_next().await {
        match joined {
            Ok((id, Ok(output))) => {
                set.abort_all();
                return Ok(FallbackOutcome {
                    output,
                    winner: id,
                    chain_tried: tried,
                });
            }
            Ok((_id, Err(e))) => last_err = Some(e),
            Err(e) => last_err = Some(VoiceError::Other(anyhow::anyhow!(e))),
        }
    }
    Err(last_err.unwrap_or(VoiceError::Provider {
        provider: "parallel",
        reason: "race yielded no winners".into(),
        code: ReasonCode::EmptyAudio,
    }))
}

async fn execute_parallel_cheapest_wins(
    providers: &[Arc<dyn TtsProvider>],
    max_cost: CostClassFilter,
    breakers: &BreakerRegistry,
    req: &SynthesizeRequest<'_>,
) -> Result<FallbackOutcome> {
    let admitted: Vec<Arc<dyn TtsProvider>> = providers
        .iter()
        .filter(|p| {
            p.is_configured() && max_cost.admits(p.cost_class()) && !breakers.is_open(p.id())
        })
        .cloned()
        .collect();
    if admitted.is_empty() {
        return Err(VoiceError::UnknownProvider(
            "no providers under max_cost".into(),
        ));
    }
    // Race only the admitted set; otherwise identical to
    // parallel-first-wins.
    let tried: Vec<&'static str> = admitted.iter().map(|p| p.id()).collect();
    let mut set = tokio::task::JoinSet::new();
    for p in admitted {
        let body = req.body_ssml.to_string();
        let voice = req.voice_id.to_string();
        let style = req.style.map(|s| s.to_string());
        let model_id = req.model_id.map(|s| s.to_string());
        let speed = req.speed;
        let pitch = req.pitch_pct;
        let fmt = req.format;
        set.spawn(async move {
            let r = SynthesizeRequest {
                body_ssml: &body,
                voice_id: &voice,
                speed,
                pitch_pct: pitch,
                style: style.as_deref(),
                model_id: model_id.as_deref(),
                format: fmt,
            };
            let id = p.id();
            let res = p.synthesize(r).await;
            (id, res)
        });
    }
    let mut last_err: Option<VoiceError> = None;
    while let Some(joined) = set.join_next().await {
        match joined {
            Ok((id, Ok(output))) => {
                set.abort_all();
                return Ok(FallbackOutcome {
                    output,
                    winner: id,
                    chain_tried: tried,
                });
            }
            Ok((_id, Err(e))) => last_err = Some(e),
            Err(e) => last_err = Some(VoiceError::Other(anyhow::anyhow!(e))),
        }
    }
    Err(last_err.unwrap_or(VoiceError::Provider {
        provider: "parallel_cheapest",
        reason: "race yielded no winners".into(),
        code: ReasonCode::EmptyAudio,
    }))
}

/// Internal helper: rebuild a borrow-only `SynthesizeRequest` for
/// the chain executor (which keeps borrows alive through awaits
/// of the same task — no cross-task move needed).
fn borrow_req<'a>(r: &'a SynthesizeRequest<'a>) -> SynthesizeRequest<'a> {
    SynthesizeRequest {
        body_ssml: r.body_ssml,
        voice_id: r.voice_id,
        speed: r.speed,
        pitch_pct: r.pitch_pct,
        style: r.style,
        model_id: r.model_id,
        format: r.format,
    }
}

/// Per-provider retry policy. Honoured inside the chain executor
/// for retryable [`ReasonCode`]s; ignored by parallel strategies.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct RetryPolicy {
    pub max_attempts: u32,
    pub backoff_ms_initial: u64,
    pub backoff_ms_max: u64,
    pub backoff_multiplier: f64,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            max_attempts: 3,
            backoff_ms_initial: 250,
            backoff_ms_max: 2000,
            backoff_multiplier: 2.4,
        }
    }
}

impl RetryPolicy {
    /// Backoff duration before attempt `n` (1-indexed). Caps at
    /// `backoff_ms_max`.
    pub fn backoff_for(&self, attempt: u32) -> Duration {
        if attempt <= 1 {
            return Duration::ZERO;
        }
        let mut ms = self.backoff_ms_initial as f64;
        for _ in 2..attempt {
            ms *= self.backoff_multiplier;
        }
        Duration::from_millis(ms.min(self.backoff_ms_max as f64) as u64)
    }
}

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

    #[test]
    fn cost_class_filter_admits_correctly() {
        assert!(CostClassFilter::Free.admits(CostClass::Free));
        assert!(!CostClassFilter::Free.admits(CostClass::Cheap));
        assert!(!CostClassFilter::Free.admits(CostClass::Premium));

        assert!(CostClassFilter::Cheap.admits(CostClass::Free));
        assert!(CostClassFilter::Cheap.admits(CostClass::Cheap));
        assert!(!CostClassFilter::Cheap.admits(CostClass::Premium));

        assert!(CostClassFilter::Premium.admits(CostClass::Free));
        assert!(CostClassFilter::Premium.admits(CostClass::Cheap));
        assert!(CostClassFilter::Premium.admits(CostClass::Premium));
    }

    #[test]
    fn retry_backoff_grows_exponentially_with_cap() {
        let p = RetryPolicy {
            max_attempts: 5,
            backoff_ms_initial: 100,
            backoff_ms_max: 1000,
            backoff_multiplier: 2.0,
        };
        assert_eq!(p.backoff_for(1), Duration::ZERO);
        assert_eq!(p.backoff_for(2), Duration::from_millis(100));
        assert_eq!(p.backoff_for(3), Duration::from_millis(200));
        assert_eq!(p.backoff_for(4), Duration::from_millis(400));
        assert_eq!(p.backoff_for(5), Duration::from_millis(800));
        // Cap kicks in.
        assert_eq!(p.backoff_for(10), Duration::from_millis(1000));
    }

    #[test]
    fn default_config_chain_strategy_with_edge_only() {
        let c = FallbackConfig::default();
        assert_eq!(c.strategy, FallbackStrategyKind::Chain);
        assert_eq!(c.chain, vec!["edge".to_string()]);
        assert_eq!(c.max_cost, CostClassFilter::Free);
    }
}
