openlatch-client 0.1.18

//! Background client for the telemetry subsystem.
//!
//! Phase A scope: mpsc channel, background task scaffold, debug-mode stderr
//! emission, graceful drain on shutdown. No network path yet — `posthog-rs`
//! arrives in Task 3 along with the baked key.
//!
//! Invariants preserved here:
//! - I1 / I2: when consent resolves disabled, `init()` returns `None`. No
//!   channel, no task, no allocation on the hot path.
//! - I4: drops are silent. No disk queue, no retry state.
//! - I10: failures never produce telemetry events describing themselves.

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use serde_json::{json, Map, Value};
use tokio::sync::{mpsc, oneshot};

use super::consent::Resolved;
use super::events::Event;
use super::super_props::SuperProps;

const CHANNEL_CAPACITY: usize = 1000;
const BATCH_SIZE: usize = 10;
const BATCH_INTERVAL: Duration = Duration::from_secs(1);

/// Queued event waiting to be flushed. Super-properties are merged in at
/// enqueue time so the background task sees a fully-populated payload.
#[derive(Debug, Clone)]
pub struct QueuedEvent {
    pub name: String,
    pub properties: Map<String, Value>,
}

/// Traffic on the batch channel.
///
/// `Flush` exists because the batch loop's only other drain triggers are the
/// `BATCH_INTERVAL` timer and the channel closing — and neither can help a
/// process that is about to exit. The global handle lives in a `OnceLock` for
/// the life of the program, so its sender is never dropped and the
/// close-triggered final drain never runs; a `daemon_stopped` captured on the
/// way out would sit in the buffer until the runtime was torn down under it.
#[derive(Debug)]
pub enum Msg {
    Event(QueuedEvent),
    /// Flush the buffer now, then acknowledge so the caller knows the POST is
    /// done rather than merely started.
    Flush(oneshot::Sender<()>),
}

/// Handle returned from `init`. Drops are no-ops; `shutdown().await` drains
/// the queue and stops the background task cleanly.
///
/// Cheap to clone — the inner state is reference-counted.
#[derive(Clone)]
pub struct TelemetryHandle {
    inner: Arc<Inner>,
}

struct Inner {
    sender: Option<mpsc::Sender<Msg>>,
    super_props: SuperProps,
    last_sent_unix: AtomicU64,
    events_captured: AtomicU64,
    events_dropped: AtomicU64,
    debug_stderr: bool,
    disabled: AtomicBool,
}

impl TelemetryHandle {
    /// Build a disabled handle — `capture()` is a no-op. Used when consent
    /// resolves disabled, when the baked key is empty, or when the module
    /// was compiled with `full-cli-no-telemetry`.
    pub fn disabled(agent_id: String, authenticated: bool) -> Self {
        let inner = Inner {
            sender: None,
            super_props: SuperProps::new(agent_id, authenticated),
            last_sent_unix: AtomicU64::new(0),
            events_captured: AtomicU64::new(0),
            events_dropped: AtomicU64::new(0),
            debug_stderr: false,
            disabled: AtomicBool::new(true),
        };
        Self {
            inner: Arc::new(inner),
        }
    }

    pub fn is_enabled(&self) -> bool {
        !self.inner.disabled.load(Ordering::Relaxed) && self.inner.sender.is_some()
    }

    /// Non-blocking, non-fallible capture. Drops on full channel (I4).
    pub fn capture(&self, event: Event) {
        if self.inner.disabled.load(Ordering::Relaxed) {
            return;
        }
        let Some(sender) = self.inner.sender.as_ref() else {
            return;
        };

        let mut properties = event.properties;
        self.inner.super_props.merge_into(&mut properties);

        let queued = QueuedEvent {
            name: event.name,
            properties,
        };

        if self.inner.debug_stderr {
            emit_debug(&queued);
        }

        match sender.try_send(Msg::Event(queued)) {
            Ok(()) => {
                self.inner.events_captured.fetch_add(1, Ordering::Relaxed);
            }
            Err(_) => {
                self.inner.events_dropped.fetch_add(1, Ordering::Relaxed);
            }
        }
    }

