//! Connection pool for Microsoft Edge WebSocket synth clients.
//!
//! Cuts the per-synth handshake (~150-300ms TCP + TLS + WS upgrade)
//! by keeping warm clients ready for reuse.
//!
//! Design:
//! - LIFO acquire so the freshest client serves first (TLS keys
//!   warmest, fewest socket-idle TCP keepalives missed).
//! - On synth success, the client is returned to the pool with
//!   its `last_used` timestamp refreshed.
//! - On synth FAILURE, the client is dropped (WS protocol may be
//!   half-broken — never return a tainted connection).
//! - TTL eviction at acquire time: any client whose `last_used` is
//!   older than `idle_ttl` is dropped before being handed out.
//! - `max_size` caps how many clients live concurrently. When the
//!   pool is at the cap AND empty (every client checked out), the
//!   caller falls back to a fresh `connect()` instead of blocking.
//!
//! Caveat: msedge-tts's WS protocol may not support multi-turn over
//! a single connection in every case. The pool is **opportunistic**:
//! on the first reuse failure, the client is dropped and a fresh
//! one is built. Worst case = no speedup; best case = ~200ms p50.

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use msedge_tts::tts::client::MSEdgeTTSClient;

use crate::error::{ReasonCode, Result, VoiceError};

/// Operator-tunable pool settings. Defaults: 4 conns max, 5 min
/// idle TTL, mirrored from the SDK predecessor's behaviour.
#[derive(Debug, Clone)]
pub struct EdgeWsPoolConfig {
    pub max_size: usize,
    pub idle_ttl: Duration,
}

impl Default for EdgeWsPoolConfig {
    fn default() -> Self {
        Self {
            max_size: 4,
            idle_ttl: Duration::from_secs(5 * 60),
        }
    }
}

/// Owned wrapper around a pooled client. The pool tracks the
/// `last_used` timestamp internally; `into_inner` hands the raw
/// client to the caller for the synthesise call.
pub struct PooledClient {
    inner: MSEdgeTTSClient<std::net::TcpStream>,
}

impl std::fmt::Debug for PooledClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PooledClient").finish_non_exhaustive()
    }
}

impl PooledClient {
    pub fn into_inner(self) -> MSEdgeTTSClient<std::net::TcpStream> {
        self.inner
    }
}

struct PoolEntry {
    client: MSEdgeTTSClient<std::net::TcpStream>,
    last_used: Instant,
}

impl std::fmt::Debug for PoolEntry {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // MSEdgeTTSClient doesn't impl Debug upstream; surface
        // just the timestamp so operator logs stay useful.
        f.debug_struct("PoolEntry")
            .field("last_used_ago_secs", &self.last_used.elapsed().as_secs())
            .finish_non_exhaustive()
    }
}

/// Cloneable pool handle. Clones share state via Arc.
#[derive(Clone)]
pub struct EdgeWsPool {
    inner: Arc<Mutex<VecDeque<PoolEntry>>>,
    config: EdgeWsPoolConfig,
    in_use: Arc<std::sync::atomic::AtomicUsize>,
}

impl std::fmt::Debug for EdgeWsPool {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let idle = self.inner.lock().map(|g| g.len()).unwrap_or(0);
        f.debug_struct("EdgeWsPool")
            .field("max_size", &self.config.max_size)
            .field("idle_ttl_secs", &self.config.idle_ttl.as_secs())
            .field("idle", &idle)
            .field(
                "in_use",
                &self.in_use.load(std::sync::atomic::Ordering::SeqCst),
            )
            .finish()
    }
}

impl EdgeWsPool {
    pub fn new(config: EdgeWsPoolConfig) -> Self {
        Self {
            inner: Arc::new(Mutex::new(VecDeque::new())),
            config,
            in_use: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
        }
    }

