fundaia 0.26.0

use std::time::Duration;

use anyhow::{anyhow, bail, Context, Result};
use futures_util::StreamExt;
use percent_encoding::{utf8_percent_encode, NON_ALPHANUMERIC};
use reqwest::{Method, StatusCode};
use serde::de::DeserializeOwned;
use serde::Serialize;

use super::models::*;
use crate::config::{host_of, normalise_url};

/// Every call this tool makes, in one place.
///
/// Thin on purpose: it owns the base address, the bearer header and the shape
/// of a failure, and knows nothing about what any of it means. The commands
/// hold the meaning, the renderer holds the appearance, and this holds the
/// wire — which is what lets a command be read without a browser tab open on
/// the API.
/// Cloning one is cheap: `reqwest::Client` is a handle onto a shared pool, so a
/// clone shares the connections rather than opening a second set.
#[derive(Clone)]
pub struct ApiClient {
    base: String,
    http: reqwest::Client,
    /// The one for event streams. See `new` for why it cannot be the same one.
    streaming: reqwest::Client,
    token: Option<String>,
}

/// A server error, as much of it as is worth showing.
#[derive(Debug)]
pub struct ApiError {
    pub status: StatusCode,
    pub code: String,
    pub message: String,
}

impl std::fmt::Display for ApiError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(formatter, "{} ({})", self.message, self.status.as_u16())
    }
}

impl std::error::Error for ApiError {}

#[derive(serde::Deserialize)]
struct ErrorBody {
    #[serde(default)]
    error: Option<String>,
    #[serde(default)]
    message: Option<String>,
}

impl ApiClient {
    pub fn new(base: &str, token: Option<String>) -> Result<Self> {
        require_transport_security(base)?;

        let http = reqwest::Client::builder()
            // Long enough for a cold start behind the gateway, short enough
            // that a wedged connection does not hang a terminal for ever.
            .timeout(Duration::from_secs(60))
            // Separate from the whole-request budget: a host that is down
            // should fail in seconds, not spend the sixty a slow answer gets.
            .connect_timeout(Duration::from_secs(10))
            .user_agent(concat!("fundaia/", env!("CARGO_PKG_VERSION")))
            .build()
            .context("could not build the HTTP client")?;

        /*
         * A second client, because `timeout` above is the whole request.
         *
         * It covers reading the body too, which is right for a JSON answer and
         * wrong for a stream: a build that prints for more than a minute had
         * its log cut off mid-line, and `watch` — which is silent by design
         * until something happens — died after exactly sixty seconds every
         * time. What a stream needs is a limit on *silence*, and the server
         * sends a keep-alive comment every twenty seconds, so a read that
         * blocks far longer than that is a connection nobody is on the other
         * end of.
         */
        let streaming = reqwest::Client::builder()
            .read_timeout(Duration::from_secs(90))
            .connect_timeout(Duration::from_secs(10))
            .user_agent(concat!("fundaia/", env!("CARGO_PKG_VERSION")))
            .build()
            .context("could not build the streaming HTTP client")?;

        Ok(Self {
            base: normalise_url(base),
            http,
            streaming,
            token,
        })
    }

    pub fn base(&self) -> &str {
        &self.base
    }

    pub async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T> {
        self.send(Method::GET, path, None::<&()>).await
    }

    pub async fn post<T: DeserializeOwned, B: Serialize>(&self, path: &str, body: &B) -> Result<T> {
        self.send(Method::POST, path, Some(body)).await
    }

    pub async fn put<T: DeserializeOwned, B: Serialize>(&self, path: &str, body: &B) -> Result<T> {
        self.send(Method::PUT, path, Some(body)).await
    }

    /// PATCH answers with a small body every caller here ignores.
    pub async fn patch<B: Serialize>(&self, path: &str, body: &B) -> Result<()> {
        let response = self.request(Method::PATCH, path, Some(body)).send().await?;
        Self::check(response).await?;
        Ok(())
    }

    /// The PATCHes whose answer is worth reading — a resize hands back the row
    /// it wrote, which is what `--json` prints and what the table re-reads.
    pub async fn patch_for<T: DeserializeOwned, B: Serialize>(
        &self,
        path: &str,
        body: &B,
    ) -> Result<T> {
        self.send(Method::PATCH, path, Some(body)).await
    }

    /// For the DELETEs that answer with something worth reading — how many
    /// references a disconnection removed, and nothing else so far.
    pub async fn delete_for<T: DeserializeOwned>(&self, path: &str) -> Result<T> {
        self.send(Method::DELETE, path, None::<&()>).await
    }

    /// For the endpoints that answer 204, where decoding a body would fail.
    pub async fn delete(&self, path: &str) -> Result<()> {
        let response = self
            .request(Method::DELETE, path, None::<&()>)
            .send()
            .await?;
        Self::check(response).await?;
        Ok(())
    }

    async fn send<T: DeserializeOwned, B: Serialize>(
        &self,
        method: Method,
        path: &str,
        body: Option<&B>,
    ) -> Result<T> {
        let response = self.request(method, path, body).send().await?;
        let text = Self::check(response).await?;

        // An empty body decoded as `T` fails with a message about the input
        // being empty, which sends the reader looking in the wrong place.
        if text.trim().is_empty() {
            bail!("the server answered {path} with nothing at all");
        }

        serde_json::from_str(&text)
            .with_context(|| format!("could not read the answer from {path}"))
    }

    fn request<B: Serialize>(
        &self,
        method: Method,
        path: &str,
        body: Option<&B>,
    ) -> reqwest::RequestBuilder {
        self.request_with(&self.http, method, path, body)
    }

    /// The same request, on the client that tolerates a long quiet body.
    fn streaming_request(&self, path: &str) -> reqwest::RequestBuilder {
        self.request_with(&self.streaming, Method::GET, path, None::<&()>)
            .header("accept", "text/event-stream")
    }

