crdhcpc 0.1.1

/// Unix socket client for JSON-RPC communication
///
/// Client used by crrouter_web to communicate with crrouterd daemon
use super::{
    credentials::PeerCredentials,
    error::{ProtocolError, ProtocolResult},
    framing::MessageFraming,
    jsonrpc::{JsonRpcRequest, JsonRpcMessage},
};
use serde_json::Value;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tokio::{
    io::AsyncWriteExt,
    net::UnixStream,
    sync::Mutex,
    time::timeout,
};

/// Client configuration
#[derive(Debug, Clone)]
pub struct ClientConfig {
    /// Path to Unix socket
    pub socket_path: String,
    /// Request timeout
    pub timeout: Duration,
    /// Maximum retries on connection failure
    pub max_retries: u32,
    /// Retry delay
    pub retry_delay: Duration,
}

impl Default for ClientConfig {
    fn default() -> Self {
        Self {
            socket_path: "/var/run/crrouterd.sock".to_string(),
            timeout: Duration::from_secs(30),
            max_retries: 3,
            retry_delay: Duration::from_millis(100),
        }
    }
}

impl ClientConfig {
    /// Create configuration from environment variables
    pub fn from_env() -> Self {
        Self {
            socket_path: std::env::var("DAEMON_SOCKET_PATH")
                .unwrap_or_else(|_| "/var/run/crrouterd.sock".to_string()),
            timeout: std::env::var("DAEMON_TIMEOUT")
                .ok()
                .and_then(|v| v.parse().ok())
                .map(Duration::from_secs)
                .unwrap_or(Duration::from_secs(30)),
            max_retries: std::env::var("DAEMON_MAX_RETRIES")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(3),
            retry_delay: Duration::from_millis(100),
        }
    }
}

/// Unix socket client
pub struct UnixClient {
    config: ClientConfig,
    stream: Arc<Mutex<Option<UnixStream>>>,
    request_id: Arc<Mutex<i64>>,
}

impl UnixClient {
    /// Create a new client with configuration
    pub fn new(config: ClientConfig) -> Self {
        Self {
            config,
            stream: Arc::new(Mutex::new(None)),
            request_id: Arc::new(Mutex::new(0)),
        }
    }

    /// Create client with default configuration
    pub fn with_socket_path(socket_path: impl Into<String>) -> Self {
        Self::new(ClientConfig {
            socket_path: socket_path.into(),
            ..Default::default()
        })
    }

    /// Connect to the daemon
    pub async fn connect(&self) -> ProtocolResult<()> {
        let path = Path::new(&self.config.socket_path);

        if !path.exists() {
            return Err(ProtocolError::Internal(format!(
                "Socket path does not exist: {}",
                self.config.socket_path
            )));
        }

        let stream = timeout(
            self.config.timeout,
            UnixStream::connect(&self.config.socket_path),
        )
        .await
        .map_err(|_| ProtocolError::Timeout(self.config.timeout))?
        .map_err(ProtocolError::Io)?;

        *self.stream.lock().await = Some(stream);

        Ok(())
    }

    /// Ensure connection is established
    async fn ensure_connected(&self) -> ProtocolResult<()> {
        let stream_guard = self.stream.lock().await;

        if stream_guard.is_none() {
            drop(stream_guard); // Release lock before connecting
            self.connect().await?;
        }

        Ok(())
    }

    /// Generate next request ID
    async fn next_id(&self) -> i64 {
        let mut id = self.request_id.lock().await;
        *id += 1;
        *id
    }

    /// Send a JSON-RPC request and wait for response
    pub async fn call(
        &self,
        method: impl Into<String>,
        params: Option<Value>,
    ) -> ProtocolResult<Value> {
        self.ensure_connected().await?;

        let id = self.next_id().await;
        let request = JsonRpcRequest::new(method, params, Value::Number(id.into()));

        // Send request
        let mut stream_guard = self.stream.lock().await;
        let stream = stream_guard
            .as_mut()
            .ok_or(ProtocolError::ConnectionClosed)?;

        timeout(
            self.config.timeout,
            MessageFraming::send_json(stream, &request),
        )
        .await
        .map_err(|_| ProtocolError::Timeout(self.config.timeout))?
        .map_err(ProtocolError::Io)?;

        // Receive response
        let message: JsonRpcMessage = timeout(
            self.config.timeout,
            MessageFraming::recv_json(stream),
        )
        .await
        .map_err(|_| ProtocolError::Timeout(self.config.timeout))?
        .map_err(ProtocolError::Io)?;

        drop(stream_guard);

        // Parse response
        match message {
            JsonRpcMessage::Response(resp) => {
                if resp.id != Value::Number(id.into()) {
                    return Err(ProtocolError::InvalidFormat(
                        "Response ID mismatch".to_string(),
                    ));
                }
                Ok(resp.result)
            }
            JsonRpcMessage::ErrorResponse(err) => {
                Err(ProtocolError::from_jsonrpc_error(
                    err.error.code,
                    err.error.message,
                ))
            }
            _ => Err(ProtocolError::InvalidFormat(
                "Unexpected message type".to_string(),
            )),
        }
    }

