chateau 0.3.2

//! Unix Domain Socket transport implementation for client connections.
//!
//! This module contains the [`UnixTransport`] type, which is a [`tower::Service`] that connects to
//! Unix domain sockets. Unlike TCP transports, Unix sockets use filesystem paths as addresses.
//!
//! The transport extracts the socket path from the URI authority or path component and establishes
//! a connection to the Unix domain socket at that location.

use std::io;
use std::marker::PhantomData;
use std::path::{Path, PathBuf};
use std::task::{Context, Poll};
use std::time::Duration;

use thiserror::Error;
use tokio::io::{AsyncRead, AsyncWrite};
use tracing::trace;

use crate::info::{HasConnectionInfo, UnixAddr};
use crate::stream::unix::UnixStream;

/// A request to a unix transport
#[derive(Debug, Clone)]
pub struct UnixRequest<R> {
    request: R,
    address: UnixAddr,
}

impl<R> UnixRequest<R> {
    /// Create a new UnixRequest, binding the request object and the address.
    pub fn new(request: R, address: UnixAddr) -> Self {
        Self { request, address }
    }

    /// Access the request
    pub fn request(&self) -> &R {
        &self.request
    }

    /// Access the address
    pub fn address(&self) -> &UnixAddr {
        &self.address
    }

    /// Consumes the request, returning the request.
    pub fn into_request(self) -> R {
        self.request
    }

    /// Consumes the request, returning the request and address.
    pub fn into_parts(self) -> (R, UnixAddr) {
        (self.request, self.address)
    }
}

/// A Unix Domain Socket connector for client connections.
///
/// This type is a [`tower::Service`] that connects to Unix domain sockets using filesystem paths.
/// The socket path is extracted from the URI - it can be specified in the authority component
/// (for URIs like `unix:///path/to/socket`) or as the path component.
#[derive(Debug)]
pub struct UnixTransport<IO = UnixStream> {
    config: UnixTransportConfig,
    stream: PhantomData<fn() -> IO>,
}

impl<IO> Clone for UnixTransport<IO> {
    fn clone(&self) -> Self {
        Self {
            config: self.config.clone(),
            stream: PhantomData,
        }
    }
}

impl<IO> Default for UnixTransport<IO> {
    fn default() -> Self {
        Self::new(UnixTransportConfig::default())
    }
}

impl<IO> UnixTransport<IO> {
    /// Create a new Unix transport with the given configuration.
    pub fn new(config: UnixTransportConfig) -> Self {
        Self {
            config,
            stream: PhantomData,
        }
    }

    /// Get the configuration for the Unix transport.
    pub fn config(&self) -> &UnixTransportConfig {
        &self.config
    }

    /// Set the configuration for the Unix transport.
    pub fn with_config(mut self, config: UnixTransportConfig) -> Self {
        self.config = config;
        self
    }
}

type BoxFuture<'a, T, E> = crate::BoxFuture<'a, Result<T, E>>;

impl<IO, R> tower::Service<UnixRequest<R>> for UnixTransport<IO>
where
    UnixStream: Into<IO>,
    IO: HasConnectionInfo + AsyncRead + AsyncWrite + Send + Unpin + 'static,
    IO::Addr: Clone + Unpin + Send + 'static,
{
    type Response = IO;
    type Error = UnixConnectionError;
    type Future = BoxFuture<'static, Self::Response, Self::Error>;

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: UnixRequest<R>) -> Self::Future {
        let config = self.config.clone();
        let (_, address) = req.into_parts();

        Box::pin(async move {
            let path = address.path().ok_or(UnixConnectionError::UnnamedAddress)?;
            let stream = connect_unix_socket(path, config.connect_timeout).await?;

            trace!(path = %path.display(), "unix socket connected");

            let stream = stream.into();
            Ok(stream)
        })
    }
}

/// Connect to a Unix domain socket at the given path.
async fn connect_unix_socket<P: AsRef<Path>>(
    path: P,
    connect_timeout: Option<Duration>,
) -> Result<UnixStream, UnixConnectionError> {
    let connect_future = UnixStream::connect(path);

    match connect_timeout {
        Some(timeout) => match tokio::time::timeout(timeout, connect_future).await {
            Ok(Ok(stream)) => Ok(stream),
            Ok(Err(error)) => {
                trace!(kind=%error.kind(), "unix connection error: {error}");
                Err(UnixConnectionError::ConnectionError(error))
            }
            Err(_) => {
                trace!(timeout=?timeout, "unix connection timed out");
                Err(UnixConnectionError::Timeout(timeout))
            }
        },
        None => connect_future.await.map_err(|error| {
            trace!(kind=%error.kind(), "unix connection error: {error}");
            UnixConnectionError::ConnectionError(error)
        }),
    }
}

