//! SSE sender
//!
//! Provides a clonable `SseSender` that user code uses to push events to
//! a connected SSE client. Internally the channel carries pre-formatted
//! SSE frames (`String`) so that keep-alive comments and events can
//! travel through the same pipe.

use crate::upgrade::sse::Event;
use tokio::sync::mpsc;

/// SSE sender for transmitting events to clients
///
/// Cloneable, multi-producer. All clones write to the same underlying
/// channel; dropping the last clone ends the response stream cleanly.
#[derive(Clone)]
pub struct SseSender {
    tx: mpsc::UnboundedSender<String>,
}

impl SseSender {
    /// Create a new sender (internal — use `SseStream::sender()`).
    pub(crate) fn new(tx: mpsc::UnboundedSender<String>) -> Self {
        Self { tx }
    }

    /// Send an event to the client.
    ///
    /// Returns `Err` if the client has disconnected.
    pub async fn send(&self, event: Event) -> Result<(), mpsc::error::SendError<String>> {
        self.tx.send(event.format())
    }

    /// Send a data-only event (no event type or id).
    pub async fn send_data(
        &self,
        data: impl Into<String>,
    ) -> Result<(), mpsc::error::SendError<String>> {
        self.send(Event::data(data)).await
    }

    /// Send an event with a specific event type.
    pub async fn send_event(
        &self,
        event_type: impl Into<String>,
        data: impl Into<String>,
    ) -> Result<(), mpsc::error::SendError<String>> {
        self.send(Event::new(data).with_event(event_type)).await
    }

    /// Send a comment frame (`: text\n\n`). Comments are ignored by SSE
    /// clients but keep intermediaries from closing idle connections.
    pub async fn send_comment(
        &self,
        text: impl Into<String>,
    ) -> Result<(), mpsc::error::SendError<String>> {
        self.tx.send(Event::comment(text))
    }

    /// Whether the receiving end is still connected.
    pub fn is_connected(&self) -> bool {
        !self.tx.is_closed()
    }
}

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

    #[tokio::test]
    async fn test_send_formats_event() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let sender = SseSender::new(tx);

        sender.send(Event::new("hi")).await.unwrap();
        assert_eq!(rx.recv().await.unwrap(), "data: hi\n\n");
    }

    #[tokio::test]
    async fn test_send_data() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let sender = SseSender::new(tx);

        sender.send_data("hello").await.unwrap();
        assert_eq!(rx.recv().await.unwrap(), "data: hello\n\n");
    }

    #[tokio::test]
    async fn test_send_event() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let sender = SseSender::new(tx);

        sender.send_event("greet", "hi").await.unwrap();
        assert_eq!(rx.recv().await.unwrap(), "event: greet\ndata: hi\n\n");
    }

    #[tokio::test]
    async fn test_send_comment() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let sender = SseSender::new(tx);

        sender.send_comment("ping").await.unwrap();
        assert_eq!(rx.recv().await.unwrap(), ": ping\n\n");
    }

    #[tokio::test]
    async fn test_disconnect_detected() {
        let (tx, rx) = mpsc::unbounded_channel::<String>();
        let sender = SseSender::new(tx);
        drop(rx);
        assert!(!sender.is_connected());
        assert!(sender.send_data("x").await.is_err());
    }

    #[test]
    fn test_clone_shares_channel() {
        let (tx, _rx) = mpsc::unbounded_channel::<String>();
        let s1 = SseSender::new(tx);
        let s2 = s1.clone();
        assert!(s1.is_connected());
        assert!(s2.is_connected());
    }

    #[test]
    fn test_send_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<SseSender>();
    }
}
