sail-rs 0.2.10

//! Typed client over the sailbox HTTP API (lifecycle, listeners, and volumes).
//!
//! Each method builds the request body, runs the HTTP call (with retry +
//! Idempotency-Key from [`HttpCore`]), maps a non-2xx response onto the
//! canonical [`SailError`] taxonomy, validates the echoed status, and
//! deserializes the body into a typed struct.

use std::time::Duration;

use serde_json::{json, Value};

use crate::error::SailError;
use crate::http::{HttpCore, IdempotencyKey, Method, RequestSpec};
use crate::retry::{RetryPolicy, DEFAULT_RETRY_POLICY, NO_RETRY};
use crate::sailbox::types::{
    AddListenerResponse, CreateSailboxRequest, IngressProtocol, IssuedUserCert, ListQuery,
    NfsVolume, SailboxCheckpoint, SailboxHandle, SailboxInfo, SailboxPage, SailboxStatus,
    MAX_SAILBOX_DISK_GIB, MAX_SAILBOX_MEMORY_MIB, MAX_SAILBOX_VCPUS, MIN_SAILBOX_MEMORY_MIB,
};
use crate::worker::Listener;

/// Validate the optional create-time resource limits against the platform bounds
/// ([`MAX_SAILBOX_VCPUS`], etc.). Bindings call this to fail fast with a clear
/// error before doing any work; `create` also calls it so every path is covered.
/// `0` (and `None`) means "use the platform default" and is always accepted.
pub fn validate_resource_limits(
    cpu: Option<i64>,
    memory_mib: Option<i64>,
    state_disk_size_gib: Option<i64>,
) -> Result<(), SailError> {
    let invalid = |message: String| Err(SailError::InvalidArgument { message });
    if let Some(cpu) = cpu {
        if cpu < 0 {
            return invalid("max_cpu must be at least 0".to_string());
        }
        if cpu > MAX_SAILBOX_VCPUS {
            return invalid(format!("max_cpu must be at most {MAX_SAILBOX_VCPUS}"));
        }
    }
    if let Some(memory) = memory_mib {
        if memory != 0 && memory < MIN_SAILBOX_MEMORY_MIB {
            return invalid(format!(
                "max_memory_mib must be at least {MIN_SAILBOX_MEMORY_MIB}"
            ));
        }
        if memory > MAX_SAILBOX_MEMORY_MIB {
            return invalid(format!(
                "max_memory_mib must be at most {MAX_SAILBOX_MEMORY_MIB}"
            ));
        }
    }
    if let Some(disk) = state_disk_size_gib {
        if disk < 0 {
            return invalid("max_disk_gib must be at least 0".to_string());
        }
        if disk > MAX_SAILBOX_DISK_GIB {
            return invalid(format!(
                "max_disk_gib must be at most {MAX_SAILBOX_DISK_GIB}"
            ));
        }
    }
    Ok(())
}

/// Result of an in-guest agent upgrade.
#[derive(Debug, Clone, serde::Serialize)]
pub struct UpgradeOutcome {
    /// True when applied immediately (running box); false when deferred to the
    /// next wake.
    pub applied: bool,
    /// Lifecycle status of the sailbox after the upgrade call.
    pub status: SailboxStatus,
}

/// Typed client over the sailbox HTTP API host.
pub struct SailboxApi<'a> {
    http: &'a HttpCore,
}