/// Error type for Unix socket connections.
#[derive(Debug, Error)]
pub enum UnixConnectionError {
    /// Error when no unix address is found in request extensions.
    #[error("No unix address in request extensions")]
    NoAddress,

    /// Error when the unix address is unnamed.
    #[error("Unnamed unix address provided")]
    UnnamedAddress,

    /// Error when the unix connection fails.
    #[error("Unix connection: {0}")]
    ConnectionError(#[from] io::Error),

    /// Error when the unix connection times out.
    #[error("Connection timed out after {}ms", .0.as_millis())]
    Timeout(Duration),
}

/// Configuration for Unix domain socket connections.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct UnixTransportConfig {
    /// The timeout for connecting to a Unix socket.
    pub connect_timeout: Option<Duration>,
}

impl Default for UnixTransportConfig {
    fn default() -> Self {
        Self {
            connect_timeout: Some(Duration::from_secs(10)),
        }
    }
}

/// A Unix Domain Socket connector that always connects to a single static address.
///
/// Unlike [`UnixTransport`], which extracts the socket path from request extensions,
/// this transport is configured with a single Unix socket path and always connects
/// to that address regardless of the request.
#[derive(Debug)]
pub struct StaticAddressUnixTransport<IO = UnixStream> {
    address: PathBuf,
    config: UnixTransportConfig,
    stream: PhantomData<fn() -> IO>,
}

impl<IO> Clone for StaticAddressUnixTransport<IO> {
    fn clone(&self) -> Self {
        Self {
            address: self.address.clone(),
            config: self.config.clone(),
            stream: PhantomData,
        }
    }
}

impl<IO> StaticAddressUnixTransport<IO> {
    /// Create a new static Unix transport that always connects to the given path.
    pub fn new<P: Into<PathBuf>>(path: P) -> Self {
        Self {
            address: path.into(),
            config: UnixTransportConfig::default(),
            stream: PhantomData,
        }
    }

    /// Create a new static Unix transport with the given configuration.
    pub fn with_config<P: Into<PathBuf>>(path: P, config: UnixTransportConfig) -> Self {
        Self {
            address: path.into(),
            config,
            stream: PhantomData,
        }
    }

    /// Get the static address this transport connects to.
    pub fn address(&self) -> &Path {
        &self.address
    }

    /// Get the configuration for the Unix transport.
    pub fn config(&self) -> &UnixTransportConfig {
        &self.config
    }

    /// Set the configuration for the Unix transport.
    pub fn set_config(&mut self, config: UnixTransportConfig) {
        self.config = config;
    }
}

