fundaia 0.32.1

use std::ops::ControlFlow;
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};

mod accounts;
mod data;
mod deployments;
mod domains;
mod projects;
mod services;
mod storage;
mod variables;
mod workflows;

/// Every call this tool makes: the wire in this file, and one file per resource
/// under `client/` with the calls themselves.
///
/// 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,
    token: Option<String>,
}

/// Long enough for a cold start behind the gateway, short enough that a wedged
/// connection does not hang a terminal for ever.
const REQUEST_TIMEOUT: Duration = Duration::from_secs(60);

/*
 * What a stream or an upload gets instead of `REQUEST_TIMEOUT`, which is the
 * whole request, body included.
 *
 * That 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.
 */
const SILENCE_TIMEOUT: Duration = Duration::from_secs(90);

/// 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)?;

        /*
         * One client for every request, so one TLS configuration is built — the
         * root store is parsed once per client, and it was the largest cost of
         * starting a command — and every request shares one connection pool.
         * The silence limit is the client's; the whole-request budget is added
         * per request, and only to the requests that are not streams.
         */
        let http = reqwest::Client::builder()
            .read_timeout(SILENCE_TIMEOUT)
            // 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")?;

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

    /// The same instance and connection pool, carrying another token.
    pub fn with_token(&self, token: String) -> Self {
        Self {
            token: Some(token),
            ..self.clone()
        }
    }

    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
    }

    /// A POST that is a verb with nothing to say: the body is `{}`, which is
    /// what the server's JSON parser expects rather than no body at all.
    pub async fn post_empty<T: DeserializeOwned>(&self, path: &str) -> Result<T> {
        self.post(path, &serde_json::json!({})).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.unbounded_request(method, path, body)
            .timeout(REQUEST_TIMEOUT)
    }

    /// A stream: bounded by silence rather than by its length.
    fn streaming_request(&self, path: &str) -> reqwest::RequestBuilder {
        self.unbounded_request(Method::GET, path, None::<&()>)
            .header("accept", "text/event-stream")
    }

    /// With no whole-request budget: only the client's limit on silence.
    fn unbounded_request<B: Serialize>(
        &self,
        method: Method,
        path: &str,
        body: Option<&B>,
    ) -> reqwest::RequestBuilder {
        let mut builder = self.http.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())
    }

    /// 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),
    {
        self.read_events("/api/infra/stream", |event| {
            // `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);
                }
            }
            ControlFlow::Continue(())
        })
        .await
    }

    /// A stream of server-sent events, one whole event at a time, until it ends
    /// or `on_event` says stop.
    ///
    /// 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.
    ///
    /// Split as bytes and decoded an event at a time: the blank line between
    /// two events is ASCII, so a cut there never lands inside a character — a
    /// chunk boundary can — and what has been handed over leaves the buffer
    /// once per chunk rather than being shifted out of it once per event.
    async fn read_events(
        &self,
        path: &str,
        mut on_event: impl FnMut(&str) -> ControlFlow<()>,
    ) -> Result<()> {
        let response = self.streaming_request(path).send().await?;

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

        let mut stream = response.bytes_stream();
        let mut pending: Vec<u8> = Vec::new();

        while let Some(chunk) = stream.next().await {
            pending.extend_from_slice(&chunk?);
            if hand_over_events(&mut pending, &mut on_event).is_break() {
                break;
            }
        }

        Ok(())
    }
}

/// What a person can call a project, a service or a workspace.
trait Named {
    fn id(&self) -> &str;
    fn slug(&self) -> &str;
    fn name(&self) -> &str;
}

macro_rules! named {
    ($($kind:ty),+) => {$(
        impl Named for $kind {
            fn id(&self) -> &str { &self.id }
            fn slug(&self) -> &str { &self.slug }
            fn name(&self) -> &str { &self.name }
        }
    )+};
}

named!(Project, Service, Workspace);

impl<T: Named> Named for &T {
    fn id(&self) -> &str {
        (*self).id()
    }
    fn slug(&self) -> &str {
        (*self).slug()
    }
    fn name(&self) -> &str {
        (*self).name()
    }
}

