sail/

client.rs

//! The Sail client: the canonical async surface that owns configuration and
//! transport, shared by every binding (Python, TypeScript, CLI).
//!
//! [`Client`] is a cheap-to-clone handle (`Arc` inside, like `reqwest::Client`):
//! clone it freely to share the connection pools and config. Construct it with
//! [`Client::from_env`] or [`Client::builder`].
//!
//! Every method is `async`. Synchronous callers (the PyO3 bridge with the GIL
//! released, the CLI) drive these futures with
//! [`crate::block_on`]; an async host awaits them directly.
//!
//! ```no_run
//! # async fn run() -> Result<(), sail::error::SailError> {
//! use sail::Client;
//!
//! // From the environment (SAIL_API_KEY, SAIL_MODE, …):
//! let client = Client::from_env()?;
//! let page = client.list_sailboxes(&Default::default()).await?;
//! println!("{} sailboxes", page.items.len());
//!
//! // Or build one explicitly for the dev environment:
//! let client = Client::builder("sk_...").mode("dev").build()?;
//! let app = client.find_app("my-app", true).await?;
//! # let _ = (client, app);
//! # Ok(())
//! # }
//! ```

use std::sync::Arc;
use std::time::Duration;

use crate::app::{self, App};
use crate::config::Config;
use crate::error::SailError;
use crate::exec::{ExecOptions, ExecParams, ExecProcess};
use crate::http::HttpCore;
use crate::imagebuilder::ImageBuilder;
use crate::sailbox::api::{SailboxApi, UpgradeResult};
use crate::sailbox::object::Sailbox;
use crate::sailbox::types::{
    CreateSailboxRequest, ListSailboxesQuery, SailboxCheckpoint, SailboxHandle, SailboxInfo,
    SailboxPage, VolumeInfo,
};
use crate::worker::{FileReader, FileWriter, Listener, WorkerProxy, WriteOptions};

/// A configured Sail client. Cheap to clone; shares transport across clones.
#[derive(Clone)]
pub struct Client {
    inner: Arc<Inner>,
}

struct Inner {
    config: Config,
    /// Sailbox-API host: lifecycle, list/get, listeners, volume.
    sailbox_http: HttpCore,
    /// Central public-API host: app find, inference, voyages.
    api_http: HttpCore,
    /// Per-sailbox worker proxy: exec, files, listener reads. Its own `Arc` so
    /// the streaming file/exec methods (which take `&Arc<Self>`) can share it.
    worker: Arc<WorkerProxy>,
    imagebuilder: ImageBuilder,
}

/// Builds a [`Client`] from explicit values, falling back to a mode's endpoint
/// defaults. Prefer [`Client::from_env`] for the common env-driven case.
#[derive(Debug, Default, Clone)]
pub struct ClientBuilder {
    mode: Option<String>,
    api_key: Option<String>,
    api_url: Option<String>,
    sailbox_api_url: Option<String>,
    imagebuilder_url: Option<String>,
    ingress_url: Option<String>,
}

impl ClientBuilder {
    /// A builder with the given API key; endpoints default to prod unless a
    /// `mode` or explicit URL is set.
    pub fn new(api_key: impl Into<String>) -> ClientBuilder {
        ClientBuilder {
            api_key: Some(api_key.into()),
            ..ClientBuilder::default()
        }
    }

    /// Select the named environment (`prod`/`dev`/`staging`/`local`), which
    /// picks the endpoint defaults.
    pub fn mode(mut self, mode: impl Into<String>) -> ClientBuilder {
        self.mode = Some(mode.into());
        self
    }

    /// Override the central public-API URL.
    pub fn api_url(mut self, api_url: impl Into<String>) -> ClientBuilder {
        self.api_url = Some(api_url.into());
        self
    }

    /// Override the sailbox-API URL.
    pub fn sailbox_api_url(mut self, url: impl Into<String>) -> ClientBuilder {
        self.sailbox_api_url = Some(url.into());
        self
    }

    /// Override the imagebuilder dispatcher URL.
    pub fn imagebuilder_url(mut self, url: impl Into<String>) -> ClientBuilder {
        self.imagebuilder_url = Some(url.into());
        self
    }

    /// Override the listener ingress base URL (what `SAILBOX_INGRESS_URL`
    /// sets from the environment), for custom or local sailbox stacks.
    pub fn ingress_url(mut self, url: impl Into<String>) -> ClientBuilder {
        self.ingress_url = Some(url.into());
        self
    }

