openlatch-client 0.5.4

//! PostHog `/batch/` HTTP poster.
//!
//! We POST directly rather than depend on `posthog-rs` (still <1.0, unstable
//! API). The endpoint contract is well-documented and stable:
//! <https://posthog.com/docs/api/capture#batch-events>.
//!
//! Body shape:
//! ```json
//! {
//!   "api_key": "phc_...",
//!   "batch": [
//!     {
//!       "event": "cli_initialized",
//!       "distinct_id": "agt_...",
//!       "properties": { ... },
//!       "timestamp": "2026-04-13T...Z"
//!     }
//!   ]
//! }
//! ```
//!
//! Failures are swallowed: events drop silently with no retry, no disk queue
//! (invariant I4). Telemetry never blocks anything else, never produces a
//! user-visible error.

use std::future::Future;
use std::pin::Pin;
use std::sync::OnceLock;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

use serde_json::{json, Value};

use super::client::QueuedEvent;

/// How long one POST may wait on the ingest host. Enforced by [`WaitBudget`], never by
/// the reqwest client — see there for why.
const POST_TIMEOUT: Duration = Duration::from_secs(5);

/// Resolve the PostHog project key. Runtime env var wins (developer override),
/// otherwise fall back to the value baked at build time by `build.rs`.
pub fn posthog_key() -> &'static str {
    static KEY: OnceLock<String> = OnceLock::new();
    KEY.get_or_init(|| {
        std::env::var("OPENLATCH_POSTHOG_KEY")
            .unwrap_or_else(|_| env!("OPENLATCH_POSTHOG_KEY").to_string())
    })
}

/// True if a non-empty key is available — gates whether `init()` even bothers
/// to construct the network client (invariant I1).
pub fn key_is_present() -> bool {
    !posthog_key().is_empty()
}

/// Resolve the ingestion host from the platform origin `api_url`.
///
/// The default is the platform's own first-party proxy, `{api_url}/ingest`, which
/// forwards to PostHog unchanged. That is what lets a customer allowlist ONE FQDN for
/// everything this client sends. There is deliberately no fallback to PostHog
/// directly: a proxy that is down drops the event (I4), because a fallback would
/// reopen the second destination the proxy exists to close.
///
/// A non-empty `OPENLATCH_POSTHOG_HOST` wins so tests and the e2e suite can point at a
/// mock without rebuilding. An EMPTY override falls through rather than being
/// honoured. Both consumers read this one resolver — product telemetry and the crash
/// path — so an exported `OPENLATCH_POSTHOG_HOST=""` would otherwise send one of them
/// to `""` while the other kept working. One resolver, one behaviour, is the point.
pub fn resolve_host(api_url: &str) -> String {
    resolve_host_with(
        std::env::var("OPENLATCH_POSTHOG_HOST").ok().as_deref(),
        api_url,
    )
}

/// The resolver's logic, with the environment read hoisted out so it can be tested
/// without mutating process-global state.
///
/// A blank `api_url` — what `Config::load` yields for `OPENLATCH_API_URL=""` — is the
/// compiled-in platform origin, never an empty host and never a PostHog one.
pub(super) fn resolve_host_with(host_override: Option<&str>, api_url: &str) -> String {
    if let Some(host) = host_override.filter(|h| !h.is_empty()) {
        return host.to_string();
    }
    let api_url = api_url.trim();
    let origin = if api_url.is_empty() {
        crate::config::CloudConfig::default().api_url
    } else {
        api_url.to_string()
    };
    format!("{}/ingest", origin.trim_end_matches('/'))
}

/// The deployment environment this binary reports as.
///
/// A debug build is always `development`. A RELEASE build is `development` too when no
/// project key was baked — which is exactly what a developer's own `cargo build
/// --release` and every CI/e2e binary produce. Without that second clause every local
/// release build would report `production` and trip the per-occurrence alerting.
///
/// There is no `staging` for the client: one key is baked per release, and it is the
/// production project's.
pub fn environment() -> &'static str {
    if cfg!(debug_assertions) || env!("OPENLATCH_POSTHOG_KEY").is_empty() {
        "development"
    } else {
        "production"
    }
}

/// The environment string a debug build — or a release build with no baked project key —
/// reports as. Lives here, beside [`environment`], because both send paths gate on it.
pub const ENV_DEVELOPMENT: &str = "development";

