//! SSE stream
//!
//! Bridges an `SseSender` (user-facing) to an HTTP streaming response
//! (`text/event-stream`). Optionally configures a periodic keep-alive
//! comment frame to keep the connection alive through buffering proxies.

use crate::core::StatusCode;
use crate::domain::Response;
use crate::upgrade::sse::{KeepAlive, SseSender};
use bytes::Bytes;
use futures_util::stream::Stream;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::sync::mpsc;

/// SSE stream for sending Server-Sent Events to clients.
///
/// Build the stream, grab a [`SseSender`] from it, then convert it into
/// a streaming `Response` with [`SseStream::into_response`].
pub struct SseStream {
    tx: mpsc::UnboundedSender<String>,
    rx: Option<mpsc::UnboundedReceiver<String>>,
    keep_alive: Option<KeepAlive>,
}

impl SseStream {
    /// Create a new SSE stream.
    pub fn new() -> Self {
        let (tx, rx) = mpsc::unbounded_channel();
        Self {
            tx,
            rx: Some(rx),
            keep_alive: None,
        }
    }

    /// Get a sender for this stream. Multiple senders may be created;
    /// they all write to the same client.
    pub fn sender(&self) -> SseSender {
        SseSender::new(self.tx.clone())
    }

    /// Enable periodic keep-alive comment frames.
    ///
    /// Sends `: keep-alive\n\n` every `interval` to prevent idle-timeout
    /// disconnects from proxies. The keep-alive task ends automatically
    /// when the client disconnects.
    pub fn with_keep_alive(mut self, keep_alive: KeepAlive) -> Self {
        self.keep_alive = Some(keep_alive);
        self
    }

    /// Convert the stream into an HTTP response with the standard SSE
    /// headers set (`Content-Type: text/event-stream`, `Cache-Control:
    /// no-cache`, `Connection: keep-alive`, `X-Accel-Buffering: no`).
    pub fn into_response(mut self) -> Response {
        let rx = self.rx.take().expect("receiver already taken");

        // Spawn keep-alive *after* taking the receiver so the channel is
        // still open. The keep-alive task lives until rx is dropped.
        if let Some(ka) = self.keep_alive {
            ka.spawn(SseSender::new(self.tx.clone()));
        }

        let event_stream = EventStream { rx };

        Response::new()
            .with_status_code(StatusCode::OK)
            .with_header("content-type", "text/event-stream")
            .with_header("cache-control", "no-cache")
            .with_header("connection", "keep-alive")
            // Tell nginx (and similar) not to buffer this response.
            .with_header("x-accel-buffering", "no")
            .with_stream(event_stream)
    }
}

impl Default for SseStream {
    fn default() -> Self {
        Self::new()
    }
}

/// Adapter from the SSE channel to the byte stream the response expects.
struct EventStream {
    rx: mpsc::UnboundedReceiver<String>,
}

impl Stream for EventStream {
    type Item = Result<Bytes, std::io::Error>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match self.rx.poll_recv(cx) {
            Poll::Ready(Some(frame)) => Poll::Ready(Some(Ok(Bytes::from(frame)))),
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::upgrade::sse::Event;
    use std::time::Duration;

    #[test]
    fn test_new_and_sender() {
        let stream = SseStream::new();
        assert!(stream.sender().is_connected());
    }

    #[tokio::test]
    async fn test_into_response_sets_headers() {
        let response = SseStream::new().into_response();
        assert_eq!(response.status(), StatusCode::OK);
        let h = response.headers();
        assert_eq!(h.get("content-type").unwrap(), "text/event-stream");
        assert_eq!(h.get("cache-control").unwrap(), "no-cache");
        assert_eq!(h.get("connection").unwrap(), "keep-alive");
        assert_eq!(h.get("x-accel-buffering").unwrap(), "no");
    }

    #[tokio::test]
    async fn test_event_stream_yields_frames() {
        use futures_util::StreamExt;

        let (tx, rx) = mpsc::unbounded_channel();
        let mut stream = EventStream { rx };

        tx.send(Event::new("hi").format()).unwrap();
        let item = stream.next().await.unwrap().unwrap();
        assert_eq!(&item[..], b"data: hi\n\n");
    }

    #[tokio::test]
    async fn test_event_stream_ends_on_drop() {
        use futures_util::StreamExt;

        let (tx, rx) = mpsc::unbounded_channel::<String>();
        let mut stream = EventStream { rx };
        drop(tx);
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn test_with_keep_alive_emits_comments() {
        // Build the stream, attach a fast keep-alive, take ownership of
        // the internal receiver via into_response_for_test, and verify
        // a comment frame arrives.
        let mut stream =
            SseStream::new().with_keep_alive(KeepAlive::new(Duration::from_millis(20)));
        let rx = stream.rx.take().unwrap();

        // Manually spawn keep-alive (mirrors what into_response does).
        if let Some(ka) = stream.keep_alive {
            ka.spawn(SseSender::new(stream.tx.clone()));
        }

        let mut event_stream = EventStream { rx };
        use futures_util::StreamExt;

        let frame = tokio::time::timeout(Duration::from_millis(200), event_stream.next())
            .await
            .expect("keep-alive frame should arrive")
            .expect("stream should yield a frame")
            .expect("frame should be Ok");
        assert_eq!(&frame[..], b": keep-alive\n\n");
    }

    #[test]
    fn test_send() {
        fn assert_send<T: Send>() {}
        assert_send::<SseStream>();
    }
}