    /// Build the client, resolving any unset endpoint from the mode defaults.
    pub fn build(self) -> Result<Client, SailError> {
        let api_key = self.api_key.unwrap_or_default();
        let config = Config::resolve(
            self.mode.as_deref(),
            api_key,
            self.api_url,
            self.sailbox_api_url,
            self.imagebuilder_url,
            self.ingress_url,
        )?;
        Client::from_config(config)
    }
}

impl Client {
    /// Start a [`ClientBuilder`].
    pub fn builder(api_key: impl Into<String>) -> ClientBuilder {
        ClientBuilder::new(api_key)
    }

    /// Build a client from the environment (`SAIL_MODE`, `SAIL_API_KEY`, …).
    pub fn from_env() -> Result<Client, SailError> {
        Client::from_config(Config::from_env()?)
    }

    /// Build a client from an already-resolved [`Config`].
    pub fn from_config(config: Config) -> Result<Client, SailError> {
        let sailbox_http = HttpCore::new(&config.sailbox_api_url, &config.api_key)?;
        let api_http = HttpCore::new(&config.api_url, &config.api_key)?;
        let worker = Arc::new(WorkerProxy::new(&config.api_key)?);
        let imagebuilder = ImageBuilder::new(&config.imagebuilder_url, &config.api_key)?;
        Ok(Client {
            inner: Arc::new(Inner {
                config,
                sailbox_http,
                api_http,
                worker,
                imagebuilder,
            }),
        })
    }

    /// The resolved configuration.
    pub fn config(&self) -> &Config {
        &self.inner.config
    }

    /// The worker proxy for exec, file copy, and listener reads.
    #[doc(hidden)]
    pub fn worker(&self) -> Arc<WorkerProxy> {
        Arc::clone(&self.inner.worker)
    }

    /// The imagebuilder dispatcher client.
    #[doc(hidden)]
    pub fn imagebuilder(&self) -> &ImageBuilder {
        &self.inner.imagebuilder
    }

    /// The sailbox-API HTTP host (for binding-built requests).
    #[doc(hidden)]
    pub fn sailbox_http(&self) -> &HttpCore {
        &self.inner.sailbox_http
    }

    /// The central public-API HTTP host (for binding-built requests).
    #[doc(hidden)]
    pub fn api_http(&self) -> &HttpCore {
        &self.inner.api_http
    }

