use std::time::Duration;

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::sync::Mutex;
use tokio::time::timeout;

#[cfg(unix)]
use std::env;
#[cfg(unix)]
use std::path::{Path, PathBuf};
#[cfg(unix)]
use tokio::net::UnixStream;

use super::error::PineError;
use super::protocol::{DEFAULT_PCSX2_SLOT, MAX_RESPONSE_SIZE};

/// A PCSX2 Pine slot used to derive the default endpoint.
///
/// # Examples
///
/// ```rust
/// use pine_client::PineSlot;
///
/// let slot = PineSlot::new(28_011)?;
/// assert_eq!(slot.get(), 28_011);
/// # Ok::<(), pine_client::PineError>(())
/// ```
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub struct PineSlot(u16);

impl PineSlot {
    /// Creates a slot from a raw value.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use pine_client::PineSlot;
    ///
    /// let slot = PineSlot::new(28_012)?;
    /// assert_eq!(slot.get(), 28_012);
    /// # Ok::<(), pine_client::PineError>(())
    /// ```
    pub fn new(value: u32) -> Result<Self, PineError> {
        let slot = u16::try_from(value).map_err(|_| PineError::InvalidSlot { slot: value })?;
        Ok(Self(slot))
    }

    /// Returns the raw slot value.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use pine_client::PineSlot;
    ///
    /// let slot = PineSlot::new(28_012)?;
    /// assert_eq!(slot.get(), 28_012);
    /// # Ok::<(), pine_client::PineError>(())
    /// ```
    #[must_use]
    pub const fn get(self) -> u16 {
        self.0
    }
}

impl Default for PineSlot {
    fn default() -> Self {
        Self(DEFAULT_PCSX2_SLOT)
    }
}

/// A connection target for the Pine server.
///
/// # Examples
///
/// ```rust
/// use pine_client::{PineEndpoint, PineSlot};
///
/// let endpoint = PineEndpoint::pcsx2(PineSlot::default());
/// let _description = endpoint.describe();
/// ```
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum PineEndpoint {
    /// A TCP endpoint, used by the Windows Pine server.
    Tcp { host: String, port: u16 },
    /// A Unix domain socket endpoint, used by macOS and Linux Pine servers.
    #[cfg(unix)]
    Unix { path: PathBuf },
}

impl PineEndpoint {
    /// Returns the default PCSX2 endpoint for the given slot.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use pine_client::{PineEndpoint, PineSlot};
    ///
    /// let endpoint = PineEndpoint::pcsx2(PineSlot::default());
    /// let _ = endpoint.describe();
    /// ```
    #[must_use]
    pub fn pcsx2(slot: PineSlot) -> Self {
        #[cfg(unix)]
        {
            Self::Unix {
                path: pcsx2_socket_path(slot),
            }
        }
        #[cfg(not(unix))]
        {
            Self::Tcp {
                host: "127.0.0.1".to_string(),
                port: slot.get(),
            }
        }
    }

    /// Formats the endpoint as a stable URI-like string.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use pine_client::PineEndpoint;
    ///
    /// let endpoint = PineEndpoint::Tcp {
    ///     host: "127.0.0.1".to_string(),
    ///     port: 28_011,
    /// };
    /// assert_eq!(endpoint.describe(), "tcp://127.0.0.1:28011");
    /// ```
    #[must_use]
    pub fn describe(&self) -> String {
        match self {
            Self::Tcp { host, port } => format!("tcp://{host}:{port}"),
            #[cfg(unix)]
            Self::Unix { path } => format!("unix://{}", path.display()),
        }
    }
}

/// Timeout settings used when opening and using Pine connections.
///
/// # Examples
///
/// ```rust
/// use std::time::Duration;
///
/// use pine_client::PineOptions;
///
/// let options = PineOptions {
///     connect_timeout: Duration::from_secs(1),
///     io_timeout: Duration::from_secs(2),
/// };
/// assert_eq!(options.connect_timeout, Duration::from_secs(1));
/// ```
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct PineOptions {
    /// Maximum duration allowed for establishing a socket.
    pub connect_timeout: Duration,
    /// Maximum duration allowed for each read or write operation.
    pub io_timeout: Duration,
}

impl Default for PineOptions {
    fn default() -> Self {
        Self {
            connect_timeout: Duration::from_secs(2),
            io_timeout: Duration::from_secs(2),
        }
    }
}