    fn request_with<B: Serialize>(
        &self,
        client: &reqwest::Client,
        method: Method,
        path: &str,
        body: Option<&B>,
    ) -> reqwest::RequestBuilder {
        let mut builder = client.request(method, format!("{}{path}", self.base));

        if let Some(token) = &self.token {
            builder = builder.bearer_auth(token);
        }
        if let Some(body) = body {
            builder = builder.json(body);
        }

        builder
    }

    /// Turns a non-2xx into an `ApiError` carrying the server's own words.
    async fn check(response: reqwest::Response) -> Result<String> {
        let status = response.status();
        let text = response.text().await.unwrap_or_default();

        if status.is_success() {
            return Ok(text);
        }

        let body: ErrorBody = serde_json::from_str(&text).unwrap_or(ErrorBody {
            error: None,
            message: None,
        });

        let message = body.message.unwrap_or_else(|| match status {
            StatusCode::UNAUTHORIZED => "not signed in — run `fundaia login`".to_owned(),
            StatusCode::NOT_FOUND => "there is nothing there".to_owned(),
            _ => status
                .canonical_reason()
                .unwrap_or("the request failed")
                .to_owned(),
        });

        Err(ApiError {
            status,
            code: body.error.unwrap_or_else(|| "unknown_error".to_owned()),
            message,
        }
        .into())
    }

    // ---------------------------------------------------------------- reads

    pub async fn projects(&self) -> Result<Vec<Project>> {
        Ok(self
            .get::<ProjectsPage>("/api/infra/projects")
            .await?
            .projects)
    }

    /// Finds a project by id, slug or name, in that order of specificity.
    ///
    /// A CLI is typed at, so it has to accept the thing a person can see on a
    /// screen. Ambiguity is an error rather than a guess: two projects whose
    /// names differ only in case is unlikely, and picking one silently is the
    /// kind of helpfulness that deploys to the wrong place.
    pub async fn find_project(&self, reference: &str) -> Result<Project> {
        let projects = self.projects().await?;
        let needle = reference.to_lowercase();

        let matched: Vec<Project> = projects
            .into_iter()
            .filter(|project| {
                project.id == reference
                    || project.slug.to_lowercase() == needle
                    || project.name.to_lowercase() == needle
            })
            .collect();

        match matched.len() {
            0 => Err(anyhow!("there is no project called `{reference}`")),
            1 => Ok(matched.into_iter().next().expect("length checked")),
            other => Err(anyhow!(
                "`{reference}` matches {other} projects — use the id"
            )),
        }
    }

    pub async fn services(&self, project_id: &str) -> Result<ServicesPage> {
        self.get(&format!("/api/infra/projects/{project_id}/services"))
            .await
    }

    pub async fn canvas(&self, project_id: &str) -> Result<Canvas> {
        self.get(&format!("/api/infra/projects/{project_id}/canvas"))
            .await
    }

    /// Several nodes in one write, and the canvas afterwards.
    ///
    /// A PATCH per service would be the obvious shape and is the wrong one: the
    /// server writes a whole request inside one transaction, so a half-applied
    /// arrangement cannot exist — and the answer saves reading the canvas back.
    pub async fn place_on_canvas(
        &self,
        project_id: &str,
        positions: &[CanvasPlacement],
    ) -> Result<Canvas> {
        self.patch_for(
            &format!("/api/infra/projects/{project_id}/canvas"),
            &serde_json::json!({ "positions": positions }),
        )
        .await
    }

    pub async fn arrange_canvas(&self, project_id: &str, layout: &str) -> Result<Canvas> {
        self.post(
            &format!("/api/infra/projects/{project_id}/canvas/arrange"),
            &serde_json::json!({ "layout": layout }),
        )
        .await
    }

    /// The same lookup for a service, within one project.
    pub async fn find_service(&self, project_id: &str, reference: &str) -> Result<Service> {
        let page = self.services(project_id).await?;
        let needle = reference.to_lowercase();

        let matched: Vec<Service> = page
            .services
            .into_iter()
            .filter(|service| {
                service.id == reference
                    || service.slug.to_lowercase() == needle
                    || service.name.to_lowercase() == needle
            })
            .collect();

        match matched.len() {
            0 => Err(anyhow!("this project has no service called `{reference}`")),
            1 => Ok(matched.into_iter().next().expect("length checked")),
            other => Err(anyhow!(
                "`{reference}` matches {other} services — use the id"
            )),
        }
    }

    pub async fn service_detail(&self, service_environment_id: &str) -> Result<ServiceDetail> {
        self.get(&format!("/api/infra/services/{service_environment_id}"))
            .await
    }

    pub async fn deployments(&self, service_environment_id: &str) -> Result<Vec<Deployment>> {
        Ok(self
            .get::<DeploymentsPage>(&format!(
                "/api/infra/services/{service_environment_id}/deployments"
            ))
            .await?
            .deployments)
    }

    /// The diff of one commit, from the repository the service is built from.
    pub async fn commit_change(
        &self,
        service_environment_id: &str,
        sha: &str,
    ) -> Result<CommitChangeResponse> {
        self.get(&format!(
            "/api/infra/services/{service_environment_id}/commits/{sha}"
        ))
        .await
    }

    pub async fn deployment(&self, deployment_id: &str) -> Result<DeploymentDetail> {
        self.get(&format!("/api/infra/deployments/{deployment_id}"))
            .await
    }

    pub async fn variables(&self, service_environment_id: &str) -> Result<Vec<Variable>> {
        Ok(self
            .get::<VariablesPage>(&format!(
                "/api/infra/services/{service_environment_id}/variables"
            ))
            .await?
            .variables)
    }

    pub async fn incidents(
        &self,
        service_environment_id: &str,
        decided: bool,
    ) -> Result<Vec<Incident>> {
        Ok(self
            .get::<IncidentsPage>(&format!(
                "/api/infra/services/{service_environment_id}/incidents?decided={decided}"
            ))
            .await?
            .incidents)
    }

