vessel-pty 0.17.4

//! Client for communicating with the vessel server.
//!
//! Handles Unix socket connection and auto-starting the server.

use crate::has_systemd_run;
use crate::protocol::{Request, Response};
use crate::runtime::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use crate::runtime::net::UnixStream;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::Duration;
use thiserror::Error;
use tracing::{debug, info, warn};

/// Errors that can occur in the client.
#[derive(Debug, Error)]
pub enum ClientError {
    #[error("failed to connect to server: {0}")]
    Connect(#[source] std::io::Error),

    #[error("failed to send request: {0}")]
    Send(#[source] std::io::Error),

    #[error("failed to receive response: {0}")]
    Receive(#[source] std::io::Error),

    #[error("failed to serialize request: {0}")]
    Serialize(#[source] serde_json::Error),

    #[error("failed to deserialize response: {0}")]
    Deserialize(#[source] serde_json::Error),

    #[error("failed to start server: {0}")]
    ServerStart(#[source] std::io::Error),

    #[error("server did not start in time")]
    ServerTimeout,

    #[error("server returned error: {0}")]
    ServerError(String),

    #[error("connection lost")]
    ConnectionLost,
}

/// Get the default socket path for the vessel server.
///
/// Uses `/run/user/$UID/vessel.sock` if the directory exists (regardless of
/// whether `XDG_RUNTIME_DIR` is set), falling back to `/tmp/vessel-$UID.sock`.
/// This ensures all clients resolve to the same path even when environment
/// variables differ across contexts (e.g., cron, hooks, direct shell).
#[must_use]
pub fn default_socket_path() -> PathBuf {
    let uid = crate::sys::getuid();
    let runtime_dir = PathBuf::from(format!("/run/user/{uid}"));
    if runtime_dir.is_dir() {
        runtime_dir.join("vessel.sock")
    } else {
        PathBuf::from(format!("/tmp/vessel-{uid}.sock"))
    }
}

/// Client for the vessel server.
pub struct Client {
    socket_path: PathBuf,
    stream: Option<BufReader<UnixStream>>,
}

impl Client {
    /// Create a new client that will connect to the given socket path.
    #[must_use]
    pub const fn new(socket_path: PathBuf) -> Self {
        Self {
            socket_path,
            stream: None,
        }
    }

    /// Create a client with the default socket path.
    #[must_use]
    pub fn with_default_path() -> Self {
        Self::new(default_socket_path())
    }

    /// Connect to the server, starting it if necessary.
    pub async fn connect(&mut self) -> Result<(), ClientError> {
        if self.stream.is_some() {
            return Ok(());
        }

        // Try to connect directly first
        match UnixStream::connect(&self.socket_path).await {
            Ok(stream) => {
                debug!("Connected to existing server");
                self.stream = Some(BufReader::new(stream));
                return Ok(());
            }
            Err(e) => {
                debug!("Could not connect to server: {}", e);
            }
        }

        // Start the server
        self.start_server().await?;

        // Try to connect with retries
        for i in 0..50 {
            crate::runtime::time::sleep(Duration::from_millis(100)).await;
            match UnixStream::connect(&self.socket_path).await {
                Ok(stream) => {
                    info!("Connected to server after {} attempts", i + 1);
                    self.stream = Some(BufReader::new(stream));
                    return Ok(());
                }
                Err(e) => {
                    if i % 10 == 9 {
                        debug!("Still waiting for server (attempt {}): {}", i + 1, e);
                    }
                }
            }
        }

        Err(ClientError::ServerTimeout)
    }

    /// Start the server as a background process.
    ///
    /// Attempts to launch the server in its own systemd scope so it survives
    /// the death of the calling pane's cgroup. Falls back to a bare spawn if
    /// `systemd-run` is not available.
    #[allow(clippy::unused_async)] // async for API consistency with other methods
    async fn start_server(&self) -> Result<(), ClientError> {
        info!("Starting server...");

        let exe = std::env::current_exe().map_err(ClientError::ServerStart)?;

        // Try launching in a dedicated systemd scope first
        if has_systemd_run() {
            info!("Launching server via systemd-run --scope");
            let result = Command::new("systemd-run")
                .args([
                    "--user",
                    "--scope",
                    "--collect",
                    "--unit=vessel-server",
                    "--",
                ])
                .arg(&exe)
                .args(["server", "--daemon"])
                .spawn();

            match result {
                Ok(_) => return Ok(()),
                Err(e) => {
                    warn!("systemd-run failed, falling back to bare spawn: {e}");
                }
            }
        }

        // Fallback: bare spawn (inherits caller's cgroup)
        Command::new(&exe)
            .arg("server")
            .arg("--daemon")
            .spawn()
            .map_err(ClientError::ServerStart)?;

        Ok(())
    }

    /// Send a request to the server and wait for a response.
    pub async fn request(&mut self, request: Request) -> Result<Response, ClientError> {
        // Ensure we're connected
        self.connect().await?;

        let stream = self.stream.as_mut().ok_or(ClientError::ConnectionLost)?;

        // Serialize and send request
        let mut json = serde_json::to_string(&request).map_err(ClientError::Serialize)?;
        json.push('\n');

        stream
            .get_mut()
            .write_all(json.as_bytes())
            .await
            .map_err(ClientError::Send)?;

        // Read response
        let mut line = String::new();
        let n = stream
            .read_line(&mut line)
            .await
            .map_err(ClientError::Receive)?;

        if n == 0 {
            return Err(ClientError::ConnectionLost);
        }

        let response: Response = serde_json::from_str(&line).map_err(ClientError::Deserialize)?;

        // Check for server error
        if let Response::Error { message } = &response {
            warn!("Server returned error: {}", message);
        }

        Ok(response)
    }

    /// Get the socket path.
    pub fn socket_path(&self) -> &Path {
        &self.socket_path
    }
}