impl<'a> SailboxApi<'a> {
    /// Build a client bound to a sailbox-API `HttpCore`.
    pub fn new(http: &'a HttpCore) -> SailboxApi<'a> {
        SailboxApi { http }
    }

    /// Create a sailbox (`POST /v1/sailboxes`) and wait for it to come up.
    ///
    /// Optional CPU, memory, and state-disk sizes are included only when set.
    /// A non-2xx response or an echoed `failed` status maps to
    /// [`SailError::Creation`]; the returned handle requires a `running` status.
    ///
    /// `create_timeout` bounds each attempt: this is a synchronous long-poll the
    /// scheduler can block on for many minutes (capacity queue + VM boot), so
    /// without a bound the call can hang. Create is still retried (reattaching to
    /// the same box via its idempotency key), so it returns as soon as the box is
    /// ready and gives up after the policy's attempts, roughly
    /// `max_attempts * create_timeout`. `None` leaves each attempt unbounded. If
    /// the budget is exhausted the box may still be coming up server-side; the
    /// caller never saw its id, so find or terminate it by `name`.
    pub async fn create(
        &self,
        req: &CreateSailboxRequest,
        create_timeout: Option<Duration>,
    ) -> Result<SailboxHandle, SailError> {
        validate_resource_limits(req.cpu, req.memory_mib, req.state_disk_size_gib)?;
        // Serialize the typed ImageSpec to canonical proto-JSON (camelCase field
        // names and enum value names, via its serde attributes): an ordinary JSON
        // object the backend decodes with protojson, no shape-matching by hand.
        let image = serde_json::to_value(&req.image).map_err(|e| SailError::Internal {
            message: format!("failed to serialize image spec: {e}"),
        })?;
        let mut body = json!({
            "app_id": req.app_id,
            "name": req.name,
            "ingress_ports": req.ingress_ports,
            "volume_mounts": req.volume_mounts,
            "image": image,
        });
        if let Some(cpu) = req.cpu {
            body["cpu"] = json!(cpu);
        }
        if let Some(mem) = req.memory_mib {
            body["memory_mib"] = json!(mem);
        }
        if let Some(disk) = req.state_disk_size_gib {
            body["state_disk_size_gib"] = json!(disk);
        }
        if let Some(timeout) = req.autosleep_timeout_min {
            body["autosleep_timeout_min"] = json!(timeout);
        }
        let bytes = serde_json::to_vec(&body).map_err(|e| SailError::Internal {
            message: format!("failed to serialize request body: {e}"),
        })?;
        let (status, data) = self
            .request(
                Method::Post,
                "/v1/sailboxes",
                &[],
                Some(bytes),
                DEFAULT_RETRY_POLICY,
                create_timeout.map(|d| d.as_secs_f64()),
            )
            .await?;
        // create's documented failure type is Creation; only auth stays
        // PermissionDenied.
        raise_for_create_status(status, &data)?;
        if status_of(&data) == SailboxStatus::Failed {
            return Err(SailError::Creation {
                message: format!(
                    "Sailbox creation failed: {}",
                    data.get("error_message")
                        .and_then(Value::as_str)
                        .unwrap_or("")
                ),
                status,
                body: data,
            });
        }
        require_status(&data, SailboxStatus::Running, status).map_err(|_| SailError::Creation {
            message: format!(
                "Sailbox creation returned unexpected status: {}",
                data.get("status").and_then(Value::as_str).unwrap_or("")
            ),
            status,
            body: data.clone(),
        })?;
        Ok(handle_from(&data, &req.name))
    }

    /// Fetch the current state of one sailbox (`GET /v1/sailboxes/{id}`).
    pub async fn get(&self, sailbox_id: &str) -> Result<SailboxInfo, SailError> {
        let (status, data) = self
            .get_request(&format!("/v1/sailboxes/{sailbox_id}"), &[])
            .await?;
        raise_api_error(status, &data, &format!("Sailbox {sailbox_id:?}"))?;
        info_from(data)
    }

    /// List sailboxes (`GET /v1/sailboxes`) with paging and optional filters.
    ///
    /// App, status, search, and max-guest-schema-version filters are sent only
    /// when present; returns one page plus the paging cursor fields.
    pub async fn list(&self, query: &ListQuery) -> Result<SailboxPage, SailError> {
        let mut params: Vec<(String, String)> = vec![
            ("limit".to_string(), query.limit.to_string()),
            ("offset".to_string(), query.offset.to_string()),
        ];
        if let Some(app) = &query.app {
            params.push(("app".to_string(), app.clone()));
        }
        if let Some(s) = &query.status {
            params.push(("status".to_string(), s.as_str().to_string()));
        }
        if let Some(s) = &query.search {
            params.push(("search".to_string(), s.clone()));
        }
        if let Some(v) = query.max_guest_schema_version {
            params.push(("max_guest_schema_version".to_string(), v.to_string()));
        }
        let (status, data) = self.get_request("/v1/sailboxes", &params).await?;
        raise_api_error(status, &data, "")?;
        let items = data
            .get("data")
            .and_then(Value::as_array)
            .ok_or_else(|| missing_field("data"))?
            .iter()
            .cloned()
            .map(info_from)
            .collect::<Result<Vec<_>, _>>()?;
        // Tolerate a missing pagination counter: echo the request's limit/offset
        // and fall back to the page size for total, so one absent field does not
        // sink the whole list.
        let total = int_field(&data, "total").unwrap_or(items.len() as i64);
        Ok(SailboxPage {
            limit: int_field(&data, "limit").unwrap_or(query.limit),
            offset: int_field(&data, "offset").unwrap_or(query.offset),
            total,
            has_more: data
                .get("has_more")
                .and_then(Value::as_bool)
                .unwrap_or(false),
            items,
        })
    }

    /// Idempotent: a resource-miss 404 is the same outcome as a fresh terminate.
    pub async fn terminate(&self, sailbox_id: &str) -> Result<(), SailError> {
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/terminate"), &json!({}))
            .await?;
        if status == 404 && is_resource_not_found(&data) {
            return Ok(());
        }
        raise_api_error(status, &data, "")
    }

    /// Pause a sailbox (`POST /v1/sailboxes/{id}/pause`), keeping it in memory.
    ///
    /// Requires the echoed status to be `paused`.
    pub async fn pause(&self, sailbox_id: &str) -> Result<(), SailError> {
        self.stop(sailbox_id, "pause", SailboxStatus::Paused).await
    }

    /// Sleep a sailbox (`POST /v1/sailboxes/{id}/sleep`), persisting it to disk.
    ///
    /// Requires the echoed status to be `sleeping`.
    pub async fn sleep(&self, sailbox_id: &str) -> Result<(), SailError> {
        self.stop(sailbox_id, "sleep", SailboxStatus::Sleeping)
            .await
    }

    async fn stop(
        &self,
        sailbox_id: &str,
        action: &str,
        expected: SailboxStatus,
    ) -> Result<(), SailError> {
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/{action}"), &json!({}))
            .await?;
        raise_api_error(status, &data, "")?;
        require_status(&data, expected, status)
    }

    /// Resume a paused or sleeping sailbox (`POST /v1/sailboxes/{id}/resume`).
    ///
    /// A `terminal_unavailable` resume state maps to [`SailError::NotFound`]
    /// since a terminated sailbox cannot be brought back; otherwise requires a
    /// `running` status and returns a fresh handle.
    pub async fn resume(&self, sailbox_id: &str) -> Result<SailboxHandle, SailError> {
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/resume"), &json!({}))
            .await?;
        raise_api_error(status, &data, "")?;
        if data.get("resume_state").and_then(Value::as_str) == Some("terminal_unavailable") {
            // A terminated sailbox is gone for good; NotFound maps to the
            // binding's lookup error.
            return Err(SailError::NotFound {
                message: data
                    .get("error_message")
                    .and_then(Value::as_str)
                    .filter(|s| !s.is_empty())
                    .unwrap_or("sailbox cannot be resumed")
                    .to_string(),
            });
        }
        require_status(&data, SailboxStatus::Running, status)?;
        Ok(handle_from(&data, ""))
    }

    /// Checkpoint a running sailbox (`POST /v1/sailboxes/{id}/checkpoint`).
    ///
    /// Returns the new checkpoint's id and generation; a response missing
    /// `checkpoint_id` is treated as an [`SailError::Api`] failure.
    pub async fn checkpoint(&self, sailbox_id: &str) -> Result<SailboxCheckpoint, SailError> {
        let (status, data) = self
            .post(
                &format!("/v1/sailboxes/{sailbox_id}/checkpoint"),
                &json!({}),
            )
            .await?;
        raise_api_error(status, &data, "")?;
        let checkpoint_id = data
            .get("checkpoint_id")
            .and_then(Value::as_str)
            .filter(|s| !s.is_empty());
        let checkpoint_id = if let Some(id) = checkpoint_id {
            id.to_string()
        } else {
            require_status(&data, SailboxStatus::Running, status)?;
            return Err(SailError::Api {
                status,
                message: "checkpoint sailbox did not return checkpoint_id".to_string(),
                body: data,
            });
        };
        Ok(SailboxCheckpoint {
            checkpoint_id,
            sailbox_id: data
                .get("sailbox_id")
                .and_then(Value::as_str)
                .unwrap_or(sailbox_id)
                .to_string(),
            checkpoint_generation: data
                .get("checkpoint_generation")
                .and_then(Value::as_i64)
                .unwrap_or(0),
            status: status_of(&data),
        })
    }

    /// Create a sailbox from an existing checkpoint
    /// (`POST /v1/sailboxes/from_checkpoint`).
    ///
    /// `checkpoint_id` must be non-empty and `timeout_seconds`, when given, must
    /// be positive; both are validated as [`SailError::InvalidArgument`] before
    /// the call. Requires a `running` status and returns the new sailbox handle.
    pub async fn from_checkpoint(
        &self,
        checkpoint_id: &str,
        name: Option<&str>,
        timeout_seconds: Option<i64>,
    ) -> Result<SailboxHandle, SailError> {
        if checkpoint_id.is_empty() {
            return Err(SailError::InvalidArgument {
                message: "checkpoint_id is required".to_string(),
            });
        }
        let mut body = json!({ "checkpoint_id": checkpoint_id });
        if let Some(name) = name {
            body["name"] = json!(name);
        }
        if let Some(timeout) = timeout_seconds {
            if timeout <= 0 {
                return Err(SailError::InvalidArgument {
                    message: "timeout_seconds must be greater than 0".to_string(),
                });
            }
            body["timeout_seconds"] = json!(timeout);
        }
        let (status, data) = self.post("/v1/sailboxes/from_checkpoint", &body).await?;
        raise_api_error(status, &data, "")?;
        require_status(&data, SailboxStatus::Running, status)?;
        Ok(handle_from(&data, name.unwrap_or("")))
    }

    /// Upgrade a sailbox's in-guest agent (`POST /v1/sailboxes/{id}/upgrade`).
    ///
    /// The returned [`UpgradeOutcome`] reports whether the upgrade applied
    /// immediately (running box) or was deferred to the next wake.
    pub async fn upgrade(&self, sailbox_id: &str) -> Result<UpgradeOutcome, SailError> {
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/upgrade"), &json!({}))
            .await?;
        raise_api_error(status, &data, "")?;
        Ok(UpgradeOutcome {
            applied: data
                .get("applied")
                .and_then(Value::as_bool)
                .unwrap_or(false),
            status: status_of(&data),
        })
    }

    /// Expose a runtime ingress port; returns the resolved [`AddListenerResponse`].
    pub async fn expose(
        &self,
        sailbox_id: &str,
        guest_port: u32,
        protocol: IngressProtocol,
        allowlist: &[String],
    ) -> Result<AddListenerResponse, SailError> {
        // Always send allowlist (even empty) so re-exposing a port can clear it
        // and the request is explicit rather than relying on absence semantics.
        let body = json!({
            "guest_port": guest_port,
            "protocol": protocol.as_str(),
            "allowlist": allowlist,
        });
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/listeners"), &body)
            .await?;
        raise_api_error(status, &data, "")?;
        serde_json::from_value(data).map_err(|e| SailError::Internal {
            message: format!("failed to parse add-listener response: {e}"),
        })
    }

    /// Remove a runtime ingress listener
    /// (`DELETE /v1/sailboxes/{id}/listeners/{guest_port}`).
    ///
    /// Runs without retry so a lost response on a committed delete does not
    /// re-issue the call and surface an already-unexposed port as an error.
    pub async fn unexpose(&self, sailbox_id: &str, guest_port: u32) -> Result<(), SailError> {
        // No retry: a committed DELETE whose response is lost would, on retry,
        // hit the scheduler as an already-unexposed port (NotFound) and raise
        // even though the port was removed.
        let (status, data) = self
            .request(
                Method::Delete,
                &format!("/v1/sailboxes/{sailbox_id}/listeners/{guest_port}"),
                &[],
                None,
                NO_RETRY,
                None,
            )
            .await?;
        raise_api_error(status, &data, "")
    }

    /// List a sailbox's ingress listeners
    /// (`GET /v1/sailboxes/{id}/listeners`). Served from the control plane, so
    /// it does not resume (wake) a paused or sleeping box.
    pub async fn list_listeners(&self, sailbox_id: &str) -> Result<Vec<Listener>, SailError> {
        let (status, data) = self
            .get_request(&format!("/v1/sailboxes/{sailbox_id}/listeners"), &[])
            .await?;
        raise_api_error(status, &data, &format!("Sailbox {sailbox_id:?}"))?;
        let rows = data.get("data").cloned().unwrap_or(Value::Null);
        serde_json::from_value(rows).map_err(|e| SailError::Internal {
            message: format!("failed to parse listeners: {e}"),
        })
    }

    /// Fetch one ingress listener by guest port
    /// (`GET /v1/sailboxes/{id}/listeners/{guest_port}`); a missing port is a
    /// [`SailError::NotFound`]. Served without resuming the box.
    pub async fn get_listener(
        &self,
        sailbox_id: &str,
        guest_port: u32,
    ) -> Result<Listener, SailError> {
        let (status, data) = self
            .get_request(
                &format!("/v1/sailboxes/{sailbox_id}/listeners/{guest_port}"),
                &[],
            )
            .await?;
        raise_api_error(
            status,
            &data,
            &format!("Listener {sailbox_id:?}:{guest_port}"),
        )?;
        serde_json::from_value(data).map_err(|e| SailError::Internal {
            message: format!("failed to parse listener: {e}"),
        })
    }

    /// Ingress-identity headers for this sailbox.
    pub async fn ingress_auth_headers(
        &self,
        sailbox_id: &str,
    ) -> Result<Vec<(String, String)>, SailError> {
        let (status, data) = self
            .get_request(&format!("/v1/sailboxes/{sailbox_id}/ingress-auth"), &[])
            .await?;
        raise_api_error(status, &data, &format!("Sailbox {sailbox_id:?}"))?;
        let headers = data
            .get("headers")
            .and_then(Value::as_object)
            .ok_or_else(|| missing_field("headers"))?;
        Ok(headers
            .iter()
            .map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string()))
            .collect())
    }

    // --- NFS volumes (sailbox-volume API on the same host) ---

    /// Look up an NFS volume by name (`POST /v1/sailbox-volumes/get`),
    /// optionally creating it when missing.
    ///
    /// `name` must be non-empty. When `mint_if_missing` is set the volume is
    /// created if it does not yet exist.
    pub async fn get_volume(
        &self,
        name: &str,
        mint_if_missing: bool,
    ) -> Result<NfsVolume, SailError> {
        if name.trim().is_empty() {
            return Err(SailError::InvalidArgument {
                message: "name is required".to_string(),
            });
        }
        let body = json!({
            "name": name,
            "backend": "nfs",
            "mint_if_missing": mint_if_missing,
            "create_if_missing": mint_if_missing,
        });
        let (status, data) = self.post("/v1/sailbox-volumes/get", &body).await?;
        raise_api_error(status, &data, "")?;
        volume_from(data)
    }

    /// List the org's NFS volumes (`GET /v1/sailbox-volumes`).
    ///
    /// `max_objects`, when given, caps the number returned and must be
    /// non-negative.
    pub async fn list_volumes(
        &self,
        max_objects: Option<i64>,
    ) -> Result<Vec<NfsVolume>, SailError> {
        let mut query: Vec<(String, String)> = Vec::new();
        if let Some(max) = max_objects {
            if max < 0 {
                return Err(SailError::InvalidArgument {
                    message: "max_objects cannot be negative".to_string(),
                });
            }
            query.push(("max_objects".to_string(), max.to_string()));
        }
        let (status, data) = self.get_request("/v1/sailbox-volumes", &query).await?;
        raise_api_error(status, &data, "")?;
        data.get("volumes").and_then(Value::as_array).map_or_else(
            || Ok(Vec::new()),
            |rows| rows.iter().cloned().map(volume_from).collect(),
        )
    }

    /// Delete a volume by id; `Ok(None)` on a 204 (no body), the deleted handle
    /// otherwise.
    pub async fn delete_volume(
        &self,
        volume_id: &str,
        allow_missing: bool,
    ) -> Result<Option<NfsVolume>, SailError> {
        if volume_id.trim().is_empty() {
            return Err(SailError::InvalidArgument {
                message: "volume_id is required".to_string(),
            });
        }
        let path = format!("/v1/sailbox-volumes/{}", volume_id.trim());
        let query: Vec<(String, String)> = if allow_missing {
            vec![("allow_missing".to_string(), "true".to_string())]
        } else {
            Vec::new()
        };
        let policy = if allow_missing {
            DEFAULT_RETRY_POLICY
        } else {
            NO_RETRY
        };
        let (status, data) = self
            .request(Method::Delete, &path, &query, None, policy, None)
            .await?;
        if status == 204 {
            return Ok(None);
        }
        raise_api_error(status, &data, "")?;
        Ok(Some(volume_from(data)?))
    }

    /// Fetch the caller org's SSH certificate-authority public key, creating the
    /// CA on first use. A box trusts this key, so installing it is what lets the
    /// org's signed user certificates authenticate.
    pub async fn org_ssh_ca_public_key(&self) -> Result<String, SailError> {
        let (status, data) = self.get_request("/v1/ssh/ca", &[]).await?;
        raise_api_error(status, &data, "ssh ca")?;
        str_field(&data, "public_key")
    }

    /// Exchange a local SSH public key for a short-lived certificate the org CA
    /// signs (principal `root`). Returns the OpenSSH certificate and its key id
    /// (`org=<id>;fp=...;iat=...`), which identifies the signing org.
    ///
    /// `timeout` bounds a single attempt with no retry (used by the auto-refresh
    /// hook, which runs inside the ssh connect path and must fail open quickly);
    /// `None` uses the default retrying transport.
    pub async fn issue_user_cert(
        &self,
        public_key: &str,
        timeout: Option<f64>,
    ) -> Result<IssuedUserCert, SailError> {
        let body = json!({ "public_key": public_key });
        let bytes = serde_json::to_vec(&body).map_err(|e| SailError::Internal {
            message: format!("failed to serialize request body: {e}"),
        })?;
        let policy = if timeout.is_some() {
            NO_RETRY
        } else {
            DEFAULT_RETRY_POLICY
        };
        let (status, data) = self
            .request(
                Method::Post,
                "/v1/ssh/certificate",
                &[],
                Some(bytes),
                policy,
                timeout,
            )
            .await?;
        raise_api_error(status, &data, "ssh certificate")?;
        Ok(IssuedUserCert {
            certificate: str_field(&data, "certificate")?,
            key_id: data
                .get("key_id")
                .and_then(Value::as_str)
                .unwrap_or_default()
                .to_string(),
        })
    }

    // --- transport helpers ---

    async fn post(&self, path: &str, body: &Value) -> Result<(u16, Value), SailError> {
        let bytes = serde_json::to_vec(body).map_err(|e| SailError::Internal {
            message: format!("failed to serialize request body: {e}"),
        })?;
        self.request(
            Method::Post,
            path,
            &[],
            Some(bytes),
            DEFAULT_RETRY_POLICY,
            None,
        )
        .await
    }

    async fn get_request(
        &self,
        path: &str,
        query: &[(String, String)],
    ) -> Result<(u16, Value), SailError> {
        self.request(Method::Get, path, query, None, DEFAULT_RETRY_POLICY, None)
            .await
    }

    async fn request(
        &self,
        method: Method,
        path: &str,
        query: &[(String, String)],
        body: Option<Vec<u8>>,
        policy: RetryPolicy,
        timeout: Option<f64>,
    ) -> Result<(u16, Value), SailError> {
        let spec = RequestSpec {
            method,
            path: path.to_string(),
            query: query.to_vec(),
            body,
            extra_headers: Vec::new(),
            timeout,
            policy,
            idempotency_key: IdempotencyKey::Auto,
        };
        self.http.request(&spec).await
    }
}