/// THE DEVELOPMENT GATE, shared by product telemetry and the crash path.
///
/// A build reporting `development` may send only to a LOOPBACK destination. Consent and
/// a key are not enough: a developer who exports `OPENLATCH_POSTHOG_KEY` gets a debug
/// build with a key present, and the project a PostHog event lands in is chosen by that
/// key alone — `/ingest` is a transparent proxy, so the destination host never picks the
/// project. Without this gate those events land in a real project tagged
/// `environment = development`.
///
/// LOOPBACK, specifically, is what keeps the e2e suite reachable. No CI job bakes a
/// project key, so every binary the suite runs resolves `development`; the suite points
/// `OPENLATCH_POSTHOG_HOST` (or, for the derived-host case, `OPENLATCH_API_URL`) at a
/// mock on `127.0.0.1`. A loopback address cannot be a real PostHog project, so the
/// carve-out costs nothing — whereas admitting ANY explicit host, as this gate once did,
/// re-opens it for a developer who exports a real host and a real key.
///
/// `host` is the effective destination — [`resolve_host`]'s answer, not the raw
/// override — so an `OPENLATCH_POSTHOG_HOST=""` that the resolver ignores cannot open
/// the gate either: it falls through to the platform origin, which is not loopback.
pub fn may_send_to(environment: &str, host: &str) -> bool {
    environment != ENV_DEVELOPMENT || host_is_loopback(host)
}

/// Whether `host` names a loopback interface: `localhost` / `*.localhost`, anything in
/// `127.0.0.0/8`, or `::1`.
///
/// Parse with the same URL implementation `reqwest` uses to send the request, rather
/// than extracting an authority by hand. In particular, a backslash terminates an HTTP
/// authority for URL parsing: `http://evil.example\\@127.0.0.1` goes to `evil.example`,
/// not loopback. Anything that does not parse to a recognised loopback host closes the
/// gate rather than opening it.
fn host_is_loopback(host: &str) -> bool {
    let Ok(url) = reqwest::Url::parse(host) else {
        return false;
    };
    let Some(hostname) = url.host_str() else {
        return false;
    };
    // `Url::host_str` keeps brackets around an IPv6 literal; `IpAddr` accepts its
    // address form without them.
    let hostname = hostname.trim_matches(['[', ']']);
    hostname.eq_ignore_ascii_case("localhost")
        || hostname.to_ascii_lowercase().ends_with(".localhost")
        || hostname
            .parse::<std::net::IpAddr>()
            .is_ok_and(|ip| ip.is_loopback())
}

/// The git sha baked at build time, reported as the event's `release`.
pub fn release() -> &'static str {
    env!("OPENLATCH_RELEASE_SHA")
}

/// Build a long-lived reqwest client. Reused across all batch POSTs so the
/// connection pool stays warm.
///
/// Deliberately without a client-level deadline: [`post_batch`] bounds each POST with a
/// [`WaitBudget`] instead, because reqwest's own deadline discards exactly the handshake
/// the first batch needs.
///
/// `None` means "skip POSTs" — telemetry never fails a user command over its own
/// transport (I10), so a client we cannot build is a warning and nothing more.
pub fn build_client(egress: &crate::egress::EgressConfig) -> Option<reqwest::Client> {
    match crate::egress::build_client(crate::egress::Consumer::Telemetry, egress) {
        Ok(c) => Some(c),
        Err(e) => {
            tracing::warn!(error = %e, "telemetry http client init failed; events will be dropped");
            None
        }
    }
}

/// POST a batch to `{host}/batch/`, `host` being what [`resolve_host`] returned.
/// Silent on failure (I4 — no retry, no log surface, no telemetry-about-telemetry).
/// Returns true if the POST succeeded at the HTTP level (2xx); used only to stamp
/// `last_sent_unix` in the handle.
pub async fn post_batch(client: &reqwest::Client, host: &str, batch: &[QueuedEvent]) -> bool {
    if batch.is_empty() {
        return true;
    }
    // Also enforce the no-key invariant at the transport boundary. Tests can
    // construct enabled handles without a baked key; never send those batches
    // to the production ingestion host with an empty credential.
    if !key_is_present() {
        return false;
    }
    // THE DEVELOPMENT GATE, at the same transport boundary and from the same predicate
    // the crash path uses. A development build sends only to a loopback destination; a
    // developer who exports a real key and a real host would otherwise post into a real
    // project, which is the one thing the environment tag cannot undo after the fact.
    if !may_send_to(environment(), host) {
        return false;
    }
    let url = batch_url(host);
    let body = build_body(batch);
    match WaitBudget::new(client.post(&url).json(&body).send(), POST_TIMEOUT).await {
        Some(Ok(resp)) => resp.status().is_success(),
        Some(Err(_)) | None => false,
    }
}

