//! Monitored channel wrapper with backpressure monitoring and timeout-based sends

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::timeout;
use tonic::Status;
use tracing::{debug, warn};

use super::backpressure::{BackpressureMonitor, BackpressureStatus};
use super::metrics::StreamMetrics;

/// Default send timeout (100ms - balances responsiveness with HFT requirements)
const DEFAULT_SEND_TIMEOUT_MS: u64 = 100;

/// Monitored sender wrapper with backpressure monitoring
///
/// Wraps tokio::sync::mpsc::Sender with:
/// - Automatic backpressure monitoring
///
/// - Timeout-based sends
/// - Prometheus metrics integration
///
/// - Graceful degradation with logging
pub struct MonitoredSender<T> {
    inner: mpsc::Sender<T>,
    monitor: Arc<BackpressureMonitor>,
    metrics: Arc<StreamMetrics>,
    send_timeout: Duration,
}

impl<T> MonitoredSender<T> {
    /// Create a new monitored sender
    pub fn new(
        inner: mpsc::Sender<T>,
        monitor: Arc<BackpressureMonitor>,
        metrics: Arc<StreamMetrics>,
    ) -> Self {
        Self {
            inner,
            monitor,
            metrics,
            send_timeout: Duration::from_millis(DEFAULT_SEND_TIMEOUT_MS),
        }
    }

    /// Set custom send timeout
    pub fn with_timeout(mut self, timeout_ms: u64) -> Self {
        self.send_timeout = Duration::from_millis(timeout_ms);
        self
    }

    /// Send a message with backpressure monitoring and timeout
    ///
    /// Returns Ok(()) if the message was sent successfully.
    ///
    /// Returns Err(Status) if:
    /// - Send timeout expired (backpressure)
    ///
    /// - Channel closed
    /// - Buffer full
    pub async fn send_monitored(&self, value: T) -> Result<(), Status> {
        // Check backpressure status before attempting send
        let current_capacity = self.inner.capacity();
        let buffer_size = self.inner.max_capacity() - current_capacity;
        let status = self.monitor.check(buffer_size);

        // Update metrics
        self.metrics
            .set_buffer_utilization(status.utilization_pct() as f64);

        // Log warning/critical status
        if status.is_warning() {
            self.monitor.log_status(&status, self.metrics.stream_name());

            if matches!(status, BackpressureStatus::Warning { .. }) {
                self.metrics.inc_backpressure_warnings();
            } else if matches!(status, BackpressureStatus::Critical { .. }) {
                self.metrics.inc_backpressure_critical();
            }
        }

        // Handle full buffer immediately
        if status == BackpressureStatus::Full {
            self.monitor.record_drop();
            self.metrics.inc_messages_dropped("buffer_full");
            return Err(Status::resource_exhausted(format!(
                "Stream buffer full: {}",
                self.metrics.stream_name()
            )));
        }

        // Attempt send with timeout
        match timeout(self.send_timeout, self.inner.send(value)).await {
            Ok(Ok(())) => {
                // Success
                self.monitor.record_send();
                self.metrics.inc_messages_sent();
                Ok(())
            },
            Ok(Err(_send_error)) => {
                // Channel closed
                self.monitor.record_drop();
                self.metrics.inc_messages_dropped("channel_closed");
                Err(Status::internal("Stream channel closed"))
            },
            Err(_timeout_error) => {
                // Timeout expired
                warn!(
                    stream = self.metrics.stream_name(),
                    timeout_ms = self.send_timeout.as_millis(),
                    "Stream send timeout - dropping message"
                );
                self.monitor.record_drop();
                self.metrics.inc_send_timeouts();
                self.metrics.inc_messages_dropped("timeout");
                Err(Status::deadline_exceeded(format!(
                    "Stream send timeout after {}ms",
                    self.send_timeout.as_millis()
                )))
            },
        }
    }

    /// Send a message with best-effort delivery (logs but doesn't fail on drop)
    ///
    /// Use this for non-critical updates where occasional message loss is acceptable.
    pub async fn send_best_effort(&self, value: T) {
        if let Err(e) = self.send_monitored(value).await {
            debug!(
                stream = self.metrics.stream_name(),
                error = %e,
                "Best-effort send dropped message (expected under load)"
            );
        }
    }