    /// Records a decision and answers with the incident as it now stands.
    ///
    /// `send` rather than `patch`, which throws the body away: this endpoint
    /// answers with the updated row and `--json` is supposed to print it.
    pub async fn decide_incident(&self, incident_id: &str, status: &str) -> Result<Incident> {
        self.send(
            Method::PATCH,
            &format!("/api/infra/incidents/{incident_id}"),
            Some(&serde_json::json!({ "status": status })),
        )
        .await
    }

    /// What is waiting in a whole project, which is the question worth asking:
    /// a held commit is somebody else's push, so nobody knows which service.
    pub async fn project_approvals(&self, project_id: &str) -> Result<Vec<Approval>> {
        Ok(self
            .get::<ApprovalsPage>(&format!("/api/infra/projects/{project_id}/approvals"))
            .await?
            .approvals)
    }

    /// One service's requests, answered ones included.
    pub async fn service_approvals(&self, service_environment_id: &str) -> Result<Vec<Approval>> {
        Ok(self
            .get::<ApprovalsPage>(&format!(
                "/api/infra/services/{service_environment_id}/approvals"
            ))
            .await?
            .approvals)
    }

    /// Whether this service's repository can be read, and who is being asked.
    pub async fn repository_access(
        &self,
        service_environment_id: &str,
    ) -> Result<RepositoryAccessPage> {
        self.get::<RepositoryAccessPage>(&format!(
            "/api/infra/services/{service_environment_id}/repository-access"
        ))
        .await
    }

    /// Emails the owner an approval link. `None` reuses an address already known.
    pub async fn request_repository_access(
        &self,
        service_environment_id: &str,
        email: Option<&str>,
    ) -> Result<SentRepositoryAccess> {
        let body = match email {
            Some(address) => serde_json::json!({ "email": address }),
            None => serde_json::json!({}),
        };

        self.post::<SentRepositoryAccess, _>(
            &format!("/api/infra/services/{service_environment_id}/repository-access"),
            &body,
        )
        .await
    }

    pub async fn approve_deployment(&self, approval_id: &str) -> Result<Deployment> {
        Ok(self
            .post::<DeploymentAccepted, _>(
                &format!("/api/infra/approvals/{approval_id}/approve"),
                &serde_json::json!({}),
            )
            .await?
            .deployment)
    }

    pub async fn reject_deployment(&self, approval_id: &str) -> Result<serde_json::Value> {
        self.post(
            &format!("/api/infra/approvals/{approval_id}/reject"),
            &serde_json::json!({}),
        )
        .await
    }

    pub async fn autofix_runs(&self, service_environment_id: &str) -> Result<Vec<AutofixRun>> {
        Ok(self
            .get::<AutofixPage>(&format!(
                "/api/infra/services/{service_environment_id}/autofix"
            ))
            .await?
            .runs)
    }

    /// Writes whatever the caller named and answers with the settings as they
    /// now stand — so `--json` prints the result rather than the request.
    pub async fn set_autofix(
        &self,
        service_environment_id: &str,
        changes: &serde_json::Value,
    ) -> Result<serde_json::Value> {
        self.put(
            &format!("/api/infra/services/{service_environment_id}/autofix"),
            changes,
        )
        .await
    }

    /// Starts a run and answers with it. The work happens after this returns —
    /// the platform does not hold the request open for four minutes.
    pub async fn fix_incident(&self, incident_id: &str) -> Result<AutofixRun> {
        self.post(
            &format!("/api/infra/incidents/{incident_id}/fix"),
            &serde_json::json!({}),
        )
        .await
    }

    pub async fn domains(&self, service_environment_id: &str) -> Result<Vec<Domain>> {
        Ok(self
            .get::<DomainsPage>(&format!(
                "/api/infra/services/{service_environment_id}/domains"
            ))
            .await?
            .domains)
    }

    pub async fn usage(
        &self,
        service_environment_id: &str,
        minutes: u32,
    ) -> Result<Vec<UsageSample>> {
        Ok(self
            .get::<UsagePage>(&format!(
                "/api/infra/services/{service_environment_id}/usage?minutes={minutes}"
            ))
            .await?
            .samples)
    }

    /// What an agent component has spent, in tokens and in dollars.
    pub async fn consumption(
        &self,
        service_environment_id: &str,
        range: &str,
    ) -> Result<ConsumptionPage> {
        self.get::<ConsumptionPage>(&format!(
            "/api/infra/services/{service_environment_id}/consumption?range={range}"
        ))
        .await
    }

    /// Every request an agent received over a window, and what it is doing now.
    ///
    /// The outcome filter narrows the list only — the counts always describe the
    /// whole window, so the numbers do not move as somebody flips between them.
    pub async fn agent_activity(
        &self,
        service_environment_id: &str,
        range: &str,
        outcome: Option<&str>,
    ) -> Result<AgentActivityPage> {
        let filter = outcome.map_or_else(String::new, |one| format!("&outcome={one}"));

        self.get::<AgentActivityPage>(&format!(
            "/api/infra/services/{service_environment_id}/activity?range={range}{filter}"
        ))
        .await
    }

    /// The catalogue, categories and all.
    ///
    /// The whole page rather than just the entries: the grouping is decided on
    /// the server precisely so this and the canvas present it the same way, and
    /// throwing it away here would mean inventing one.
    pub async fn templates(&self) -> Result<TemplatesPage> {
        self.get::<TemplatesPage>("/api/infra/templates").await
    }

    pub async fn agent_setup(&self, service_environment_id: &str) -> Result<AgentSetup> {
        self.get::<AgentSetup>(&format!(
            "/api/infra/services/{service_environment_id}/agent"
        ))
        .await
    }

    /// Changes what a Claude Code component runs on. Only what the patch names.
    pub async fn configure_agent(
        &self,
        service_environment_id: &str,
        patch: &AgentSetupPatch,
    ) -> Result<AgentSetup> {
        self.put::<AgentSetup, _>(
            &format!("/api/infra/services/{service_environment_id}/agent"),
            patch,
        )
        .await
    }

