fundaia 0.7.2

use std::time::Duration;

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

use super::models::*;
use crate::config::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> {
        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))
            .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))
            .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(())
    }

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

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

    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 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)
    }

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

    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)
    }

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

    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
    }

    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
    }

    pub async fn archive_project(&self, project_id: &str) -> Result<()> {
        self.delete(&format!("/api/infra/projects/{project_id}"))
            .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. The count is what to report.
    pub async fn disconnect(&self, consumer_id: &str, provider_id: &str) -> Result<u32> {
        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 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"
            )),
        }
    }

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

    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, mut on_line: F) -> Result<()>
    where
        F: FnMut(LogLine),
    {
        let response = self
            .streaming_request(&format!(
                "/api/infra/deployments/{deployment_id}/log/stream?after={after}"
            ))
            .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) -> Result<Vec<LogLine>> {
        Ok(self
            .get::<LogPage>(&format!(
                "/api/infra/deployments/{deployment_id}/log?after={after}"
            ))
            .await?
            .lines)
    }
}

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

#[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());
    }

    #[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);
    }
}