pub(crate) struct PineTransport {
    endpoint: PineEndpoint,
    options: PineOptions,
    connection: Option<PineConnection>,
}

impl PineTransport {
    pub(crate) fn new(endpoint: PineEndpoint, options: PineOptions) -> Self {
        Self {
            endpoint,
            options,
            connection: None,
        }
    }

    pub(crate) fn endpoint(&self) -> PineEndpoint {
        self.endpoint.clone()
    }

    pub(crate) fn is_connected(&self) -> bool {
        self.connection.is_some()
    }

    pub(crate) async fn connect(&mut self) -> Result<(), PineError> {
        if self.connection.is_none() {
            let connection = PineConnection::connect(&self.endpoint, self.options.connect_timeout).await?;
            self.connection = Some(connection);
        }
        Ok(())
    }

    pub(crate) fn disconnect(&mut self) {
        self.connection = None;
    }

    pub(crate) async fn round_trip(&mut self, packet: &[u8]) -> Result<Vec<u8>, PineError> {
        self.connect().await?;

        let Some(connection) = self.connection.as_mut() else {
            unreachable!("connection is established by PineTransport::connect");
        };

        let response = connection.round_trip(packet, self.options.io_timeout).await;

        if let Err(error) = &response {
            if error.invalidates_connection() {
                self.connection = None;
            }
        }

        response
    }
}

pub(crate) type SharedTransport = Mutex<PineTransport>;

struct PineConnection {
    stream: PineStream,
}

impl PineConnection {
    async fn connect(endpoint: &PineEndpoint, connect_timeout: Duration) -> Result<Self, PineError> {
        let stream = PineStream::connect(endpoint, connect_timeout).await?;
        Ok(Self { stream })
    }

    async fn round_trip(&mut self, packet: &[u8], io_timeout: Duration) -> Result<Vec<u8>, PineError> {
        self.stream.write_all(packet, io_timeout, "write request").await?;

        let mut header = [0_u8; 4];
        self.stream
            .read_exact(&mut header, io_timeout, "read response header")
            .await?;

        let frame_size = u32::from_le_bytes(header);
        let frame_size = usize::try_from(frame_size).map_err(|_| PineError::InvalidPacket {
            message: "response frame size does not fit usize".to_string(),
        })?;

        if !(5..=MAX_RESPONSE_SIZE).contains(&frame_size) {
            return Err(PineError::InvalidPacket {
                message: format!("response frame size out of range: {frame_size}"),
            });
        }

        // 预分配精确的连续内存并一次性读取，省去 extend 和临时 body 数组的拷贝开销
        let mut response = vec![0_u8; frame_size];
        response[0..4].copy_from_slice(&header);

        self.stream
            .read_exact(&mut response[4..], io_timeout, "read response body")
            .await?;

        Ok(response)
    }
}

enum PineStream {
    Tcp(TcpStream),
    #[cfg(unix)]
    Unix(UnixStream),
}

impl PineStream {
    async fn connect(endpoint: &PineEndpoint, connect_timeout: Duration) -> Result<Self, PineError> {
        match endpoint {
            PineEndpoint::Tcp { host, port } => {
                let address = format!("{host}:{port}");
                let stream = timeout(connect_timeout, TcpStream::connect(&address))
                    .await
                    .map_err(|_| PineError::Timeout {
                        phase: "connect",
                        timeout: connect_timeout,
                    })?
                    .map_err(|source| PineError::Connect {
                        endpoint: endpoint.describe(),
                        source,
                    })?;
                stream.set_nodelay(true).map_err(|source| PineError::Connect {
                    endpoint: endpoint.describe(),
                    source,
                })?;
                Ok(Self::Tcp(stream))
            }
            #[cfg(unix)]
            PineEndpoint::Unix { path } => {
                let stream = timeout(connect_timeout, UnixStream::connect(path))
                    .await
                    .map_err(|_| PineError::Timeout {
                        phase: "connect",
                        timeout: connect_timeout,
                    })?
                    .map_err(|source| PineError::Connect {
                        endpoint: endpoint.describe(),
                        source,
                    })?;
                Ok(Self::Unix(stream))
            }
        }
    }

