sail/

client.rs

//! The Sail client: the canonical async surface that owns configuration and
//! transport, shared by every binding (Python, CLI, future hosts).
//!
//! [`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, UpgradeOutcome};
use crate::sailbox::types::{
    AddListenerResponse, CreateSailboxRequest, ListQuery, NfsVolume, SailboxCheckpoint,
    SailboxHandle, SailboxInfo, SailboxPage,
};
use crate::worker::{FileReader, FileWriter, UploadOptions, WorkerProxy};

/// 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>,
}

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
    }

    /// 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,
            // The builder targets a mode or explicit endpoints; listener ingress
            // follows the mode default.
            None,
        )?;
        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. `create_timeout` bounds each attempt of the synchronous
    /// create (which the scheduler can block on for minutes); the call still
    /// retries, reattaching to the same box, so it returns as soon as the box is
    /// ready and gives up after roughly `max_attempts * create_timeout`. `None`
    /// leaves each attempt unbounded. See
    /// [`SailboxApi::create`](crate::internal::SailboxApi::create).
    pub async fn create_sailbox(
        &self,
        req: &CreateSailboxRequest,
        create_timeout: Option<Duration>,
    ) -> Result<SailboxHandle, SailError> {
        self.sailbox_api().create(req, create_timeout).await
    }

    /// Fetch a single sailbox.
    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: &ListQuery) -> Result<SailboxPage, SailError> {
        self.sailbox_api().list(query).await
    }

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

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

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

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

    /// Checkpoint a running sailbox.
    pub async fn checkpoint_sailbox(
        &self,
        sailbox_id: &str,
    ) -> Result<SailboxCheckpoint, SailError> {
        self.sailbox_api().checkpoint(sailbox_id).await
    }

    /// Create a new sailbox from a checkpoint.
    pub async fn create_from_checkpoint(
        &self,
        checkpoint_id: &str,
        name: Option<&str>,
        timeout_seconds: Option<i64>,
    ) -> Result<SailboxHandle, SailError> {
        self.sailbox_api()
            .from_checkpoint(checkpoint_id, name, timeout_seconds)
            .await
    }

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

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

    /// Remove a runtime ingress port.
    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.
    pub async fn list_listeners(
        &self,
        sailbox_id: &str,
    ) -> Result<Vec<crate::worker::Listener>, SailError> {
        self.sailbox_api().list_listeners(sailbox_id).await
    }

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

    /// Ingress-identity headers for this sailbox.
    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<f64>,
    ) -> Result<crate::sailbox::types::IssuedUserCert, SailError> {
        self.sailbox_api()
            .issue_user_cert(public_key, timeout)
            .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<NfsVolume, 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<NfsVolume>, 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<NfsVolume>, 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.
    pub(crate) 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)
    }

    /// Run a command in a sailbox and return a handle to the live process.
    ///
    /// Resumes (wakes) the sailbox to reach it. The returned [`ExecProcess`]
    /// streams output, accepts stdin, and resolves the exit status; dropping it
    /// detaches without killing the command.
    ///
    /// # Runtime
    ///
    /// Spawns the output pump on the calling task's tokio runtime (see
    /// [`ExecProcess::start`]).
    pub async fn exec(
        &self,
        sailbox_id: &str,
        argv: Vec<String>,
        options: ExecOptions,
    ) -> Result<ExecProcess, SailError> {
        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,
            open_stdin: options.open_stdin,
            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
    }

    /// 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`]).
    pub async fn download_file(
        &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))
    }

    /// 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`]).
    pub async fn upload_file(
        &self,
        sailbox_id: &str,
        remote_path: &str,
        options: UploadOptions,
    ) -> 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,
        ))
    }
}