//! WebSocket connection
//!
//! Provides a high-level abstraction over WebSocket connections for sending
//! and receiving messages.
//!
//! # Examples
//!
//! ```rust,ignore
//! use ruxno::upgrade::websocket::{WebSocket, Message};
//!
//! async fn handle_websocket(mut socket: WebSocket) {
//!     // Receive messages
//!     while let Some(Ok(msg)) = socket.recv().await {
//!         match msg {
//!             Message::Text(text) => {
//!                 println!("Received: {}", text);
//!                 socket.send(Message::Text(text)).await.ok();
//!             }
//!             Message::Binary(data) => {
//!                 println!("Received {} bytes", data.len());
//!                 socket.send(Message::Binary(data)).await.ok();
//!             }
//!             Message::Ping(data) => {
//!                 socket.send(Message::Pong(data)).await.ok();
//!             }
//!             Message::Close => break,
//!             _ => {}
//!         }
//!     }
//! }
//! ```

use crate::core::CoreError;
use crate::upgrade::websocket::Message;
use futures_util::{SinkExt, Stream, StreamExt};
use hyper_tungstenite::WebSocketStream;
use hyper_util::rt::TokioIo;
use std::sync::Arc;
use tokio::sync::{broadcast, Mutex};
use uuid::Uuid;

/// Unique connection identifier
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct ConnectionId(Uuid);

impl ConnectionId {
    /// Create a new connection ID
    pub fn new() -> Self {
        Self(Uuid::new_v4())
    }

    /// Get the UUID
    pub fn as_uuid(&self) -> &Uuid {
        &self.0
    }
}

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

impl std::fmt::Display for ConnectionId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// WebSocket connection
///
/// Wraps a WebSocket stream and provides high-level methods for sending
/// and receiving messages. Handles ping/pong frames automatically.
///
/// # Design
///
/// The WebSocket is wrapped in `Arc<Mutex<>>` to allow sharing across tasks
/// while maintaining exclusive access for send/recv operations.
///
/// # Examples
///
/// ```rust,ignore
/// use ruxno::upgrade::websocket::{WebSocket, Message};
///
/// async fn echo_server(mut socket: WebSocket) {
///     while let Some(Ok(msg)) = socket.recv().await {
///         if let Message::Text(text) = msg {
///             socket.send(Message::Text(text)).await.ok();
///         }
///     }
/// }
/// ```
#[derive(Clone)]
pub struct WebSocket {
    /// WebSocket stream
    stream: Arc<Mutex<WebSocketStream<TokioIo<hyper::upgrade::Upgraded>>>>,
    /// Connection ID
    id: ConnectionId,
    /// Broadcast sender for room-based messaging
    broadcast_tx: Option<broadcast::Sender<Message>>,
}

impl WebSocket {
    /// Create a new WebSocket connection
    ///
    /// This is an internal constructor used by the upgrade process.
    pub(crate) fn new(
        stream: WebSocketStream<TokioIo<hyper::upgrade::Upgraded>>,
        id: ConnectionId,
        broadcast_tx: Option<broadcast::Sender<Message>>,
    ) -> Self {
        Self {
            stream: Arc::new(Mutex::new(stream)),
            id,
            broadcast_tx,
        }
    }

    /// Get the connection ID
    pub fn id(&self) -> ConnectionId {
        self.id
    }

    /// Send a message
    ///
    /// Sends a message over the WebSocket connection. Supports text, binary,
    /// ping, pong, and close messages.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// use ruxno::upgrade::websocket::Message;
    ///
    /// // Send text message
    /// socket.send(Message::text("Hello")).await?;
    ///
    /// // Send binary message
    /// socket.send(Message::binary(bytes)).await?;
    ///
    /// // Send ping
    /// socket.send(Message::ping(vec![])).await?;
    /// ```
    ///
    /// # Errors
    ///
    /// Returns an error if the connection is closed or if sending fails.
    pub async fn send(&mut self, msg: Message) -> Result<(), CoreError> {
        use std::pin::Pin;

        let ws_msg = Self::to_tungstenite(msg);
        let mut stream = self.stream.lock().await;
        let mut pinned = Pin::new(&mut *stream);

        pinned
            .send(ws_msg)
            .await
            .map_err(|e| CoreError::internal(format!("Failed to send WebSocket message: {}", e)))
    }