/// The one candidate a reference names: its id, or its slug or name in any case.
///
/// 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.
fn one_named<T: Named>(
    candidates: impl IntoIterator<Item = T>,
    reference: &str,
    plural: &str,
    missing: impl FnOnce() -> String,
) -> Result<T> {
    let needle = reference.to_lowercase();
    let mut matched = candidates.into_iter().filter(|candidate| {
        candidate.id() == reference
            || candidate.slug().to_lowercase() == needle
            || candidate.name().to_lowercase() == needle
    });

    let Some(found) = matched.next() else {
        return Err(anyhow!(missing()));
    };
    match matched.count() {
        0 => Ok(found),
        others => Err(anyhow!(
            "`{reference}` matches {} {plural} — use the id",
            others + 1
        )),
    }
}

fn service_named<T: Named>(services: impl IntoIterator<Item = T>, reference: &str) -> Result<T> {
    one_named(services, reference, "services", || {
        format!("this project has no service called `{reference}`")
    })
}

/// `&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}"))
}

/// Every whole event in `pending`, in order, leaving the partial one after the
/// last blank line for the next chunk to complete.
fn hand_over_events(
    pending: &mut Vec<u8>,
    on_event: &mut impl FnMut(&str) -> ControlFlow<()>,
) -> ControlFlow<()> {
    let mut start = 0;
    let mut outcome = ControlFlow::Continue(());

    while let Some(blank) = pending[start..].windows(2).position(|pair| pair == b"\n\n") {
        let end = start + blank + 2;
        outcome = on_event(&String::from_utf8_lossy(&pending[start..end]));
        start = end;
        if outcome.is_break() {
            break;
        }
    }

    pending.drain(..start);
    outcome
}

/// 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::*;

    fn workspace(id: &str, slug: &str, name: &str) -> Workspace {
        Workspace {
            id: id.to_owned(),
            slug: slug.to_owned(),
            name: name.to_owned(),
            role: None,
        }
    }

    fn spaces() -> Vec<Workspace> {
        vec![
            workspace("wsp_1", "acme", "Acme"),
            workspace("wsp_2", "equipo", "Equipo"),
            workspace("wsp_3", "otro", "equipo"),
        ]
    }

    #[test]
    fn it_should_find_a_name_whatever_its_case() {
        let found = one_named(spaces(), "ACME", "workspaces", String::new).unwrap();
        assert_eq!(found.id, "wsp_1");
    }

    #[test]
    fn it_should_find_an_id_exactly() {
        let found = one_named(spaces(), "wsp_2", "workspaces", String::new).unwrap();
        assert_eq!(found.slug, "equipo");
    }

    /// Picking one silently is how something deploys to the wrong place.
    #[test]
    fn it_should_refuse_a_reference_that_names_two() {
        let refused = one_named(spaces(), "equipo", "workspaces", String::new).unwrap_err();
        assert_eq!(
            refused.to_string(),
            "`equipo` matches 2 workspaces — use the id"
        );
    }

    #[test]
    fn it_should_say_what_it_could_not_find() {
        let refused =
            one_named(spaces(), "nadie", "workspaces", || "no está".to_owned()).unwrap_err();
        assert_eq!(refused.to_string(), "no está");
    }

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

    fn events_in(chunks: &[&[u8]]) -> Vec<String> {
        let mut pending = Vec::new();
        let mut seen = Vec::new();
        for chunk in chunks {
            pending.extend_from_slice(chunk);
            let _ = hand_over_events(&mut pending, &mut |event: &str| {
                seen.push(event.to_owned());
                ControlFlow::Continue(())
            });
        }
        seen
    }

    #[test]
    fn it_should_hand_over_every_event_in_one_chunk() {
        assert_eq!(events_in(&[b"data: 1\n\ndata: 2\n\n"]).len(), 2);
    }

    #[test]
    fn it_should_wait_for_the_rest_of_an_event_cut_by_a_chunk() {
        assert_eq!(
            events_in(&[b"data: {\"a\":", b"1}\n\n"]),
            vec!["data: {\"a\":1}\n\n"]
        );
    }

    /// `ñ` is two bytes, and a chunk can end between them: decoding each chunk on
    /// its own turned both halves into replacement characters.
    #[test]
    fn it_should_keep_a_character_cut_in_half_by_a_chunk() {
        assert_eq!(
            events_in(&[b"data: \xc3", b"\xb1\n\n"]),
            vec!["data: ñ\n\n"]
        );
    }

    #[test]
    fn it_should_stop_handing_over_once_told_to() {
        let mut pending = b"event: end\n\ndata: late\n\n".to_vec();
        let mut seen = 0;
        let _ = hand_over_events(&mut pending, &mut |_: &str| {
            seen += 1;
            ControlFlow::Break(())
        });
        assert_eq!(seen, 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);
    }
}