    /// Flush buffered events and wait up to `budget` for the POST to finish.
    ///
    /// Returns `true` if the batch task acknowledged in time (or there was
    /// nothing to do because telemetry is disabled), `false` if the budget
    /// expired or the task was already gone. Never an error path: a lost
    /// shutdown event is a missing metric, not a failed shutdown, so callers
    /// log at most and carry on.
    pub async fn flush(&self, budget: Duration) -> bool {
        if self.inner.disabled.load(Ordering::Relaxed) {
            return true;
        }
        let Some(sender) = self.inner.sender.as_ref() else {
            return true;
        };
        let (ack_tx, ack_rx) = oneshot::channel();
        // `try_send`, matching `capture`: a full channel means the batch task
        // is already saturated with work it will flush anyway, and blocking a
        // shutdown path on a full queue is worse than losing the ack.
        if sender.try_send(Msg::Flush(ack_tx)).is_err() {
            return false;
        }
        matches!(tokio::time::timeout(budget, ack_rx).await, Ok(Ok(())))
    }

    pub fn last_sent_unix(&self) -> u64 {
        self.inner.last_sent_unix.load(Ordering::Relaxed)
    }

    pub fn events_captured(&self) -> u64 {
        self.inner.events_captured.load(Ordering::Relaxed)
    }

    pub fn events_dropped(&self) -> u64 {
        self.inner.events_dropped.load(Ordering::Relaxed)
    }
}

/// Configuration resolved by `super::init()` and passed to the client.
pub(super) struct ClientConfig {
    pub resolved: Resolved,
    pub super_props: SuperProps,
    pub debug_stderr: bool,
    pub baked_key_present: bool,
}

/// Construct a live handle and spawn the background batch task. When the
/// runtime has no baked key or consent is disabled, returns a no-op handle
/// and never spawns a task (I1 / I2).
pub(super) fn start(cfg: ClientConfig) -> TelemetryHandle {
    if !cfg.resolved.enabled() || !cfg.baked_key_present {
        // Invariant I2: zero code path when disabled. No channel, no task.
        return TelemetryHandle {
            inner: Arc::new(Inner {
                sender: None,
                super_props: cfg.super_props,
                last_sent_unix: AtomicU64::new(0),
                events_captured: AtomicU64::new(0),
                events_dropped: AtomicU64::new(0),
                debug_stderr: cfg.debug_stderr,
                disabled: AtomicBool::new(true),
            }),
        };
    }

    let (tx, rx) = mpsc::channel::<Msg>(CHANNEL_CAPACITY);
    let last_sent = Arc::new(AtomicU64::new(0));

    // Run the batch loop regardless of caller context. If we're already
    // inside a tokio runtime (tests, `rt.block_on` closures) we spawn onto
    // it. Otherwise — the common case for the main `openlatch` binary, whose
    // `main()` is synchronous — we own a dedicated OS thread that builds its
    // own single-threaded runtime and drives the loop to completion.
    //
    // Shutdown is structural: when the final `TelemetryHandle` clone drops,
    // `Inner` drops, the mpsc `Sender` drops, `rx.recv()` in the batch loop
    // returns `None`, `run_batch_loop` exits, the owned runtime drops, and
    // the thread joins. No explicit stop signal needed.
    let last_sent_bg = Arc::clone(&last_sent);
    let debug = cfg.debug_stderr;
    // Build a long-lived reqwest client up front so the batch loop owns a
    // connection pool. `None` means "skip POSTs" — preserves I1 if the
    // builder fails for some unforeseen reason.
    let http = super::network::build_client();
    if let Ok(handle) = tokio::runtime::Handle::try_current() {
        handle.spawn(async move {
            run_batch_loop(rx, last_sent_bg, debug, http).await;
        });
    } else {
        let _ = std::thread::Builder::new()
            .name("openlatch-telemetry".into())
            .spawn(move || {
                // Thread spawn failures and runtime build failures are both
                // fatally silent (I10 — no telemetry about telemetry). On
                // failure the receiver drops, senders see the closed channel
                // and discard events (I4 — drops are silent).
                if let Ok(rt) = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                {
                    rt.block_on(run_batch_loop(rx, last_sent_bg, debug, http));
                }
            });
    }

    let inner = Inner {
        sender: Some(tx),
        super_props: cfg.super_props,
        last_sent_unix: (*last_sent).load(Ordering::Relaxed).into(),
        events_captured: AtomicU64::new(0),
        events_dropped: AtomicU64::new(0),
        debug_stderr: cfg.debug_stderr,
        disabled: AtomicBool::new(false),
    };
    TelemetryHandle {
        inner: Arc::new(inner),
    }
}