fn handle_from(data: &Value, fallback_name: &str) -> SailboxHandle {
    SailboxHandle {
        sailbox_id: data
            .get("sailbox_id")
            .and_then(Value::as_str)
            .unwrap_or("")
            .to_string(),
        name: data
            .get("name")
            .and_then(Value::as_str)
            .unwrap_or(fallback_name)
            .to_string(),
        status: status_of(data),
        worker_address: data
            .get("worker_address")
            .and_then(Value::as_str)
            .unwrap_or("")
            .to_string(),
        exec_endpoint: data
            .get("exec_proxy_endpoint")
            .and_then(Value::as_str)
            .unwrap_or("")
            .to_string(),
    }
}

fn info_from(data: Value) -> Result<SailboxInfo, SailError> {
    serde_json::from_value::<SailboxInfo>(data).map_err(|e| SailError::Internal {
        message: format!("failed to parse sailbox: {e}"),
    })
}

fn volume_from(data: Value) -> Result<NfsVolume, SailError> {
    serde_json::from_value::<NfsVolume>(data).map_err(|e| SailError::Internal {
        message: format!("failed to parse volume: {e}"),
    })
}

fn int_field(data: &Value, key: &str) -> Result<i64, SailError> {
    data.get(key)
        .and_then(Value::as_i64)
        .ok_or_else(|| missing_field(key))
}