    pub async fn service_config(&self, service_environment_id: &str) -> Result<ServiceConfigPage> {
        self.get::<ServiceConfigPage>(&format!(
            "/api/infra/services/{service_environment_id}/config"
        ))
        .await
    }

    pub async fn edit_service_config(
        &self,
        service_environment_id: &str,
        content: &str,
    ) -> Result<()> {
        self.put::<ServiceConfigResponse, _>(
            &format!("/api/infra/services/{service_environment_id}/config"),
            &serde_json::json!({ "content": content }),
        )
        .await?;
        Ok(())
    }

    pub async fn candidates(
        &self,
        service_environment_id: &str,
    ) -> Result<Vec<ConnectionCandidate>> {
        Ok(self
            .get::<ConnectionCandidates>(&format!(
                "/api/infra/services/{service_environment_id}/connections"
            ))
            .await?
            .candidates)
    }

    pub async fn whoami(&self) -> Result<Option<PrincipalIdentity>> {
        Ok(self.get::<Principal>("/api/auth/session").await?.principal)
    }

    pub async fn version(&self) -> Result<ServerVersion> {
        self.get("/api/version").await
    }

    // --------------------------------------------------------------- writes

    pub async fn deploy(&self, service_environment_id: &str) -> Result<Deployment> {
        Ok(self
            .post::<DeploymentAccepted, _>(
                &format!("/api/infra/services/{service_environment_id}/deploy"),
                &serde_json::json!({}),
            )
            .await?
            .deployment)
    }

    /// Sends an archive and, unless told otherwise, deploys it.
    ///
    /// The bytes are the whole body rather than a multipart form: there is one
    /// part, and a boundary would mean the server reading a hundred megabytes
    /// to find where it ends. The **streaming client** is the one used, because
    /// the default carries the timeout an ordinary API call deserves and an
    /// upload over a domestic connection is minutes.
    pub async fn upload(
        &self,
        service_environment_id: &str,
        archive: Vec<u8>,
        file_name: &str,
        deploy: bool,
    ) -> Result<UploadAccepted> {
        let path = format!(
            "/api/infra/services/{service_environment_id}/uploads{}",
            if deploy { "" } else { "?deploy=false" }
        );

        let response = self
            .request_with(&self.streaming, Method::POST, &path, None::<&()>)
            .header("content-type", "application/gzip")
            .header("x-fundaia-upload-name", file_name)
            .body(archive)
            .send()
            .await?;

        let text = Self::check(response).await?;
        serde_json::from_str(&text)
            .with_context(|| format!("could not read the answer from {path}"))
    }

    pub async fn uploads(&self, service_environment_id: &str) -> Result<UploadsPage> {
        self.get(&format!(
            "/api/infra/services/{service_environment_id}/uploads"
        ))
        .await
    }

    pub async fn restart(&self, service_environment_id: &str) -> Result<Deployment> {
        Ok(self
            .post::<DeploymentAccepted, _>(
                &format!("/api/infra/services/{service_environment_id}/restart"),
                &serde_json::json!({}),
            )
            .await?
            .deployment)
    }

    pub async fn stop(&self, service_environment_id: &str) -> Result<()> {
        let response = self
            .request(
                Method::POST,
                &format!("/api/infra/services/{service_environment_id}/stop"),
                Some(&serde_json::json!({})),
            )
            .send()
            .await?;
        Self::check(response).await?;
        Ok(())
    }

    pub async fn redeploy(&self, deployment_id: &str) -> Result<Deployment> {
        Ok(self
            .post::<DeploymentAccepted, _>(
                &format!("/api/infra/deployments/{deployment_id}/redeploy"),
                &serde_json::json!({}),
            )
            .await?
            .deployment)
    }

    pub async fn rollback(&self, deployment_id: &str) -> Result<Deployment> {
        Ok(self
            .post::<DeploymentAccepted, _>(
                &format!("/api/infra/deployments/{deployment_id}/rollback"),
                &serde_json::json!({}),
            )
            .await?
            .deployment)
    }

    /// Stops a deployment that has not finished.
    ///
    /// Answers with the deployment *and its pipeline*, the way the detail route
    /// does: there is nothing left to watch, so the whole answer arrives at once.
    pub async fn cancel(&self, deployment_id: &str) -> Result<DeploymentDetail> {
        self.post(
            &format!("/api/infra/deployments/{deployment_id}/cancel"),
            &serde_json::json!({}),
        )
        .await
    }

    pub async fn set_variable(
        &self,
        service_environment_id: &str,
        key: &str,
        value: &str,
        secret: bool,
        sealed: bool,
    ) -> Result<()> {
        let _: serde_json::Value = self
            .put(
                &format!("/api/infra/services/{service_environment_id}/variables"),
                &serde_json::json!({
                    "key": key,
                    "value": value,
                    "secret": secret,
                    "sealed": sealed,
                }),
            )
            .await?;
        Ok(())
    }

    pub async fn backups(&self, service_environment_id: &str) -> Result<BackupsPage> {
        self.get(&format!(
            "/api/infra/services/{service_environment_id}/backups"
        ))
        .await
    }

    /// Takes a copy and waits for it: the server does not answer until it is on disk.
    pub async fn take_backup(&self, service_environment_id: &str) -> Result<Backup> {
        Ok(self
            .post::<BackupTaken, _>(
                &format!("/api/infra/services/{service_environment_id}/backups"),
                &serde_json::json!({}),
            )
            .await?
            .backup)
    }

    pub async fn schedule_backups(
        &self,
        service_environment_id: &str,
        schedules: &[String],
    ) -> Result<Vec<String>> {
        Ok(self
            .put::<BackupSchedules, _>(
                &format!("/api/infra/services/{service_environment_id}/backups/schedule"),
                &serde_json::json!({ "schedules": schedules }),
            )
            .await?
            .schedules)
    }