/// A deadline charged only for the time a request spends *waiting*.
///
/// The telemetry thread can be blocked inside a poll. On Windows and macOS the TLS
/// handshake asks the OS to evaluate trust, synchronously, and on a cold cache the OS
/// fetches revocation data over the network before it answers — for seconds, and past
/// `POST_TIMEOUT` on a slow or filtered network. No timer fires on a thread that is
/// inside a call, so a wall-clock deadline comes due unobserved and is only seen once
/// the handshake has completed. reqwest checks its deadline before it polls the request,
/// so that deadline then discards the connection the handshake just produced, batch
/// unsent. The first batch is the one that opens the connection, and it carries
/// `daemon_started`.
///
/// So time spent inside the inner `poll` — work, not waiting — pushes the deadline back
/// by as much. A host that never connects, never finishes its side of the handshake or
/// never answers is still cut off at the budget: those are waits, and a parked runtime
/// observes its timers on time. The attempt stays single (I4); the deadline only stops
/// counting time it could not have enforced.
struct WaitBudget<F> {
    inner: Pin<Box<F>>,
    deadline: Pin<Box<tokio::time::Sleep>>,
}

impl<F: Future> WaitBudget<F> {
    fn new(inner: F, budget: Duration) -> Self {
        Self {
            inner: Box::pin(inner),
            deadline: Box::pin(tokio::time::sleep(budget)),
        }
    }
}

impl<F: Future> Future for WaitBudget<F> {
    /// `None` when the budget ran out first.
    type Output = Option<F::Output>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let polled_at = Instant::now();
        if let Poll::Ready(out) = self.inner.as_mut().poll(cx) {
            return Poll::Ready(Some(out));
        }
        let deadline = self.deadline.deadline() + polled_at.elapsed();
        self.deadline.as_mut().reset(deadline);
        self.deadline.as_mut().poll(cx).map(|()| None)
    }
}

fn batch_url(host: &str) -> String {
    format!("{}/batch/", host.trim_end_matches('/'))
}

fn build_body(batch: &[QueuedEvent]) -> Value {
    let events: Vec<Value> = batch.iter().map(event_to_payload).collect();
    json!({
        "api_key": posthog_key(),
        "batch": events,
    })
}