    /// Attempt to acquire a warm client. Returns:
    /// - `Ok(PooledClient)` when an idle warm client exists OR
    ///   `total_in_use < max_size` (in which case a fresh client
    ///   is connected on the spot).
    /// - `Err(...)` when `total_in_use >= max_size` and the idle
    ///   queue is empty. Caller falls back to `fresh_client()`
    ///   (which bypasses the cap — emergency valve so a flood
    ///   doesn't hang the synth path).
    pub fn acquire(&self) -> Result<PooledClient> {
        // 1. Drain expired idle entries.
        let now = Instant::now();
        {
            let mut g = self.inner.lock().map_err(poisoned_mutex)?;
            while let Some(front) = g.front() {
                if now.duration_since(front.last_used) > self.config.idle_ttl {
                    g.pop_front();
                } else {
                    break;
                }
            }
            // 2. LIFO: pop newest off the back if any.
            if let Some(entry) = g.pop_back() {
                self.in_use
                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                return Ok(PooledClient {
                    inner: entry.client,
                });
            }
        }

        // 3. Pool empty. Spawn a fresh connect IFF under cap.
        let current_in_use = self.in_use.load(std::sync::atomic::Ordering::SeqCst);
        if current_in_use >= self.config.max_size {
            return Err(VoiceError::Provider {
                provider: "edge",
                reason: format!(
                    "pool at cap ({} in use, max {})",
                    current_in_use, self.config.max_size
                ),
                code: ReasonCode::NetworkTimeout,
            });
        }

        self.in_use
            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        match msedge_tts::tts::client::connect() {
            Ok(client) => Ok(PooledClient { inner: client }),
            Err(e) => {
                self.in_use
                    .fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
                Err(VoiceError::Provider {
                    provider: "edge",
                    reason: format!("connect: {e}"),
                    code: ReasonCode::WsClosed,
                })
            }
        }
    }

    /// Emergency-valve: connect a fresh client bypassing the cap.
    /// Used by the caller when `acquire()` returned a cap error
    /// AND the synth must proceed anyway.
    pub fn fresh_client(&self) -> Result<PooledClient> {
        self.in_use
            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        match msedge_tts::tts::client::connect() {
            Ok(client) => Ok(PooledClient { inner: client }),
            Err(e) => {
                self.in_use
                    .fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
                Err(VoiceError::Provider {
                    provider: "edge",
                    reason: format!("fresh connect: {e}"),
                    code: ReasonCode::WsClosed,
                })
            }
        }
    }

    /// Return a client to the pool with refreshed `last_used`.
    /// Silently drops the client if the pool is already at
    /// `max_size` idle (rare — only happens when in_use accounting
    /// races; the drop is fine, GC closes the WS).
    pub fn release(&self, client: MSEdgeTTSClient<std::net::TcpStream>) -> Result<()> {
        self.in_use
            .fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
        let mut g = self.inner.lock().map_err(poisoned_mutex)?;
        if g.len() >= self.config.max_size {
            // Drop on the floor.
            return Ok(());
        }
        g.push_back(PoolEntry {
            client,
            last_used: Instant::now(),
        });
        Ok(())
    }

    /// Number of idle warm clients (for stats / metrics).
    pub fn idle_count(&self) -> usize {
        self.inner.lock().map(|g| g.len()).unwrap_or(0)
    }

    /// Total clients (idle + checked-out).
    pub fn in_use_count(&self) -> usize {
        self.in_use.load(std::sync::atomic::Ordering::SeqCst)
    }
}

fn poisoned_mutex<T>(_e: std::sync::PoisonError<T>) -> VoiceError {
    VoiceError::Provider {
        provider: "edge",
        reason: "pool mutex poisoned".into(),
        code: ReasonCode::NetworkError,
    }
}

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

    #[test]
    fn config_default_values() {
        let c = EdgeWsPoolConfig::default();
        assert_eq!(c.max_size, 4);
        assert_eq!(c.idle_ttl, Duration::from_secs(300));
    }

    #[test]
    fn empty_pool_reports_zero_counts() {
        let p = EdgeWsPool::new(EdgeWsPoolConfig::default());
        assert_eq!(p.idle_count(), 0);
        assert_eq!(p.in_use_count(), 0);
    }

    #[test]
    fn acquire_errors_when_in_use_exceeds_cap_with_empty_pool() {
        let cfg = EdgeWsPoolConfig {
            max_size: 0,
            idle_ttl: Duration::from_secs(60),
        };
        let p = EdgeWsPool::new(cfg);
        let err = p.acquire().unwrap_err();
        let msg = format!("{err}");
        assert!(msg.contains("pool at cap"), "got `{msg}`");
    }

    #[test]
    fn debug_format_includes_counts_and_cap() {
        let p = EdgeWsPool::new(EdgeWsPoolConfig {
            max_size: 7,
            idle_ttl: Duration::from_secs(123),
        });
        let dbg = format!("{p:?}");
        assert!(dbg.contains("max_size: 7"));
        assert!(dbg.contains("idle_ttl_secs: 123"));
        assert!(dbg.contains("idle: 0"));
        assert!(dbg.contains("in_use: 0"));
    }
}