    fn sailbox_api(&self) -> SailboxApi<'_> {
        SailboxApi::new(&self.inner.sailbox_http)
    }

    // --- sailbox lifecycle ---

    /// Create a sailbox. `timeout` bounds each attempt of the synchronous
    /// create (which the scheduler can block on for minutes); the call retries
    /// with one idempotency key so the backend can dedupe rather than
    /// duplicate, and gives up after roughly `max_attempts * timeout`.
    /// An interrupted or re-invoked create is a new request and may leave a
    /// prior box behind under the same name. 10 minutes is a good default;
    /// `None` leaves each attempt unbounded. If the budget is exhausted the
    /// box may still be coming up server-side; find or terminate it by
    /// `name`.
    pub async fn create_sailbox(
        &self,
        req: &CreateSailboxRequest,
        timeout: Option<Duration>,
    ) -> Result<Sailbox, SailError> {
        let bind = |handle: SailboxHandle| Sailbox::bind(self.clone(), handle);
        if !req.ssh {
            return self.sailbox_api().create(req, timeout).await.map(bind);
        }
        // Validate the full request now, port-22 entries included: they are
        // stripped below (their allowlist applies at the enable_ssh expose),
        // so create's own validation never sees them, and an invalid entry
        // must fail here rather than after the VM exists.
        crate::sailbox::api::validate_ingress_ports(&req.ingress_ports)?;
        // SSH setup is org-scoped: preflight the org CA (created on first use)
        // so a CA outage fails before the VM exists.
        self.org_ssh_ca_public_key().await?;
        // Port 22 belongs to enable_ssh, which exposes it only after verifying
        // the CA-only sshd owns it — never the create request — so a failed
        // setup can't leave port 22 exposed. An explicit port-22 entry
        // contributes just its allowlist, applied at that expose.
        let mut req = req.clone();
        let ssh_allowlist = req
            .ingress_ports
            .iter()
            .find(|port| port.guest_port == 22)
            .map(|port| port.allowlist.clone())
            .unwrap_or_default();
        req.ingress_ports.retain(|port| port.guest_port != 22);
        let handle = self.sailbox_api().create(&req, timeout).await?;
        let handle_id = handle.sailbox_id.clone();
        // The VM is already up, so skip the readiness probe (wait: false).
        if let Err(err) = self
            .enable_ssh(&handle_id, &ssh_allowlist, false, Duration::ZERO)
            .await
        {
            // The sailbox exists; surface its id so the caller can fetch it to
            // retry enable_ssh or terminate it.
            return Err(SailError::Creation {
                message: format!(
                    "sailbox {handle_id} was created, but SSH setup failed: {err}. Fetch it by \
                     id to retry enable_ssh or terminate it."
                ),
                status: 0,
                body: serde_json::Value::Null,
            });
        }
        Ok(bind(handle))
    }

    /// Fetch a single sailbox.
    #[doc(hidden)]
    pub async fn get_sailbox(&self, sailbox_id: &str) -> Result<SailboxInfo, SailError> {
        self.sailbox_api().get(sailbox_id).await
    }

    /// List sailboxes in the current org.
    pub async fn list_sailboxes(
        &self,
        query: &ListSailboxesQuery,
    ) -> Result<SailboxPage, SailError> {
        self.sailbox_api().list(query).await
    }

    /// Terminate a sailbox (idempotent).
    #[doc(hidden)]
    pub async fn terminate_sailbox(&self, sailbox_id: &str) -> Result<(), SailError> {
        self.sailbox_api().terminate(sailbox_id).await
    }

    /// Pause a sailbox.
    #[doc(hidden)]
    pub async fn pause_sailbox(&self, sailbox_id: &str) -> Result<(), SailError> {
        self.sailbox_api().pause(sailbox_id).await
    }

    /// Sleep a sailbox.
    #[doc(hidden)]
    pub async fn sleep_sailbox(&self, sailbox_id: &str) -> Result<(), SailError> {
        self.sailbox_api().sleep(sailbox_id).await
    }

    /// Resume a paused/sleeping sailbox.
    #[doc(hidden)]
    pub async fn resume_sailbox(&self, sailbox_id: &str) -> Result<SailboxHandle, SailError> {
        self.sailbox_api().resume(sailbox_id).await
    }

    /// Checkpoint a running sailbox.
    #[doc(hidden)]
    pub async fn checkpoint_sailbox(
        &self,
        sailbox_id: &str,
        name: Option<&str>,
        ttl_seconds: Option<i64>,
    ) -> Result<SailboxCheckpoint, SailError> {
        self.sailbox_api()
            .checkpoint(sailbox_id, name, ttl_seconds)
            .await
    }

    /// Create a new sailbox from a checkpoint.
    pub async fn create_from_checkpoint(
        &self,
        checkpoint_id: &str,
        name: Option<&str>,
        timeout: Option<Duration>,
    ) -> Result<Sailbox, SailError> {
        self.sailbox_api()
            .from_checkpoint(checkpoint_id, name, timeout.map(|t| t.as_secs() as i64))
            .await
            .map(|handle| Sailbox::bind(self.clone(), handle))
    }

    /// Upgrade the in-guest agent (applies now if running, else at next wake).
    #[doc(hidden)]
    pub async fn upgrade_sailbox(&self, sailbox_id: &str) -> Result<UpgradeResult, SailError> {
        self.sailbox_api().upgrade(sailbox_id).await
    }

    /// Expose a guest port at runtime; returns the add-listener response.
    #[doc(hidden)]
    pub async fn expose_listener(
        &self,
        sailbox_id: &str,
        guest_port: u32,
        protocol: crate::sailbox::types::IngressProtocol,
        allowlist: &[String],
    ) -> Result<Listener, SailError> {
        let mut response = self
            .sailbox_api()
            .expose(sailbox_id, guest_port, protocol, allowlist)
            .await?;
        self.fill_listener_url(sailbox_id, &mut response);
        Ok(response)
    }

    /// Remove a runtime ingress port.
    #[doc(hidden)]
    pub async fn unexpose_listener(
        &self,
        sailbox_id: &str,
        guest_port: u32,
    ) -> Result<(), SailError> {
        self.sailbox_api().unexpose(sailbox_id, guest_port).await
    }

    /// List a sailbox's ingress listeners without resuming (waking) the box.
    #[doc(hidden)]
    pub async fn list_listeners(
        &self,
        sailbox_id: &str,
    ) -> Result<Vec<crate::worker::Listener>, SailError> {
        let mut listeners = self.sailbox_api().list_listeners(sailbox_id).await?;
        for listener in &mut listeners {
            self.fill_listener_url(sailbox_id, listener);
        }
        Ok(listeners)
    }

    /// Fetch one ingress listener by guest port without resuming (waking) the
    /// box; a missing port is a [`SailError::NotFound`].
    #[doc(hidden)]
    pub async fn get_listener(
        &self,
        sailbox_id: &str,
        guest_port: u32,
    ) -> Result<crate::worker::Listener, SailError> {
        let mut listener = self
            .sailbox_api()
            .get_listener(sailbox_id, guest_port)
            .await?;
        self.fill_listener_url(sailbox_id, &mut listener);
        Ok(listener)
    }

    /// Fill an empty `public_url` on a non-TCP listener with the URL
    /// synthesized from this client's ingress config (the server leaves
    /// listener URLs empty in local/path mode).
    fn fill_listener_url(&self, sailbox_id: &str, listener: &mut crate::worker::Listener) {
        if listener.public_url.is_empty()
            && listener.protocol != crate::sailbox::types::ListenerProtocol::Tcp
        {
            listener.public_url = crate::sailbox::listeners::synthesized_public_url(
                self.config(),
                sailbox_id,
                listener.guest_port,
            );
        }
    }

    /// Ingress-identity headers for this sailbox.
    #[doc(hidden)]
    pub async fn ingress_auth_headers(
        &self,
        sailbox_id: &str,
    ) -> Result<Vec<(String, String)>, SailError> {
        self.sailbox_api().ingress_auth_headers(sailbox_id).await
    }

    /// The caller org's SSH CA public key (created on first use).
    pub async fn org_ssh_ca_public_key(&self) -> Result<String, SailError> {
        self.sailbox_api().org_ssh_ca_public_key().await
    }

    /// Sign `public_key` into a short-lived org-CA certificate (principal
    /// `root`). `timeout` (seconds) bounds a single no-retry attempt; `None`
    /// retries.
    pub async fn issue_user_cert(
        &self,
        public_key: &str,
        timeout: Option<Duration>,
    ) -> Result<crate::sailbox::types::IssuedUserCert, SailError> {
        self.sailbox_api()
            .issue_user_cert(public_key, timeout.map(|t| t.as_secs_f64()))
            .await
    }

    // --- NFS volumes ---

    /// Look up (optionally minting) an NFS volume by name.
    pub async fn get_volume(
        &self,
        name: &str,
        mint_if_missing: bool,
    ) -> Result<VolumeInfo, SailError> {
        self.sailbox_api().get_volume(name, mint_if_missing).await
    }

    /// List NFS volumes in the current org.
    pub async fn list_volumes(
        &self,
        max_objects: Option<i64>,
    ) -> Result<Vec<VolumeInfo>, SailError> {
        self.sailbox_api().list_volumes(max_objects).await
    }

    /// Delete a volume by id.
    pub async fn delete_volume(
        &self,
        volume_id: &str,
        allow_missing: bool,
    ) -> Result<Option<VolumeInfo>, SailError> {
        self.sailbox_api()
            .delete_volume(volume_id, allow_missing)
            .await
    }

    // --- apps (central API) ---

    /// Find an app by name on the central API, optionally minting it.
    pub async fn find_app(&self, name: &str, mint_if_missing: bool) -> Result<App, SailError> {
        app::find_app(&self.inner.api_http, name, mint_if_missing).await
    }

    /// Every app the current org owns, newest first, from the central app index.
    pub async fn list_apps(&self) -> Result<Vec<App>, SailError> {
        app::list_apps(&self.inner.api_http).await
    }

    // --- exec and files (per-sailbox worker proxy) ---

    /// Resolve a sailbox's current worker-proxy endpoint.
    ///
    /// `resume` wakes a paused/sleeping sailbox and returns its *current*
    /// endpoint, which is the host worker's address and changes when the sailbox
    /// migrates (e.g. after preemption). The GET sailbox API omits this routing
    /// field, so resuming is the only way to learn it, and resolving it fresh per
    /// call avoids ever dialing a stale worker.
    #[doc(hidden)]
    pub async fn exec_endpoint(&self, sailbox_id: &str) -> Result<String, SailError> {
        let handle = self.resume_sailbox(sailbox_id).await?;
        if handle.exec_endpoint.is_empty() {
            return Err(SailError::Internal {
                message: format!("sailbox {sailbox_id} resumed without an exec endpoint"),
            });
        }
        Ok(handle.exec_endpoint)
    }

    /// Id-form of [`Sailbox::exec`](crate::Sailbox::exec), which documents the
    /// contract. Spawns the output pump on the calling task's tokio runtime.
    #[doc(hidden)]
    pub async fn exec(
        &self,
        sailbox_id: &str,
        argv: Vec<String>,
        options: ExecOptions,
    ) -> Result<ExecProcess, SailError> {
        if argv.is_empty() {
            return Err(SailError::InvalidArgument {
                message: "command must be non-empty".to_string(),
            });
        }
        if options.cwd.is_some() || options.background {
            return Err(SailError::InvalidArgument {
                message: "cwd and background require a shell command; use exec_shell".to_string(),
            });
        }
        let exec_endpoint = self.exec_endpoint(sailbox_id).await?;
        let params = ExecParams {
            sailbox_id: sailbox_id.to_string(),
            exec_endpoint,
            argv,
            // The wire is whole seconds where 0 means "no limit", so a set
            // sub-second timeout rounds up to 1s rather than collapsing to 0.
            timeout_seconds: options
                .timeout
                .map_or(0, |d| d.as_secs_f64().ceil().max(1.0) as u32),
            idempotency_key: options.idempotency_key,
            // A pty always feeds keystrokes to the command, so it implies an
            // open stdin regardless of the flag.
            open_stdin: options.open_stdin || options.pty,
            pty: options.pty,
            term: options.term,
            cols: options.cols,
            rows: options.rows,
            retry_timeout: options.retry_timeout.as_secs_f64(),
            extra_metadata: Vec::new(),
        };
        ExecProcess::start(self.worker(), params).await
    }

    /// Run a shell command in a sailbox via `/bin/sh -lc`, honoring the
    /// `cwd`/`background` conveniences in [`ExecOptions`]. See [`Client::exec`]
    /// for the argv form and the runtime notes.
    #[doc(hidden)]
    pub async fn exec_shell(
        &self,
        sailbox_id: &str,
        command: &str,
        mut options: ExecOptions,
    ) -> Result<ExecProcess, SailError> {
        let argv = crate::exec::shell_argv(command, &options)?;
        // The conveniences are baked into the argv now; clear them so the argv
        // path's guard does not re-reject them.
        options.cwd = None;
        options.background = false;
        self.exec(sailbox_id, argv, options).await
    }

    /// Open a streaming read of a guest file. Resumes (wakes) the sailbox to
    /// reach it; the returned [`FileReader`] yields chunks until end of file.
    ///
    /// # Runtime
    ///
    /// Spawns the read pump on the calling task's tokio runtime (see
    /// [`crate::worker::WorkerProxy::read_file`]).
    #[doc(hidden)]
    pub async fn read_stream(
        &self,
        sailbox_id: &str,
        remote_path: &str,
    ) -> Result<FileReader, SailError> {
        let endpoint = self.exec_endpoint(sailbox_id).await?;
        Ok(self
            .inner
            .worker
            .read_file(&endpoint, sailbox_id, remote_path))
    }

    /// Read a guest file into memory in one call (convenience over
    /// [`Client::read_stream`], which streams a large file without
    /// buffering it whole).
    #[doc(hidden)]
    pub async fn read_file(
        &self,
        sailbox_id: &str,
        remote_path: &str,
    ) -> Result<Vec<u8>, SailError> {
        let reader = self.read_stream(sailbox_id, remote_path).await?;
        let mut contents = Vec::new();
        while let Some(chunk) = reader.next().await {
            contents.extend_from_slice(&chunk?);
        }
        Ok(contents)
    }

    /// Open a streaming write to a guest file. Resumes (wakes) the sailbox to
    /// reach it; feed the returned [`FileWriter`] with `write_chunk` and end with
    /// `finish`, so a large source is never buffered whole.
    ///
    /// # Runtime
    ///
    /// Spawns the write RPC on the calling task's tokio runtime (see
    /// [`crate::worker::WorkerProxy::write_file`]).
    #[doc(hidden)]
    pub async fn write_stream(
        &self,
        sailbox_id: &str,
        remote_path: &str,
        options: WriteOptions,
    ) -> Result<FileWriter, SailError> {
        let endpoint = self.exec_endpoint(sailbox_id).await?;
        Ok(self.inner.worker.write_file(
            &endpoint,
            sailbox_id,
            remote_path,
            options.create_parents,
            options.mode,
        ))
    }

    /// Write `data` to a guest file in one call (convenience over
    /// [`Client::write_stream`], which streams a large source without
    /// buffering it whole).
    #[doc(hidden)]
    pub async fn write_file(
        &self,
        sailbox_id: &str,
        remote_path: &str,
        data: &[u8],
        options: WriteOptions,
    ) -> Result<(), SailError> {
        let mut writer = self.write_stream(sailbox_id, remote_path, options).await?;
        writer.write(data).await?;
        writer.finish().await
    }
}