    /// Overwrites the volume. The server takes a safety copy first.
    pub async fn restore_backup(&self, backup_id: &str) -> Result<Backup> {
        Ok(self
            .post::<BackupTaken, _>(
                &format!("/api/infra/backups/{backup_id}/restore"),
                &serde_json::json!({}),
            )
            .await?
            .backup)
    }

    pub async fn lock_backup(&self, backup_id: &str, locked: bool) -> Result<Backup> {
        Ok(self
            .put::<BackupTaken, _>(
                &format!("/api/infra/backups/{backup_id}/lock"),
                &serde_json::json!({ "locked": locked }),
            )
            .await?
            .backup)
    }

    pub async fn delete_backup(&self, backup_id: &str) -> Result<()> {
        self.delete(&format!("/api/infra/backups/{backup_id}"))
            .await
    }

    /// Seals a variable. One way: the server refuses every read and write after.
    pub async fn seal_variable(&self, variable_id: &str) -> Result<()> {
        let _: serde_json::Value = self
            .post(
                &format!("/api/infra/variables/{variable_id}/seal"),
                &serde_json::json!({}),
            )
            .await?;
        Ok(())
    }

    pub async fn remove_variable(&self, variable_id: &str) -> Result<()> {
        self.delete(&format!("/api/infra/variables/{variable_id}"))
            .await
    }

    pub async fn reveal_variable(&self, variable_id: &str) -> Result<String> {
        Ok(self
            .post::<RevealedVariable, _>(
                &format!("/api/infra/variables/{variable_id}/reveal"),
                &serde_json::json!({}),
            )
            .await?
            .value)
    }

    pub async fn connect(&self, consumer_id: &str, provider_id: &str) -> Result<Connection> {
        Ok(self
            .post::<ConnectionMade, _>(
                &format!("/api/infra/services/{consumer_id}/connections"),
                &serde_json::json!({ "providerId": provider_id }),
            )
            .await?
            .connection)
    }

    pub async fn add_domain(
        &self,
        service_environment_id: &str,
        host: Option<&str>,
    ) -> Result<Domain> {
        #[derive(serde::Deserialize)]
        struct Added {
            domain: Domain,
        }

        let body = match host {
            Some(host) => serde_json::json!({ "host": host }),
            None => serde_json::json!({}),
        };

        Ok(self
            .post::<Added, _>(
                &format!("/api/infra/services/{service_environment_id}/domains"),
                &body,
            )
            .await?
            .domain)
    }

    pub async fn remove_domain(&self, domain_id: &str) -> Result<()> {
        self.delete(&format!("/api/infra/domains/{domain_id}"))
            .await
    }

    /// Asks about one custom domain now, rather than on the platform's own
    /// schedule.
    ///
    /// A POST despite reading nothing but Cloudflare: it stores the answer, and
    /// what somebody wants from `fundaia domains verify` is the state after the
    /// check rather than the state before it.
    pub async fn verify_domain(&self, domain_id: &str) -> Result<Domain> {
        #[derive(serde::Deserialize)]
        struct Verified {
            domain: Domain,
        }

        Ok(self
            .post::<Verified, _>(
                &format!("/api/infra/domains/{domain_id}/verify"),
                &serde_json::json!({}),
            )
            .await?
            .domain)
    }

    pub async fn create_service(&self, project_id: &str, draft: &NewService) -> Result<String> {
        #[derive(serde::Deserialize)]
        #[serde(rename_all = "camelCase")]
        struct Created {
            service_id: String,
        }

        Ok(self
            .post::<Created, _>(&format!("/api/infra/projects/{project_id}/services"), draft)
            .await?
            .service_id)
    }

    // ------------------------------------------------- settings & capacity

    pub async fn capacity(&self) -> Result<Capacity> {
        self.get("/api/infra/capacity").await
    }

    /// Changes only the fields that were given; the rest are left alone.
    pub async fn update_settings(
        &self,
        service_environment_id: &str,
        patch: &SettingsPatch,
    ) -> Result<()> {
        self.patch(
            &format!("/api/infra/services/{service_environment_id}"),
            patch,
        )
        .await
    }

    /// Turns bucket encryption on or off, and reports what it moved.
    pub async fn set_encryption(
        &self,
        service_environment_id: &str,
        enabled: bool,
    ) -> Result<EncryptionOutcome> {
        self.put(
            &format!("/api/infra/services/{service_environment_id}/encryption"),
            &serde_json::json!({ "enabled": enabled }),
        )
        .await
    }

    /// Publishes a service on the internet, or takes it off.
    ///
    /// Off removes its hostnames; the project network is untouched, because
    /// that part was never optional.
    pub async fn set_public_access(
        &self,
        service_environment_id: &str,
        public_access: bool,
    ) -> Result<ExposureOutcome> {
        self.put(
            &format!("/api/infra/services/{service_environment_id}/public-access"),
            &serde_json::json!({ "publicAccess": public_access }),
        )
        .await
    }

    pub async fn delete_service(&self, service_id: &str) -> Result<()> {
        self.delete(&format!("/api/infra/services/{service_id}"))
            .await
    }

    // -------------------------------------------------------------- projects

    /// With no workspace named, the server puts it in the caller's own.
    pub async fn create_project(
        &self,
        name: &str,
        description: Option<&str>,
        workspace_id: Option<&str>,
    ) -> Result<Project> {
        let mut body = serde_json::Map::new();
        body.insert("name".into(), name.into());
        if let Some(description) = description {
            body.insert("description".into(), description.into());
        }
        if let Some(workspace_id) = workspace_id {
            body.insert("workspaceId".into(), workspace_id.into());
        }

        Ok(self
            .post::<ProjectCreated, _>("/api/infra/projects", &serde_json::Value::Object(body))
            .await?
            .project)
    }