/// Background loop: batch up to `BATCH_SIZE` events or up to `BATCH_INTERVAL`
/// elapsed, then flush them. When an HTTP client is present the batch is
/// POSTed to PostHog via `network::post_batch`; otherwise it is dropped after
/// the optional stderr echo.
async fn run_batch_loop(
    mut rx: mpsc::Receiver<Msg>,
    last_sent: Arc<AtomicU64>,
    _debug: bool,
    http: Option<reqwest::Client>,
) {
    let mut batch: Vec<QueuedEvent> = Vec::with_capacity(BATCH_SIZE);
    let mut deadline = Instant::now() + BATCH_INTERVAL;

    loop {
        let timeout = deadline.saturating_duration_since(Instant::now());
        tokio::select! {
            maybe_event = rx.recv() => {
                match maybe_event {
                    Some(Msg::Event(event)) => {
                        batch.push(event);
                        if batch.len() >= BATCH_SIZE {
                            flush(&mut batch, &last_sent, http.as_ref()).await;
                            deadline = Instant::now() + BATCH_INTERVAL;
                        }
                    }
                    Some(Msg::Flush(ack)) => {
                        flush(&mut batch, &last_sent, http.as_ref()).await;
                        // After the POST, so an ack means "sent", not "queued".
                        // The receiver is gone if the caller's budget expired
                        // first; that is its business, not ours.
                        let _ = ack.send(());
                        deadline = Instant::now() + BATCH_INTERVAL;
                    }
                    None => {
                        // Sender dropped — final drain and exit.
                        flush(&mut batch, &last_sent, http.as_ref()).await;
                        return;
                    }
                }
            }
            _ = tokio::time::sleep(timeout) => {
                if !batch.is_empty() {
                    flush(&mut batch, &last_sent, http.as_ref()).await;
                }
                deadline = Instant::now() + BATCH_INTERVAL;
            }
        }
    }
}

async fn flush(
    batch: &mut Vec<QueuedEvent>,
    last_sent: &AtomicU64,
    http: Option<&reqwest::Client>,
) {
    if batch.is_empty() {
        return;
    }
    if let Some(client) = http {
        let _ok = super::network::post_batch(client, batch).await;
    }
    // Stamp last-sent regardless of success — `telemetry status` shows
    // attempt timing, and we deliberately do not surface failures (I4).
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or_default();
    last_sent.store(now, Ordering::Relaxed);
    batch.clear();
}