    /// Convert our Message to tungstenite Message
    fn to_tungstenite(msg: Message) -> hyper_tungstenite::tungstenite::Message {
        use hyper_tungstenite::tungstenite::Message as WsMessage;

        match msg {
            Message::Text(text) => WsMessage::Text(text.into_string().into()),
            Message::Binary(data) => WsMessage::Binary(data.into_bytes()),
            Message::Ping(data) => WsMessage::Ping(bytes::Bytes::from(data.into_vec())),
            Message::Pong(data) => WsMessage::Pong(bytes::Bytes::from(data.into_vec())),
            Message::Close => WsMessage::Close(None),
        }
    }

    /// Convert tungstenite Message to our Message
    fn from_tungstenite(msg: hyper_tungstenite::tungstenite::Message) -> Option<Message> {
        use hyper_tungstenite::tungstenite::Message as WsMessage;

        match msg {
            WsMessage::Text(text) => Some(Message::text(text.to_string())),
            WsMessage::Binary(data) => Some(Message::binary(data)),
            WsMessage::Ping(data) => Some(Message::ping(data.to_vec())),
            WsMessage::Pong(data) => Some(Message::pong(data.to_vec())),
            WsMessage::Close(_) => Some(Message::close()),
            WsMessage::Frame(_) => None, // Raw frames not exposed
        }
    }

    /// Send a text message
    ///
    /// Convenience method for sending text messages.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// socket.send_text("Hello, World!").await?;
    /// ```
    pub async fn send_text(&mut self, text: impl Into<String>) -> Result<(), CoreError> {
        self.send(Message::text(text)).await
    }

    /// Send a binary message
    ///
    /// Convenience method for sending binary messages.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// socket.send_binary(vec![1, 2, 3, 4]).await?;
    /// ```
    pub async fn send_binary(&mut self, data: impl Into<bytes::Bytes>) -> Result<(), CoreError> {
        self.send(Message::binary(data)).await
    }

    /// Send a ping message
    ///
    /// Sends a ping frame. The peer should respond with a pong.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// socket.send_ping(vec![]).await?;
    /// ```
    pub async fn send_ping(&mut self, data: Vec<u8>) -> Result<(), CoreError> {
        self.send(Message::ping(data)).await
    }

    /// Send a pong message
    ///
    /// Sends a pong frame, typically in response to a ping.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// socket.send_pong(vec![]).await?;
    /// ```
    pub async fn send_pong(&mut self, data: Vec<u8>) -> Result<(), CoreError> {
        self.send(Message::pong(data)).await
    }

    /// Receive a message
    ///
    /// Receives the next message from the WebSocket connection.
    /// Returns `None` when the connection is closed.
    ///
    /// Raw protocol frames (e.g. `WsMessage::Frame`) are skipped transparently;
    /// the method keeps waiting until a user-visible message or stream end arrives.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// while let Some(Ok(msg)) = socket.recv().await {
    ///     match msg {
    ///         Message::Text(text) => println!("Received: {}", text),
    ///         Message::Binary(data) => println!("Received {} bytes", data.len()),
    ///         Message::Close => break,
    ///         _ => {}
    ///     }
    /// }
    /// ```
    ///
    /// # Returns
    ///
    /// - `Some(Ok(Message))` - A message was received
    /// - `Some(Err(CoreError))` - An error occurred
    /// - `None` - The connection is closed
    pub async fn recv(&mut self) -> Option<Result<Message, CoreError>> {
        use std::pin::Pin;

        let mut stream = self.stream.lock().await;
        let mut pinned = Pin::new(&mut *stream);

        loop {
            match pinned.next().await {
                Some(Ok(msg)) => {
                    match Self::from_tungstenite(msg) {
                        Some(message) => return Some(Ok(message)),
                        // Raw frame or other non-user-visible message — skip and
                        // wait for the next item rather than signalling close.
                        None => continue,
                    }
                }
                Some(Err(e)) => {
                    return Some(Err(CoreError::internal(format!(
                        "Failed to receive WebSocket message: {}",
                        e
                    ))))
                }
                None => return None,
            }
        }
    }