    /// Send a notification (no response expected)
    pub async fn notify(
        &self,
        method: impl Into<String>,
        params: Option<Value>,
    ) -> ProtocolResult<()> {
        self.ensure_connected().await?;

        let notification = super::jsonrpc::JsonRpcNotification::new(method, params);

        let mut stream_guard = self.stream.lock().await;
        let stream = stream_guard
            .as_mut()
            .ok_or(ProtocolError::ConnectionClosed)?;

        timeout(
            self.config.timeout,
            MessageFraming::send_json(stream, &notification),
        )
        .await
        .map_err(|_| ProtocolError::Timeout(self.config.timeout))?
        .map_err(ProtocolError::Io)?;

        Ok(())
    }

    /// Get peer credentials of the daemon
    pub async fn peer_credentials(&self) -> ProtocolResult<PeerCredentials> {
        self.ensure_connected().await?;

        let stream_guard = self.stream.lock().await;
        let stream = stream_guard
            .as_ref()
            .ok_or(ProtocolError::ConnectionClosed)?;

        PeerCredentials::from_socket(stream)
    }

    /// Close the connection
    pub async fn disconnect(&self) -> ProtocolResult<()> {
        let mut stream_guard = self.stream.lock().await;

        if let Some(mut stream) = stream_guard.take() {
            stream.shutdown().await.map_err(ProtocolError::Io)?;
        }

        Ok(())
    }

    /// Check if connected
    pub async fn is_connected(&self) -> bool {
        self.stream.lock().await.is_some()
    }

    /// Call with automatic retry on connection failure
    pub async fn call_with_retry(
        &self,
        method: impl Into<String> + Clone,
        params: Option<Value>,
    ) -> ProtocolResult<Value> {
        let mut last_error = None;

        for attempt in 0..=self.config.max_retries {
            match self.call(method.clone(), params.clone()).await {
                Ok(result) => return Ok(result),
                Err(e) => {
                    last_error = Some(e);

                    // Reconnect on connection errors
                    if matches!(
                        last_error,
                        Some(ProtocolError::ConnectionClosed | ProtocolError::Io(_))
                    ) {
                        self.disconnect().await.ok();

                        if attempt < self.config.max_retries {
                            tokio::time::sleep(self.config.retry_delay).await;
                            continue;
                        }
                    }

                    break;
                }
            }
        }

        Err(last_error.unwrap_or(ProtocolError::Internal(
            "Unexpected error in retry loop".to_string(),
        )))
    }
}

impl Drop for UnixClient {
    fn drop(&mut self) {
        // Attempt graceful shutdown
        // Note: This is best-effort since we can't await in Drop
        if let Some(stream) = self.stream.try_lock().ok().and_then(|mut g| g.take()) {
            // Stream will be closed when dropped
            drop(stream);
        }
    }
}

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

    #[test]
    fn test_client_config_default() {
        let config = ClientConfig::default();
        assert_eq!(config.socket_path, "/var/run/crrouterd.sock");
        assert_eq!(config.timeout, Duration::from_secs(30));
        assert_eq!(config.max_retries, 3);
    }

    #[test]
    fn test_client_creation() {
        let client = UnixClient::with_socket_path("/tmp/test.sock");
        assert_eq!(client.config.socket_path, "/tmp/test.sock");
    }

    #[tokio::test]
    async fn test_next_id() {
        let client = UnixClient::with_socket_path("/tmp/test.sock");

        let id1 = client.next_id().await;
        let id2 = client.next_id().await;
        let id3 = client.next_id().await;

        assert_eq!(id1, 1);
        assert_eq!(id2, 2);
        assert_eq!(id3, 3);
    }
}