fn emit_debug(event: &QueuedEvent) {
    let envelope = json!({
        "event": event.name,
        "properties": event.properties,
    });
    // eprintln! is appropriate here: this is explicit opt-in developer output
    // gated on `OPENLATCH_TELEMETRY_DEBUG=1`, not operational logging.
    eprintln!("[telemetry/debug] {}", envelope);
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::telemetry::consent::{ConsentState, DecidedBy};

    fn test_props() -> SuperProps {
        SuperProps::new("agt_test".into(), false)
    }

    #[test]
    fn test_disabled_handle_noop_capture() {
        let h = TelemetryHandle::disabled("agt_x".into(), false);
        h.capture(Event::cli_initialized("claude-code", 1, true));
        assert_eq!(h.events_captured(), 0);
        assert!(!h.is_enabled());
    }

    /// A shutdown path must not stall on a handle that has nothing to send.
    #[tokio::test]
    async fn test_flush_on_disabled_handle_returns_immediately() {
        let h = TelemetryHandle::disabled("agt_x".into(), false);
        assert!(h.flush(Duration::from_secs(5)).await);
    }

    /// The regression this whole flush path exists for: an event captured on
    /// the way out must be *sent*, not merely queued. Before `Msg::Flush` the
    /// only drains were the 1s batch timer and the channel closing — and the
    /// global handle's sender never closes, so a `daemon_stopped` captured
    /// immediately before exit was dropped on the floor every time.
    ///
    /// No baked key here, so `flush` drains to the debug/no-network path; what
    /// is asserted is that the batch task acknowledges *after* draining, well
    /// inside a budget far shorter than `BATCH_INTERVAL`. A timer-only drain
    /// could not satisfy this.
    #[tokio::test]
    async fn test_flush_acks_faster_than_the_batch_interval() {
        let cfg = ClientConfig {
            resolved: Resolved {
                state: ConsentState::Enabled,
                decided_by: DecidedBy::ConfigFile,
            },
            super_props: test_props(),
            debug_stderr: false,
            baked_key_present: true,
        };
        let h = start(cfg);
        h.capture(Event::cli_initialized("claude-code", 1, true));
        assert_eq!(h.events_captured(), 1);

        let started = Instant::now();
        assert!(h.flush(Duration::from_secs(5)).await, "flush was not acked");
        assert!(
            started.elapsed() < BATCH_INTERVAL,
            "flush waited for the batch timer ({:?}) instead of draining on demand",
            started.elapsed(),
        );
    }

    /// A second flush after the queue is already empty is still acknowledged —
    /// shutdown paths must be safe to call unconditionally.
    #[tokio::test]
    async fn test_flush_with_empty_buffer_is_acked() {
        let cfg = ClientConfig {
            resolved: Resolved {
                state: ConsentState::Enabled,
                decided_by: DecidedBy::ConfigFile,
            },
            super_props: test_props(),
            debug_stderr: false,
            baked_key_present: true,
        };
        let h = start(cfg);
        assert!(h.flush(Duration::from_secs(5)).await);
        assert!(h.flush(Duration::from_secs(5)).await);
    }

    #[test]
    fn test_start_with_disabled_consent_returns_noop() {
        let cfg = ClientConfig {
            resolved: Resolved {
                state: ConsentState::Disabled,
                decided_by: DecidedBy::ConfigFile,
            },
            super_props: test_props(),
            debug_stderr: false,
            baked_key_present: true,
        };
        let h = start(cfg);
        assert!(!h.is_enabled());
    }

    #[test]
    fn test_start_without_baked_key_returns_noop() {
        let cfg = ClientConfig {
            resolved: Resolved {
                state: ConsentState::Enabled,
                decided_by: DecidedBy::ConfigFile,
            },
            super_props: test_props(),
            debug_stderr: false,
            baked_key_present: false,
        };
        let h = start(cfg);
        assert!(!h.is_enabled());
    }

    #[test]
    fn test_start_without_ambient_runtime_spawns_self_hosted_thread() {
        // Pure sync context — no `#[tokio::test]`. Proves the fallback
        // std::thread + current_thread runtime actually drives the batch
        // loop. Without the fix, the handle would be "enabled" but no
        // receiver would ever drain the channel.
        let cfg = ClientConfig {
            resolved: Resolved {
                state: ConsentState::Enabled,
                decided_by: DecidedBy::ConfigFile,
            },
            super_props: test_props(),
            debug_stderr: false,
            baked_key_present: true,
        };
        let h = start(cfg);
        assert!(h.is_enabled());
        h.capture(Event::cli_initialized("claude-code", 1, true));
        assert_eq!(h.events_captured(), 1);
        // Dropping the handle closes the channel; the batch thread sees
        // `rx.recv() -> None` and exits. We don't join the thread here
        // (it's daemonic by design) but the test completes cleanly, which
        // means the spawned thread didn't panic or deadlock.
        drop(h);
    }

    #[tokio::test]
    async fn test_enabled_handle_accepts_captures() {
        let cfg = ClientConfig {
            resolved: Resolved {
                state: ConsentState::Enabled,
                decided_by: DecidedBy::ConfigFile,
            },
            super_props: test_props(),
            debug_stderr: false,
            baked_key_present: true,
        };
        let h = start(cfg);
        assert!(h.is_enabled());
        h.capture(Event::cli_initialized("claude-code", 1, true));
        // Yield so the batch loop can observe the send.
        tokio::task::yield_now().await;
        assert_eq!(h.events_captured(), 1);
    }
}