    /// Try to receive a message without blocking
    ///
    /// Non-blocking version of `recv()`. Checks if a message is immediately
    /// available without waiting. Useful for polling the connection while
    /// performing other tasks.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// // Check for messages without blocking
    /// match socket.try_recv() {
    ///     Some(Ok(Message::Close(_))) => break,
    ///     Some(Err(_)) => break,
    ///     Some(Ok(msg)) => handle_message(msg),
    ///     None => {
    ///         // No message available, continue with other work
    ///     }
    /// }
    /// ```
    ///
    /// # Returns
    ///
    /// - `Some(Ok(Message))` - A message was immediately available
    /// - `Some(Err(CoreError))` - An error occurred
    /// - `None` - No message available or connection closed
    pub fn try_recv(&mut self) -> Option<Result<Message, CoreError>> {
        use std::pin::Pin;

        // Try to acquire the lock without blocking
        let Ok(mut stream) = self.stream.try_lock() else {
            return None;
        };

        let mut pinned = Pin::new(&mut *stream);

        // Use poll_next to check without blocking
        use futures_util::task::{noop_waker, Context as FutContext};
        let waker = noop_waker();
        let mut cx = FutContext::from_waker(&waker);

        loop {
            match pinned.as_mut().poll_next(&mut cx) {
                std::task::Poll::Ready(Some(Ok(msg))) => {
                    match Self::from_tungstenite(msg) {
                        Some(message) => return Some(Ok(message)),
                        // Raw frame — skip and try next
                        None => continue,
                    }
                }
                std::task::Poll::Ready(Some(Err(e))) => {
                    return Some(Err(CoreError::internal(format!(
                        "Failed to receive WebSocket message: {}",
                        e
                    ))))
                }
                std::task::Poll::Ready(None) => return None,
                std::task::Poll::Pending => return None,
            }
        }
    }

    /// Close the connection
    ///
    /// Sends a close frame and closes the WebSocket connection gracefully.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// socket.close().await?;
    /// ```
    ///
    /// # Errors
    ///
    /// Returns an error if closing fails.
    pub async fn close(mut self) -> Result<(), CoreError> {
        self.send(Message::Close).await
    }

    /// Broadcast a message to all connections in the same room
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// socket.broadcast(Message::text("Hello everyone!")).ok();
    /// ```
    pub fn broadcast(&self, msg: Message) -> Result<(), CoreError> {
        if let Some(ref tx) = self.broadcast_tx {
            tx.send(msg)
                .map_err(|_| CoreError::internal("Failed to broadcast message"))?;
            Ok(())
        } else {
            Err(CoreError::internal(
                "Broadcasting not enabled for this connection",
            ))
        }
    }

    /// Subscribe to broadcast messages
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// if let Some(mut rx) = socket.subscribe() {
    ///     while let Ok(msg) = rx.recv().await {
    ///         socket.send(msg).await.ok();
    ///     }
    /// }
    /// ```
    pub fn subscribe(&self) -> Option<broadcast::Receiver<Message>> {
        self.broadcast_tx.as_ref().map(|tx| tx.subscribe())
    }
}

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

    #[test]
    fn test_connection_id_uniqueness() {
        let id1 = ConnectionId::new();
        let id2 = ConnectionId::new();
        assert_ne!(id1, id2);
    }

    #[test]
    fn test_connection_id_display() {
        let id = ConnectionId::new();
        let display = format!("{}", id);
        assert!(!display.is_empty());
    }

    #[test]
    fn test_connection_id_default() {
        let id = ConnectionId::default();
        assert!(!id.as_uuid().to_string().is_empty());
    }
}