    /// Get current buffer utilization percentage
    pub fn utilization_pct(&self) -> u8 {
        let current_capacity = self.inner.capacity();
        let buffer_size = self.inner.max_capacity() - current_capacity;
        let status = self.monitor.check(buffer_size);
        status.utilization_pct()
    }

    /// Check if the stream is healthy (< warning threshold)
    pub fn is_healthy(&self) -> bool {
        let current_capacity = self.inner.capacity();
        let buffer_size = self.inner.max_capacity() - current_capacity;
        let status = self.monitor.check(buffer_size);
        !status.is_warning()
    }
}

impl<T> Clone for MonitoredSender<T> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
            monitor: Arc::clone(&self.monitor),
            metrics: Arc::clone(&self.metrics),
            send_timeout: self.send_timeout,
        }
    }
}

/// Create a monitored channel with backpressure monitoring
///
/// Returns (MonitoredSender, Receiver, BackpressureMonitor, StreamMetrics)
///
/// # Example
/// ```ignore
/// let (tx, rx, monitor, metrics) = create_monitored_channel(
///     1000,
///     "order_stream",
/// );
///
/// // Send with monitoring
/// tx.send_monitored(order_event).await?;
/// ```
pub fn create_monitored_channel<T>(
    buffer_size: usize,
    stream_name: impl Into<String>,
) -> (
    MonitoredSender<T>,
    mpsc::Receiver<T>,
    Arc<BackpressureMonitor>,
    Arc<StreamMetrics>,
) {
    let stream_name = stream_name.into();

    // Create underlying channel
    let (tx, rx) = mpsc::channel(buffer_size);

    // Create monitor and metrics
    let monitor = Arc::new(BackpressureMonitor::with_capacity(
        buffer_size,
        stream_name.clone(),
    ));
    let metrics = Arc::new(StreamMetrics::new(stream_name));

    // Create monitored sender
    let monitored_tx = MonitoredSender::new(tx, Arc::clone(&monitor), Arc::clone(&metrics));

    (monitored_tx, rx, monitor, metrics)
}

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

    #[tokio::test]
    async fn test_monitored_send_success() {
        let (tx, mut rx, _monitor, _metrics) = create_monitored_channel(10, "test_stream");

        // Send should succeed
        let result = tx.send_monitored("test_message").await;
        assert!(result.is_ok());

        // Receive should work
        let msg = rx.recv().await;
        assert_eq!(msg, Some("test_message"));
    }

    #[tokio::test]
    async fn test_monitored_send_timeout() {
        let (tx, _rx, _monitor, _metrics) = create_monitored_channel(1, "test_stream");
        let tx = tx.with_timeout(10); // 10ms timeout

        // Fill the buffer
        tx.send_monitored("msg1")
            .await
            .expect("First send should succeed");

        // This should fail with ResourceExhausted since buffer is full
        // (The backpressure check happens before timeout logic)
        let result = tx.send_monitored("msg2").await;
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(matches!(err.code(), tonic::Code::ResourceExhausted));
    }

    #[tokio::test]
    async fn test_best_effort_send() {
        let (tx, mut rx, _monitor, _metrics) = create_monitored_channel(10, "test_stream");

        // Best-effort sends should not panic
        tx.send_best_effort("msg1").await;
        tx.send_best_effort("msg2").await;

        // Messages should still be received
        assert_eq!(rx.recv().await, Some("msg1"));
        assert_eq!(rx.recv().await, Some("msg2"));
    }

    #[tokio::test]
    async fn test_utilization_tracking() {
        let (tx, _rx, _monitor, _metrics) = create_monitored_channel(10, "test_stream");

        // Empty buffer should show low utilization
        assert!(tx.utilization_pct() < 10);

        // Fill buffer partially
        for i in 0..5 {
            tx.send_monitored(format!("msg{}", i))
                .await
                .expect("Send should succeed");
        }

        // Should show ~50% utilization
        let util = tx.utilization_pct();
        assert!((40..=60).contains(&util), "Expected ~50%, got {}%", util);
    }
}