    async fn write_all(&mut self, buffer: &[u8], io_timeout: Duration, phase: &'static str) -> Result<(), PineError> {
        match self {
            Self::Tcp(stream) => timeout(io_timeout, stream.write_all(buffer))
                .await
                .map_err(|_| PineError::Timeout {
                    phase,
                    timeout: io_timeout,
                })?
                .map_err(map_disconnect(phase)),
            #[cfg(unix)]
            Self::Unix(stream) => timeout(io_timeout, stream.write_all(buffer))
                .await
                .map_err(|_| PineError::Timeout {
                    phase,
                    timeout: io_timeout,
                })?
                .map_err(map_disconnect(phase)),
        }
    }

    async fn read_exact(
        &mut self,
        buffer: &mut [u8],
        io_timeout: Duration,
        phase: &'static str,
    ) -> Result<(), PineError> {
        match self {
            Self::Tcp(stream) => {
                let io_result =
                    timeout(io_timeout, stream.read_exact(buffer))
                        .await
                        .map_err(|_| PineError::Timeout {
                            phase,
                            timeout: io_timeout,
                        })?;

                io_result.map(|_| ()).map_err(map_disconnect(phase))
            }
            #[cfg(unix)]
            Self::Unix(stream) => {
                let io_result =
                    timeout(io_timeout, stream.read_exact(buffer))
                        .await
                        .map_err(|_| PineError::Timeout {
                            phase,
                            timeout: io_timeout,
                        })?;

                io_result.map(|_| ()).map_err(map_disconnect(phase))
            }
        }
    }
}

fn map_disconnect(phase: &'static str) -> impl FnOnce(std::io::Error) -> PineError {
    move |source| match source.kind() {
        std::io::ErrorKind::UnexpectedEof
        | std::io::ErrorKind::BrokenPipe
        | std::io::ErrorKind::ConnectionReset
        | std::io::ErrorKind::ConnectionAborted
        | std::io::ErrorKind::NotConnected => PineError::Disconnected { phase },
        _ => PineError::Connect {
            endpoint: format!("active-connection:{phase}"),
            source,
        },
    }
}

pub(crate) fn validate_endpoint(endpoint: &PineEndpoint) -> Result<(), PineError> {
    match endpoint {
        PineEndpoint::Tcp { .. } => Ok(()),
        #[cfg(unix)]
        PineEndpoint::Unix { path } => validate_socket_path(path),
    }
}

#[cfg(unix)]
fn pcsx2_socket_path(slot: PineSlot) -> PathBuf {
    let mut path = runtime_directory();
    path.push("pcsx2.sock");
    if slot.get() != DEFAULT_PCSX2_SLOT {
        let file_name = format!("pcsx2.sock.{}", slot.get());
        path.set_file_name(file_name);
    }
    path
}

#[cfg(unix)]
fn runtime_directory() -> PathBuf {
    #[cfg(target_os = "macos")]
    let runtime = env::var_os("TMPDIR");
    #[cfg(not(target_os = "macos"))]
    let runtime = env::var_os("XDG_RUNTIME_DIR");

    runtime.map_or_else(|| PathBuf::from("/tmp"), PathBuf::from)
}

#[cfg(unix)]
fn validate_socket_path(path: &Path) -> Result<(), PineError> {
    let path_string = path.as_os_str().to_string_lossy();
    if path_string.is_empty() {
        return Err(PineError::InvalidSocketPath {
            path: path.to_path_buf(),
        });
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{PineEndpoint, PineSlot, validate_endpoint};

    #[test]
    fn pine_slot_validates_u16_range() {
        assert!(PineSlot::new(28_011).is_ok());
        assert!(PineSlot::new(u32::from(u16::MAX) + 1).is_err());
    }

    #[test]
    fn tcp_endpoint_description_is_stable() {
        let endpoint = PineEndpoint::Tcp {
            host: "127.0.0.1".to_string(),
            port: 28_011,
        };
        assert_eq!(endpoint.describe(), "tcp://127.0.0.1:28011");
        assert!(validate_endpoint(&endpoint).is_ok());
    }

    #[cfg(unix)]
    #[test]
    fn unix_slot_suffix_matches_pcsx2_convention() {
        let slot = PineSlot::new(28_012).expect("slot is valid");
        let endpoint = PineEndpoint::pcsx2(slot);
        let PineEndpoint::Unix { path } = endpoint else {
            unreachable!("unix targets should use unix sockets");
        };
        assert_eq!(
            path.file_name().and_then(|value| value.to_str()),
            Some("pcsx2.sock.28012")
        );
    }
}