    /// An absent field keeps what the project already has, which is why the
    /// description is optional rather than emptied by omission.
    pub async fn rename_project(
        &self,
        project_id: &str,
        name: Option<&str>,
        description: Option<&str>,
    ) -> Result<()> {
        let mut body = serde_json::Map::new();
        if let Some(name) = name {
            body.insert("name".into(), name.into());
        }
        if let Some(description) = description {
            body.insert("description".into(), description.into());
        }

        self.patch(
            &format!("/api/infra/projects/{project_id}"),
            &serde_json::Value::Object(body),
        )
        .await
    }

    /// Renames a service, which rewrites its wiring rather than a label.
    ///
    /// On the service and not the service-environment: the name belongs to the
    /// service, and the server rewrites the `${{ Name.KEY }}` expressions and
    /// the internal addresses in the same transaction.
    pub async fn rename_service(&self, service_id: &str, name: &str) -> Result<()> {
        self.patch(
            &format!("/api/infra/services/{service_id}/name"),
            &serde_json::json!({ "name": name }),
        )
        .await
    }

    /// The confirmation travels: the route refuses without it.
    ///
    /// Not merely the dialog's belt and braces — this endpoint used to archive,
    /// so a `fundaia` from before the change sends the same `DELETE` it always
    /// sent. Without the parameter the server turns it away instead of
    /// destroying a project on behalf of a binary that thinks it is archiving.
    pub async fn delete_project(&self, project_id: &str, confirmation: &str) -> Result<()> {
        self.delete(&format!(
            "/api/infra/projects/{project_id}?confirm={}",
            utf8_percent_encode(confirmation, NON_ALPHANUMERIC)
        ))
        .await
    }

    pub async fn move_project(&self, project_id: &str, workspace_id: &str) -> Result<()> {
        self.patch(
            &format!("/api/infra/projects/{project_id}/workspace"),
            &serde_json::json!({ "workspaceId": workspace_id }),
        )
        .await
    }

    /// Undoes a connection, which means deleting the variable references that
    /// were the whole of it. Their keys are what to report.
    pub async fn disconnect(&self, consumer_id: &str, provider_id: &str) -> Result<Vec<String>> {
        Ok(self
            .delete_for::<Disconnected>(&format!(
                "/api/infra/services/{consumer_id}/connections/{provider_id}"
            ))
            .await?
            .removed)
    }

    // --------------------------------------------------------------- volumes

    pub async fn volumes(&self, service_environment_id: &str) -> Result<Vec<Volume>> {
        Ok(self
            .get::<VolumesPage>(&format!(
                "/api/infra/services/{service_environment_id}/volumes"
            ))
            .await?
            .volumes)
    }

    pub async fn attach_volume(
        &self,
        service_environment_id: &str,
        mount_path: &str,
    ) -> Result<Volume> {
        Ok(self
            .post::<VolumeAttached, _>(
                &format!("/api/infra/services/{service_environment_id}/volumes"),
                &serde_json::json!({ "mountPath": mount_path }),
            )
            .await?
            .volume)
    }

    pub async fn resize_volume(&self, volume_id: &str, size_limit_mb: u64) -> Result<Volume> {
        Ok(self
            .patch_for::<VolumeAttached, _>(
                &format!("/api/infra/volumes/{volume_id}"),
                &serde_json::json!({ "sizeLimitMb": size_limit_mb }),
            )
            .await?
            .volume)
    }

    pub async fn detach_volume(&self, volume_id: &str) -> Result<()> {
        self.delete(&format!("/api/infra/volumes/{volume_id}"))
            .await
    }

    // ---------------------------------------------------------------- builds

    pub async fn recipe(&self, service_environment_id: &str) -> Result<Recipe> {
        self.get(&format!(
            "/api/infra/services/{service_environment_id}/recipe"
        ))
        .await
    }

    /// Once edited, never regenerated: what shipped is always what was on
    /// screen. The server says so by answering `edited`.
    pub async fn edit_recipe(
        &self,
        service_environment_id: &str,
        containerfile: &str,
    ) -> Result<String> {
        Ok(self
            .put::<EditedRecipe, _>(
                &format!("/api/infra/services/{service_environment_id}/recipe"),
                &serde_json::json!({ "containerfile": containerfile }),
            )
            .await?
            .source)
    }

    /// One request for the whole file: fifty keys set one at a time is fifty
    /// audit entries and fifty chances to stop halfway.
    pub async fn import_variables(
        &self,
        service_environment_id: &str,
        variables: &[VariableDraft],
    ) -> Result<u32> {
        let imported: ImportedVariables = self
            .post(
                &format!("/api/infra/services/{service_environment_id}/variables/import"),
                &serde_json::json!({ "variables": variables }),
            )
            .await?;

        // A server that does not count them back is not a failure; what was
        // asked for is what to report.
        Ok(match imported.imported {
            0 => variables.len() as u32,
            counted => counted,
        })
    }

    // ----------------------------------------------------------- workspaces

    pub async fn workspaces(&self) -> Result<Vec<Workspace>> {
        Ok(self
            .get::<WorkspacesPage>("/api/infra/workspaces")
            .await?
            .workspaces)
    }

    /// The same id-or-slug-or-name lookup projects get, for the same reason.
    ///
    /// With no reference it answers the first one, which is the workspace
    /// somebody created rather than one they were invited into — the same
    /// default the server applies when a request does not name one.
    pub async fn find_workspace(&self, reference: Option<&str>) -> Result<Workspace> {
        let workspaces = self.workspaces().await?;

        let Some(reference) = reference else {
            return workspaces
                .into_iter()
                .next()
                .ok_or_else(|| anyhow!("this account is in no workspace"));
        };

        let needle = reference.to_lowercase();
        let matched: Vec<Workspace> = workspaces
            .into_iter()
            .filter(|workspace| {
                workspace.id == reference
                    || workspace.slug.to_lowercase() == needle
                    || workspace.name.to_lowercase() == needle
            })
            .collect();

        match matched.len() {
            0 => Err(anyhow!("there is no workspace called `{reference}`")),
            1 => Ok(matched.into_iter().next().expect("length checked")),
            other => Err(anyhow!(
                "`{reference}` matches {other} workspaces — use the id"
            )),
        }
    }