fn event_to_payload(event: &QueuedEvent) -> Value {
    // Use the canonical identity captured with the event, including aliases.
    // agent_id remains a fallback for older queued-event producers.
    let distinct_id = event
        .properties
        .get("distinct_id")
        .or_else(|| event.properties.get("agent_id"))
        .and_then(|v| v.as_str())
        .unwrap_or("agt_unknown")
        .to_string();
    let timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
    json!({
        "event": event.name,
        "distinct_id": distinct_id,
        "properties": event.properties,
        "timestamp": timestamp,
    })
}

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

    fn make_event(name: &str, agent: &str) -> QueuedEvent {
        let mut props = Map::new();
        props.insert("agent_id".into(), json!(agent));
        props.insert("os".into(), json!("linux-x64"));
        QueuedEvent {
            name: name.into(),
            properties: props,
        }
    }

    #[test]
    fn test_build_body_wraps_with_api_key_and_batch() {
        let events = vec![make_event("cli_initialized", "agt_a")];
        let body = build_body(&events);
        assert!(body["api_key"].is_string());
        let batch = body["batch"].as_array().unwrap();
        assert_eq!(batch.len(), 1);
        assert_eq!(batch[0]["event"], "cli_initialized");
        assert_eq!(batch[0]["distinct_id"], "agt_a");
        assert!(batch[0]["timestamp"].is_string());
    }

    #[test]
    fn test_event_to_payload_falls_back_when_agent_id_missing() {
        let mut props = Map::new();
        props.insert("os".into(), json!("linux-x64"));
        let ev = QueuedEvent {
            name: "test".into(),
            properties: props,
        };
        let p = event_to_payload(&ev);
        assert_eq!(p["distinct_id"], "agt_unknown");
    }

    #[test]
    fn test_batch_uses_canonical_user_and_organization() {
        let mut event = make_event("daemon_started", "agt_a");
        let mut props = super::super::super_props::SuperProps::new("agt_a".into(), true);
        props.user_db_id = Some("user_1".into());
        props.org_id = Some("org_1".into());
        props.merge_into(&mut event.properties);
        let body = build_body(&[event]);
        assert_eq!(body["batch"][0]["distinct_id"], "user_1");
        assert_eq!(body["batch"][0]["properties"]["agent_id"], "agt_a");
        assert_eq!(
            body["batch"][0]["properties"]["$groups"]["organization"],
            "org_1"
        );
    }

    #[test]
    fn test_alias_top_level_identity_matches_canonical_user() {
        let event = super::super::identity::create_alias_event("agt_a", "user_1");
        let mut props = event.properties;
        super::super::super_props::SuperProps::new("agt_a".into(), false).merge_into(&mut props);
        let payload = event_to_payload(&QueuedEvent {
            name: event.name,
            properties: props,
        });
        assert_eq!(payload["distinct_id"], "user_1");
        assert_eq!(payload["properties"]["distinct_id"], "user_1");
        assert_eq!(payload["properties"]["alias"], "agt_a");
    }

    #[test]
    fn test_post_batch_empty_short_circuits() {
        // No client needed: empty batch returns true without making a request.
        let rt = tokio::runtime::Runtime::new().unwrap();
        let client = build_client(&crate::egress::EgressConfig::direct()).unwrap();
        let ok = rt.block_on(post_batch(&client, "http://127.0.0.1:9", &[]));
        assert!(ok);
    }

    /// Stands in for a TLS handshake whose trust evaluation runs synchronously: the
    /// first poll blocks the thread past the whole budget, then the request still has a
    /// real wait ahead of it before the answer arrives.
    struct BlocksThenWaits {
        blocked: bool,
        block_for: Duration,
        answer_at: Option<tokio::time::Instant>,
    }

    impl Future for BlocksThenWaits {
        type Output = &'static str;

        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            if !self.blocked {
                self.blocked = true;
                std::thread::sleep(self.block_for);
                self.answer_at = Some(tokio::time::Instant::now() + Duration::from_millis(50));
            }
            let answer_at = self.answer_at.expect("set on first poll");
            if tokio::time::Instant::now() >= answer_at {
                return Poll::Ready("answered");
            }
            let waker = cx.waker().clone();
            tokio::spawn(async move {
                tokio::time::sleep_until(answer_at).await;
                waker.wake();
            });
            Poll::Pending
        }
    }

    /// The regression: a wall-clock deadline that came due while the thread was inside
    /// the handshake discarded the finished handshake, and with it the first batch.
    #[tokio::test]
    async fn test_wait_budget_does_not_charge_time_spent_inside_a_poll() {
        let budget = Duration::from_millis(200);
        let blocking = BlocksThenWaits {
            blocked: false,
            block_for: budget * 3,
            answer_at: None,
        };
        assert_eq!(WaitBudget::new(blocking, budget).await, Some("answered"));
    }

    /// And a host that never answers is still cut off at the budget.
    #[tokio::test]
    async fn test_wait_budget_still_bounds_a_peer_that_never_answers() {
        let budget = Duration::from_millis(200);
        let started = Instant::now();
        assert_eq!(
            WaitBudget::new(std::future::pending::<()>(), budget).await,
            None
        );
        assert!(
            started.elapsed() < budget * 10,
            "the budget did not bound an idle wait: {:?}",
            started.elapsed()
        );
    }

    #[test]
    fn test_host_override_wins_over_the_platform_origin() {
        assert_eq!(
            resolve_host_with(Some("http://127.0.0.1:8123"), "https://app.openlatch.ai"),
            "http://127.0.0.1:8123"
        );
    }

    /// `OPENLATCH_POSTHOG_HOST=""` is the shell's way of neutralising a variable, so
    /// it must fall through to the derived host rather than become an empty one.
    #[test]
    fn test_empty_host_override_is_ignored() {
        assert_eq!(
            resolve_host_with(Some(""), "https://staging.example"),
            "https://staging.example/ingest"
        );
    }

    #[test]
    fn test_default_platform_origin_derives_the_first_party_ingest_host() {
        let default_api_url = crate::config::CloudConfig::default().api_url;
        assert_eq!(
            resolve_host_with(None, &default_api_url),
            "https://app.openlatch.ai/ingest"
        );
    }

    #[test]
    fn test_custom_platform_origin_with_and_without_trailing_slash() {
        for api_url in ["https://ol.corp.example", "https://ol.corp.example/"] {
            assert_eq!(
                resolve_host_with(None, api_url),
                "https://ol.corp.example/ingest",
                "api_url = {api_url:?}"
            );
        }
    }

    /// A blank `OPENLATCH_API_URL` reaches the resolver as a blank `api_url`, both from
    /// `Config::load` and from the load-failure fallback. It is the default origin —
    /// never `/ingest` on its own, and never a PostHog host.
    #[test]
    fn test_blank_platform_origin_is_the_default() {
        for api_url in ["", "   "] {
            assert_eq!(
                resolve_host_with(None, api_url),
                "https://app.openlatch.ai/ingest",
                "api_url = {api_url:?}"
            );
        }
    }

    /// THE DEVELOPMENT GATE, as pure logic. Both send paths read this one predicate —
    /// `post_batch` above and `crash::should_send` — so this matrix is the whole rule.
    #[test]
    fn test_a_development_build_may_send_to_loopback_only() {
        for host in [
            "http://127.0.0.1:8123",
            "http://127.0.0.1:8123/ingest",
            "http://localhost:9",
            "http://LocalHost:9/ingest",
            "http://az-qa-1.localhost:20313/ingest",
            "http://127.9.9.9",
            "http://[::1]:8000/ingest",
            // The e2e derived-host shape: no override, `OPENLATCH_API_URL` at a mock.
            "http://127.0.0.1:34567/ingest",
        ] {
            assert!(
                may_send_to(ENV_DEVELOPMENT, host),
                "loopback must stay reachable: {host:?}"
            );
        }

        for host in [
            "https://eu.i.posthog.com",
            "https://us.i.posthog.com",
            "https://app.openlatch.ai/ingest",
            "https://ol.corp.example/ingest",
            // Contains a loopback address; is not one.
            "https://127.0.0.1.evil.example/ingest",
            "",
        ] {
            assert!(
                !may_send_to(ENV_DEVELOPMENT, host),
                "a development build must not send to {host:?}"
            );
        }
    }

    /// A production build is not gated on the destination at all — that is the whole
    /// point of the environment half of the predicate.
    #[test]
    fn test_a_production_build_sends_wherever_the_host_resolves() {
        for host in [
            "https://app.openlatch.ai/ingest",
            "https://eu.i.posthog.com",
            "http://127.0.0.1:8123",
            "",
        ] {
            assert!(may_send_to("production", host), "host = {host:?}");
        }
    }

    /// The empty override, end to end: it is ignored by the resolver, so the gate sees
    /// the platform origin and stays shut. Exporting `OPENLATCH_POSTHOG_HOST=""` must
    /// never be the thing that opens it.
    #[test]
    fn test_an_empty_host_override_does_not_open_the_development_gate() {
        assert!(!may_send_to(
            ENV_DEVELOPMENT,
            &resolve_host_with(Some(""), "https://app.openlatch.ai")
        ));
    }

    #[test]
    fn test_host_is_loopback_uses_the_transport_url_parser() {
        assert!(host_is_loopback("http://127.0.0.1:8000"));
        assert!(host_is_loopback("http://user:pw@127.0.0.1:8000/ingest"));
        assert!(host_is_loopback("http://[::1]"));
        assert!(host_is_loopback("http://az-qa-1.localhost:20313"));
        assert!(!host_is_loopback("http://127.0.0.1.example.com/ingest"));
        assert!(!host_is_loopback("https://example.com/127.0.0.1"));
        assert!(!host_is_loopback("https://example.com/?h=localhost"));
        assert!(!host_is_loopback(
            "http://evil.example\\@127.0.0.1:8123/ingest"
        ));
        assert!(!host_is_loopback("http://az-qa-1.localhost.evil.example"));
        assert!(!host_is_loopback("http://[::1"));
        assert!(!host_is_loopback(""));
    }

    #[test]
    fn test_batches_post_to_ingest_batch_on_the_platform() {
        assert_eq!(
            batch_url(&resolve_host_with(None, "https://app.openlatch.ai")),
            "https://app.openlatch.ai/ingest/batch/"
        );
    }
}