impl<IO, R> tower::Service<R> for StaticAddressUnixTransport<IO>
where
    UnixStream: Into<IO>,
    IO: HasConnectionInfo + AsyncRead + AsyncWrite + Send + Unpin + 'static,
    IO::Addr: Clone + Unpin + Send + 'static,
{
    type Response = IO;
    type Error = UnixConnectionError;
    type Future = BoxFuture<'static, Self::Response, Self::Error>;

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, _req: R) -> Self::Future {
        let address = self.address.clone();
        let config = self.config.clone();

        Box::pin(async move {
            trace!(path = %address.display(), "unix socket connecting");

            let stream = connect_unix_socket(&address, config.connect_timeout)
                .await
                .inspect_err(|error| {
                    trace!(path = %address.display(), "unix socket connection error: {error}");
                })?;

            trace!(path = %address.display(), "unix socket connected to static address");

            let stream = stream.into();
            Ok(stream)
        })
    }
}

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

    #[test]
    fn test_unix_connection_error_display() {
        let error = UnixConnectionError::NoAddress;
        assert_eq!(error.to_string(), "No unix address in request extensions");

        let error = UnixConnectionError::UnnamedAddress;
        assert_eq!(error.to_string(), "Unnamed unix address provided");

        let timeout = std::time::Duration::from_secs(5);
        let error = UnixConnectionError::Timeout(timeout);
        assert_eq!(error.to_string(), "Connection timed out after 5000ms");
    }

    #[test]
    fn test_unix_transport_config() {
        let config = UnixTransportConfig::default();
        assert_eq!(
            config.connect_timeout,
            Some(std::time::Duration::from_secs(10))
        );

        let custom_config = UnixTransportConfig {
            connect_timeout: Some(std::time::Duration::from_secs(30)),
        };

        let transport = UnixTransport::<UnixStream>::new(custom_config.clone());
        assert_eq!(
            transport.config().connect_timeout,
            custom_config.connect_timeout
        );

        let transport_with_config =
            UnixTransport::<UnixStream>::default().with_config(custom_config.clone());
        assert_eq!(
            transport_with_config.config().connect_timeout,
            custom_config.connect_timeout
        );
    }

    #[test]
    fn test_static_address_unix_transport_new() {
        let transport = StaticAddressUnixTransport::<UnixStream>::new("/var/run/test.sock");
        assert_eq!(transport.address(), &PathBuf::from("/var/run/test.sock"));
        assert_eq!(
            transport.config().connect_timeout,
            Some(std::time::Duration::from_secs(10))
        );
    }

    #[test]
    fn test_static_address_unix_transport_with_config() {
        let config = UnixTransportConfig {
            connect_timeout: Some(std::time::Duration::from_secs(30)),
        };
        let transport = StaticAddressUnixTransport::<UnixStream>::with_config(
            "/var/run/test.sock",
            config.clone(),
        );
        assert_eq!(transport.address(), &PathBuf::from("/var/run/test.sock"));
        assert_eq!(transport.config().connect_timeout, config.connect_timeout);
    }

    #[test]
    fn test_static_address_unix_transport_set_config() {
        let mut transport = StaticAddressUnixTransport::<UnixStream>::new("/var/run/test.sock");
        let new_config = UnixTransportConfig {
            connect_timeout: Some(std::time::Duration::from_secs(60)),
        };
        transport.set_config(new_config.clone());
        assert_eq!(
            transport.config().connect_timeout,
            new_config.connect_timeout
        );
    }

    #[test]
    fn test_static_address_unix_transport_clone() {
        let transport = StaticAddressUnixTransport::<UnixStream>::new("/var/run/test.sock");
        let cloned = transport.clone();
        assert_eq!(transport.address(), cloned.address());
        assert_eq!(
            transport.config().connect_timeout,
            cloned.config().connect_timeout
        );
    }

    #[tokio::test]
    async fn test_static_address_unix_transport_connection_failure() {
        let transport = StaticAddressUnixTransport::<UnixStream>::new("/nonexistent/socket.sock");

        let result = tower::ServiceExt::oneshot(transport, ()).await;
        assert!(result.is_err());
        match result.unwrap_err() {
            UnixConnectionError::ConnectionError(_) => {} // Expected
            other => panic!("Unexpected error type: {other:?}"),
        }
    }

    #[test]
    fn test_static_address_unix_transport_ignores_request_extensions() {
        let transport = StaticAddressUnixTransport::<UnixStream>::new("/var/run/static.sock");

        // The transport should still use its static address, not the one from the request
        assert_eq!(transport.address(), &PathBuf::from("/var/run/static.sock"));
    }

    #[tokio::test]
    async fn test_static_address_unix_transport_with_timeout() {
        let config = UnixTransportConfig {
            connect_timeout: Some(std::time::Duration::from_millis(1)),
        };
        let transport = StaticAddressUnixTransport::<UnixStream>::with_config(
            "/nonexistent/socket.sock",
            config,
        );

        let result = tower::ServiceExt::oneshot(transport, ()).await;
        assert!(result.is_err());
        // Could be either a timeout or connection error depending on system
        match result.unwrap_err() {
            UnixConnectionError::ConnectionError(_) | UnixConnectionError::Timeout(_) => {} // Both are acceptable
            other => panic!("Unexpected error type: {other:?}"),
        }
    }

    #[test]
    fn test_static_address_unix_transport_accepts_different_path_types() {
        // Test with &str
        let transport1 = StaticAddressUnixTransport::<UnixStream>::new("/var/run/test.sock");
        assert_eq!(transport1.address(), &PathBuf::from("/var/run/test.sock"));

        // Test with String
        let transport2 =
            StaticAddressUnixTransport::<UnixStream>::new(String::from("/var/run/test.sock"));
        assert_eq!(transport2.address(), &PathBuf::from("/var/run/test.sock"));

        // Test with Utf8PathBuf
        let transport3 =
            StaticAddressUnixTransport::<UnixStream>::new(PathBuf::from("/var/run/test.sock"));
        assert_eq!(transport3.address(), &PathBuf::from("/var/run/test.sock"));
    }
}