use crate::client::signals::SignalsSnapshot;
use crate::types::events::StreamingEvent;
use crate::Result;
use futures::stream;
use futures::stream::Stream;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::sync::{oneshot, OwnedSemaphorePermit};

/// Snapshot of cumulative client metrics for monitoring and routing.
#[derive(Debug, Clone, Default)]
pub struct ClientMetrics {
    pub total_requests: u64,
    pub successful_requests: u64,
    pub total_tokens: u64,
}

/// Per-call statistics for observability and model selection.
#[derive(Debug, Clone, Default)]
pub struct CallStats {
    pub model: String,
    pub operation: String,
    pub endpoint: String,
    pub http_status: u16,
    pub retry_count: u32,
    pub duration_ms: u128,
    /// For streaming calls: time from request start to first emitted event (best-effort).
    pub first_event_ms: Option<u128>,
    /// For streaming calls: whether any event was emitted to the caller.
    pub emitted_any: bool,
    /// Always present; generated by the runtime for linkage.
    pub client_request_id: String,
    /// Best-effort request identifier from provider/edge headers (if available).
    pub upstream_request_id: Option<String>,
    pub error_class: Option<String>,
    pub usage: Option<serde_json::Value>,
    /// Snapshot of runtime signals captured for this call.
    pub signals: SignalsSnapshot,
}

/// Handle to cancel an in-flight streaming request.
///
/// Call [`cancel`](Self::cancel) to signal the stream to stop. The stream will emit
/// a final `StreamEnd { finish_reason: Some("cancelled") }` and then terminate.
///
/// # Example
///
/// ```ignore
/// let (stream, cancel_handle) = client.chat()
///     .messages(msgs)
///     .stream()
///     .execute_stream_with_cancel()
///     .await?;
///
/// // Cancel from another task or on user action
/// tokio::spawn(async move {
///     tokio::time::sleep(Duration::from_secs(5)).await;
///     cancel_handle.cancel();
/// });
/// ```
pub struct CancelHandle {
    sender: Option<oneshot::Sender<()>>,
}

impl CancelHandle {
    pub fn cancel(mut self) {
        if let Some(sender) = self.sender.take() {
            let _ = sender.send(());
        }
    }
}

pub(crate) fn cancel_pair() -> (CancelHandle, oneshot::Receiver<()>) {
    let (tx, rx) = oneshot::channel();
    (CancelHandle { sender: Some(tx) }, rx)
}

pub(crate) struct ControlledStream {
    pub inner: Pin<Box<dyn Stream<Item = Result<StreamingEvent>> + Send + 'static>>,
    pub cancel_rx: Option<oneshot::Receiver<()>>,
    pub permit: Option<OwnedSemaphorePermit>,
    cancelled: bool,
}

impl ControlledStream {
    pub fn new(
        inner: Pin<Box<dyn Stream<Item = Result<StreamingEvent>> + Send + 'static>>,
        cancel_rx: Option<oneshot::Receiver<()>>,
        permit: Option<OwnedSemaphorePermit>,
    ) -> Self {
        Self {
            inner,
            cancel_rx,
            permit,
            cancelled: false,
        }
    }
}

impl Stream for ControlledStream {
    type Item = Result<StreamingEvent>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // 1) Check cancellation (emit StreamEnd once, then terminate)
        if !self.cancelled {
            if let Some(ref mut cancel_rx) = self.cancel_rx {
                match std::future::Future::poll(Pin::new(cancel_rx), cx) {
                    Poll::Ready(Ok(())) => {
                        self.cancel_rx = None;
                        self.cancelled = true;
                        // Strong cancellation:
                        // - Drop the inner stream to stop network reads ASAP.
                        // - Release backpressure permit immediately.
                        self.inner = Box::pin(stream::empty());
                        self.permit = None;
                        return Poll::Ready(Some(Ok(StreamingEvent::StreamEnd {
                            finish_reason: Some("cancelled".to_string()),
                        })));
                    }
                    Poll::Ready(Err(_)) => {
                        // Sender was dropped without sending; ignore it but log trace for debugging.
                        tracing::trace!("ControlledStream: cancel_handle sender dropped without explicit signal; ignoring.");
                        self.cancel_rx = None;
                    }
                    Poll::Pending => {}
                }
            }
        }

        // 2) Poll inner stream
        match self.inner.as_mut().poll_next(cx) {
            Poll::Ready(Some(Ok(StreamingEvent::StreamEnd { finish_reason }))) => {
                // Stream finished normally; release permit early.
                self.permit = None;
                Poll::Ready(Some(Ok(StreamingEvent::StreamEnd { finish_reason })))
            }
            Poll::Ready(None) => {
                // Underlying stream ended; release permit early.
                self.permit = None;
                Poll::Ready(None)
            }
            other => other,
        }
    }
}