fn str_field(data: &Value, key: &str) -> Result<String, SailError> {
    data.get(key)
        .and_then(Value::as_str)
        .filter(|s| !s.is_empty())
        .map(str::to_string)
        .ok_or_else(|| missing_field(key))
}

fn missing_field(key: &str) -> SailError {
    SailError::Internal {
        message: format!("API response missing field {key:?}"),
    }
}

/// A 404 body the API tags as a real missing resource (vs a route-level 404).
fn is_resource_not_found(data: &Value) -> bool {
    data.get("error")
        .and_then(|e| e.get("type"))
        .and_then(Value::as_str)
        == Some("not_found_error")
}

/// Map a non-2xx lifecycle/monitoring response onto the canonical taxonomy:
/// resource-miss 404 → NotFound, 401/403 → PermissionDenied, 400 →
/// InvalidArgument, any other → Api (a transient/unexpected API failure).
fn raise_api_error(status: u16, data: &Value, context: &str) -> Result<(), SailError> {
    if status < 300 {
        return Ok(());
    }
    let mut message = crate::http::api_error_message(data, "request failed");
    if !context.is_empty() {
        message = format!("{context}: {message}");
    }
    if status == 404 && is_resource_not_found(data) {
        return Err(SailError::NotFound { message });
    }
    if status == 401 || status == 403 {
        return Err(SailError::PermissionDenied { message });
    }
    if status == 400 {
        return Err(SailError::InvalidArgument { message });
    }
    Err(SailError::Api {
        status,
        message,
        body: data.clone(),
    })
}