    /// What has been done to a project, newest first.
    pub async fn activity(&self, project_id: &str, limit: usize) -> Result<Vec<ActivityEntry>> {
        Ok(self
            .get::<ActivityPage>(&format!(
                "/api/infra/projects/{project_id}/activity?limit={limit}"
            ))
            .await?
            .activity)
    }

    /// What has been getting into this token's own account, newest first.
    ///
    /// `/api/auth/**` rather than `/api/infra/**`, and that is not an
    /// inconsistency: the history belongs to the account, not to a project, and
    /// the bearer token resolves to the same principal on either prefix. There
    /// is no user id in the path because there is nowhere else to point it.
    pub async fn sign_ins(&self, limit: usize) -> Result<Vec<SignIn>> {
        Ok(self
            .get::<SignInPage>(&format!("/api/auth/sign-ins?limit={limit}"))
            .await?
            .sign_ins)
    }

    pub async fn auth_sessions(&self) -> Result<Vec<AuthSession>> {
        Ok(self
            .get::<AuthSessionPage>("/api/auth/sessions")
            .await?
            .sessions)
    }

    /// Signs one out. The id is the one `auth_sessions` reported.
    pub async fn revoke_session(&self, session_id: &str) -> Result<()> {
        self.delete(&format!("/api/auth/sessions/{session_id}"))
            .await
    }

    pub async fn members(&self, workspace_id: &str) -> Result<Vec<Member>> {
        Ok(self
            .get::<MembersPage>(&format!("/api/infra/workspaces/{workspace_id}/members"))
            .await?
            .members)
    }

    /// What is left of the prepaid credit the agents of a space run on.
    pub async fn credits(&self, workspace_id: &str) -> Result<Credits> {
        self.get::<Credits>(&format!("/api/infra/workspaces/{workspace_id}/credits"))
            .await
    }

    pub async fn invitations(&self, workspace_id: &str) -> Result<Vec<Invitation>> {
        Ok(self
            .get::<InvitationsPage>(&format!("/api/infra/workspaces/{workspace_id}/invitations"))
            .await?
            .invitations)
    }

    pub async fn invite(
        &self,
        workspace_id: &str,
        email: &str,
        role: &str,
    ) -> Result<SentInvitation> {
        self.post(
            &format!("/api/infra/workspaces/{workspace_id}/invitations"),
            &NewInvitation {
                email: email.to_string(),
                role: role.to_string(),
            },
        )
        .await
    }

    pub async fn revoke_invitation(&self, workspace_id: &str, invitation_id: &str) -> Result<()> {
        self.delete(&format!(
            "/api/infra/workspaces/{workspace_id}/invitations/{invitation_id}"
        ))
        .await
    }

    pub async fn remove_member(&self, workspace_id: &str, user_id: &str) -> Result<()> {
        self.delete(&format!(
            "/api/infra/workspaces/{workspace_id}/members/{user_id}"
        ))
        .await
    }

    // ------------------------------------------------------------ streaming

    /// Replays and then follows a deployment's log.
    ///
    /// Server-sent events parsed by hand rather than through a crate: the whole
    /// format this needs is `event:`, `data:` and a blank line, and a
    /// dependency for forty lines of splitting is a dependency to update.
    pub async fn stream_log<F>(
        &self,
        deployment_id: &str,
        after: u64,
        stage: Option<&str>,
        mut on_line: F,
    ) -> Result<()>
    where
        F: FnMut(LogLine),
    {
        let response = self
            .streaming_request(&format!(
                "/api/infra/deployments/{deployment_id}/log/stream?after={after}{}",
                stage_query(stage)
            ))
            .send()
            .await?;

        if !response.status().is_success() {
            Self::check(response).await?;
            return Ok(());
        }

        let mut stream = response.bytes_stream();
        let mut buffer = String::new();

        while let Some(chunk) = stream.next().await {
            buffer.push_str(&String::from_utf8_lossy(&chunk?));

            // Events are separated by a blank line; anything after the last one
            // is a partial event and stays in the buffer for the next chunk.
            while let Some(boundary) = buffer.find("\n\n") {
                let event: String = buffer.drain(..boundary + 2).collect();
                if let Some(line) = parse_log_event(&event) {
                    on_line(line);
                }
                if event.contains("event: end") {
                    return Ok(());
                }
            }
        }

        Ok(())
    }

    /// Everything the platform is doing, until Ctrl-C.
    ///
    /// The same stream the web interface holds open, filtered by the same guard:
    /// what arrives here is what this token's owner is allowed to see, decided
    /// on the server. It never ends on its own — there is no `event: end` on a
    /// firehose — so the loop runs until the process is stopped or the socket
    /// drops.
    pub async fn stream_events<F>(&self, mut on_event: F) -> Result<()>
    where
        F: FnMut(InfraEvent),
    {
        let response = self.streaming_request("/api/infra/stream").send().await?;

        if !response.status().is_success() {
            Self::check(response).await?;
            return Ok(());
        }

        let mut stream = response.bytes_stream();
        let mut buffer = String::new();

        while let Some(chunk) = stream.next().await {
            buffer.push_str(&String::from_utf8_lossy(&chunk?));

            while let Some(boundary) = buffer.find("\n\n") {
                let event: String = buffer.drain(..boundary + 2).collect();
                // `ready` carries the connection id, which is presence's
                // business and a terminal has no field to focus.
                if event.contains("event: infra") {
                    if let Some(parsed) = parse_infra_event(&event) {
                        on_event(parsed);
                    }
                }
            }
        }

        Ok(())
    }

    /// One page of log history, for the non-following case.
    pub async fn read_log(
        &self,
        deployment_id: &str,
        after: u64,
        stage: Option<&str>,
        trace: Option<&str>,
    ) -> Result<Vec<LogLine>> {
        Ok(self
            .get::<LogPage>(&format!(
                "/api/infra/deployments/{deployment_id}/log?after={after}{}{}",
                stage_query(stage),
                trace_query(trace)
            ))
            .await?
            .lines)
    }
}