/// create's ladder: every non-2xx except auth is a Creation failure.
fn raise_for_create_status(status: u16, data: &Value) -> Result<(), SailError> {
    if status < 300 {
        return Ok(());
    }
    let message = crate::http::api_error_message(data, "request failed");
    if status == 401 || status == 403 {
        return Err(SailError::PermissionDenied { message });
    }
    Err(SailError::Creation {
        message,
        status,
        body: data.clone(),
    })
}

/// The echoed lifecycle `status`, parsed into the typed enum.
fn status_of(data: &Value) -> SailboxStatus {
    SailboxStatus::from(data.get("status").and_then(Value::as_str).unwrap_or(""))
}

/// Raise an Api error when the echoed status is not the expected one.
fn require_status(data: &Value, expected: SailboxStatus, status: u16) -> Result<(), SailError> {
    let got = status_of(data);
    if got == expected {
        return Ok(());
    }
    Err(SailError::Api {
        status,
        message: format!("lifecycle call returned unexpected status: {got}"),
        body: data.clone(),
    })
}

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

    #[test]
    fn info_reads_resource_fields_and_defaults_optionals() {
        let info = info_from(json!({
            "sailbox_id": "sb-1", "app_id": "app-1", "app_name": "a", "name": "n",
            "image_id": "img-1",
            "status": "running", "memory_mib": 2048, "vcpu_count": 4,
            "state_disk_size_gib": 10,
            "cpu_requested_vcpu": 2, "cpu_used_vcpu": 1.5,
            "memory_requested_bytes": 1024, "memory_used_bytes": 512,
            "disk_requested_bytes": 4096, "disk_used_bytes": 2048,
            "architecture": "amd64", "checkpoint_generation": 7,
            "created_at": "2026-01-01T00:00:00Z", "updated_at": "2026-01-02T00:00:00Z"
        }))
        .unwrap();
        // Resource fields are read straight from the wire (the backend always sends them).
        assert_eq!(info.cpu_requested_vcpu, 2);
        assert_eq!(info.memory_used_bytes, 512);
        assert_eq!(info.checkpoint_generation, 7);
        // Genuinely-optional fields default to None when absent.
        assert_eq!(info.guest_schema_version, None);
        assert_eq!(info.error_message, None);
        assert_eq!(info.started_at, None);
    }

    #[test]
    fn api_error_ladder_maps_statuses() {
        let not_found = json!({"error": {"type": "not_found_error", "message": "gone"}});
        assert!(matches!(
            raise_api_error(404, &not_found, ""),
            Err(SailError::NotFound { .. })
        ));
        // A route-level 404 (no not_found_error type) is transient, not a miss.
        let route_404 = json!({"error": {"message": "no route"}});
        assert!(matches!(
            raise_api_error(404, &route_404, ""),
            Err(SailError::Api { .. })
        ));
        let auth = json!({"error": {"message": "nope"}});
        assert!(matches!(
            raise_api_error(403, &auth, ""),
            Err(SailError::PermissionDenied { .. })
        ));
        assert!(matches!(
            raise_api_error(400, &auth, ""),
            Err(SailError::InvalidArgument { .. })
        ));
        assert!(matches!(
            raise_api_error(503, &auth, ""),
            Err(SailError::Api { status: 503, .. })
        ));
        assert!(raise_api_error(200, &json!({}), "").is_ok());
    }

    #[test]
    fn create_ladder_maps_non_auth_to_creation() {
        let body = json!({"error": {"message": "no capacity"}});
        assert!(matches!(
            raise_for_create_status(503, &body),
            Err(SailError::Creation { .. })
        ));
        assert!(matches!(
            raise_for_create_status(401, &body),
            Err(SailError::PermissionDenied { .. })
        ));
    }

    #[test]
    fn resource_limits_accept_defaults_and_in_range() {
        // None and 0 mean "platform default"; in-range values pass.
        validate_resource_limits(None, None, None).unwrap();
        validate_resource_limits(Some(0), Some(0), Some(0)).unwrap();
        validate_resource_limits(Some(4), Some(1024), Some(32)).unwrap();
        validate_resource_limits(Some(2), Some(8192), Some(10)).unwrap();
    }

    #[test]
    fn resource_limits_reject_out_of_range() {
        let msg = |r: Result<(), SailError>| match r {
            Err(SailError::InvalidArgument { message }) => message,
            other => panic!("expected InvalidArgument, got {other:?}"),
        };
        assert!(msg(validate_resource_limits(Some(-1), None, None))
            .contains("max_cpu must be at least 0"));
        assert!(msg(validate_resource_limits(Some(5), None, None))
            .contains("max_cpu must be at most 4"));
        assert!(msg(validate_resource_limits(None, Some(512), None))
            .contains("max_memory_mib must be at least 1024"));
        assert!(msg(validate_resource_limits(None, Some(8193), None))
            .contains("max_memory_mib must be at most 8192"));
        assert!(msg(validate_resource_limits(None, None, Some(-1)))
            .contains("max_disk_gib must be at least 0"));
        assert!(msg(validate_resource_limits(None, None, Some(33)))
            .contains("max_disk_gib must be at most 32"));
    }

    // The Rust half of the cross-language image contract: ImageSpec's serde
    // attributes must produce canonical proto-JSON (camelCase fields, enum value
    // names, flattened oneofs). The Go side proves protojson decodes this exact
    // shape (TestCreateSailboxHandlerAcceptsCanonicalProtoJSON). Keep the two in sync.
    #[test]
    fn image_spec_serializes_to_canonical_proto_json() {
        use crate::image::{
            BaseImage, ImageArchitecture, ImageBuildStep, ImageSpec, PackageInstall, RunCommand,
        };
        let spec = ImageSpec {
            base: Some(BaseImage::Debian),
            architecture: ImageArchitecture::Arm64,
            python_version: "3.12".to_string(),
            build_steps: vec![
                ImageBuildStep::AptInstall(PackageInstall {
                    packages: vec!["git".to_string(), "curl".to_string()],
                }),
                ImageBuildStep::RunCommand(RunCommand {
                    command: "echo hi".to_string(),
                }),
            ],
            ..Default::default()
        };
        let json = serde_json::to_value(&spec).unwrap();
        assert_eq!(json["base"], json!("BASE_IMAGE_DEBIAN"));
        assert_eq!(json["architecture"], json!("IMAGE_ARCHITECTURE_ARM64"));
        assert_eq!(json["pythonVersion"], json!("3.12"));
        assert_eq!(
            json["buildSteps"][0]["aptInstall"]["packages"][0],
            json!("git")
        );
        assert_eq!(
            json["buildSteps"][1]["runCommand"]["command"],
            json!("echo hi")
        );
    }
}