/// `&stage=build`, or nothing at all.
///
/// The step is one of six names the server validates, so it needs no escaping —
/// and an unknown one is refused there rather than quietly returning the whole
/// log under a heading that says one step.
fn stage_query(stage: Option<&str>) -> String {
    stage.map_or_else(String::new, |name| format!("&stage={name}"))
}

/// `&traceId=…`, or nothing at all.
///
/// Unescaped for the same reason: the server validates the shape a trace id can
/// have and refuses anything else, so a value that would need escaping is one
/// that was never going to match a row.
fn trace_query(trace: Option<&str>) -> String {
    trace.map_or_else(String::new, |id| format!("&traceId={id}"))
}

/// The `data:` payload of one event, when it is a log line.
fn parse_log_event(event: &str) -> Option<LogLine> {
    serde_json::from_str::<LogLine>(&payload_of(event)?).ok()
}

/// The same, for the firehose the whole platform publishes onto.
fn parse_infra_event(event: &str) -> Option<InfraEvent> {
    serde_json::from_str::<InfraEvent>(&payload_of(event)?).ok()
}

/// The `data:` lines of one event, rejoined.
///
/// Several because a payload containing a newline is split across them on the
/// way out, and reading only the first would truncate it silently.
fn payload_of(event: &str) -> Option<String> {
    let payload: String = event
        .lines()
        .filter_map(|line| line.strip_prefix("data:"))
        .map(str::trim_start)
        .collect::<Vec<_>>()
        .join("\n");

    if payload.is_empty() {
        None
    } else {
        Some(payload)
    }
}

/// Refuses to carry a bearer token over plain HTTP to anywhere but this machine.
///
/// The token is the whole of the account — no passkey stands behind a program
/// — and `--url` is also `FUNDAIA_URL`, which a `.envrc` or a CI job can set
/// without the person at the keyboard reading it. Loopback is exempt because
/// local development is `http://localhost:5273` and the server tolerates the
/// same exception for WebAuthn. `FUNDAIA_INSECURE_HTTP=1` is the deliberate
/// override, named so it reads as what it is.
fn require_transport_security(base: &str) -> Result<()> {
    let trimmed = base.trim();
    let scheme = trimmed
        .split_once("://")
        .map(|(scheme, _)| scheme.to_ascii_lowercase());

    match scheme.as_deref() {
        Some("https") => Ok(()),
        Some("http") => {
            let host = host_of(trimmed);
            let name = host
                .rsplit_once(':')
                .map_or(host.as_str(), |(name, _)| name);
            if is_loopback(name) || std::env::var("FUNDAIA_INSECURE_HTTP").is_ok_and(|v| v == "1") {
                Ok(())
            } else {
                bail!(
                    "{trimmed} is plain http: the token would travel in clear. Use https, or set FUNDAIA_INSECURE_HTTP=1 if you really mean it"
                )
            }
        }
        _ => bail!("{trimmed} is not an http(s) address"),
    }
}

fn is_loopback(host: &str) -> bool {
    matches!(host, "localhost" | "127.0.0.1" | "[::1]" | "::1")
        || host.strip_suffix(".localhost").is_some()
        || host.starts_with("127.")
}

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

    #[test]
    fn it_should_read_a_log_line_out_of_an_event() {
        let event = "id: 4\nevent: infra\ndata: {\"sequence\":4,\"stream\":\"build\",\"level\":\"info\",\"message\":\"hello\"}\n\n";
        assert_eq!(parse_log_event(event).unwrap().message, "hello");
    }

    #[test]
    fn it_should_ignore_an_event_with_no_data() {
        assert!(parse_log_event("event: end\n\n").is_none());
    }

    #[test]
    fn it_should_read_a_deployment_out_of_an_infra_event() {
        let event = "event: infra\ndata: {\"kind\":\"deployment-status\",\"projectId\":\"prj_1\",\"status\":\"queued\",\"trigger\":\"git-push\",\"serviceName\":\"api\"}\n\n";
        assert_eq!(
            parse_infra_event(event).unwrap().service_name.unwrap(),
            "api"
        );
    }

    /// A client that refused what it does not draw would stop working the day
    /// the server learns to publish a seventh kind.
    #[test]
    fn it_should_read_an_event_whose_kind_it_has_never_heard_of() {
        let event = "event: infra\ndata: {\"kind\":\"something-new\",\"projectId\":\"prj_1\"}\n\n";
        assert_eq!(parse_infra_event(event).unwrap().kind, "something-new");
    }

    #[test]
    fn it_should_ignore_an_event_whose_data_is_not_a_log_line() {
        assert!(parse_log_event("data: {\"kind\":\"usage\"}\n\n").is_none());
    }

    /// A build can print tens of thousands of lines and every one goes through
    /// this parser; a `watch` left open overnight is the same story. The budget
    /// is far above what the code does today — it exists to catch the change
    /// that accidentally makes parsing quadratic, not to time-trial it.
    #[test]
    #[ignore = "performance budget — the pipeline runs it with --release"]
    fn it_should_parse_fifty_thousand_events_within_the_budget() {
        let event = "id: 4\nevent: infra\ndata: {\"sequence\":4,\"stream\":\"build\",\"level\":\"info\",\"message\":\"compiling something long enough to be a real line of build output\"}\n\n";

        let started = std::time::Instant::now();
        for _ in 0..50_000 {
            let _ = parse_log_event(event);
        }

        assert!(started.elapsed() < Duration::from_secs(1));
    }

    #[test]
    fn it_should_join_a_payload_split_across_several_data_lines() {
        let event = "data: {\"sequence\":1,\"message\":\n data: \"split\"}\n\n";
        // Malformed on purpose: the point is that it does not panic.
        let _ = parse_log_event(event);
    }
}