openlatch-client 0.1.18

//! Pending-alerts storage + long-poll fetcher for the config plane.
//!
//! Cloud caches deny verdicts by `(source_id, content_hash)`. When a verdict
//! is computed AFTER the daemon has already returned an optimistic allow to
//! a SessionStart, the cloud queues a pending alert. The daemon long-polls
//! `GET /api/v1/alerts/pending` piggybacked on the existing
//! `cloud_state.next_health_delay` schedule (5–60 s adaptive) — no new
//! state machine, no new dep.

use std::collections::VecDeque;
use std::sync::Arc;
use std::time::Duration;

use dashmap::DashMap;
use secrecy::ExposeSecret;
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;

use crate::cloud::{CloudState, CredentialProvider};

const MAX_ALERTS_PER_SESSION: usize = 16;
const MAX_SESSIONS: usize = 1024;
/// Minimum sleep between long-poll attempts. Mirrors the `cloud_state`
/// degraded-mode floor so a healthy daemon polls at most once a minute.
const MIN_POLL_DELAY: Duration = Duration::from_secs(5);

/// One pending alert produced by the cloud's deep-analysis worker. Surfaced
/// to the user via the next outbound hook response (translator injection)
/// and cleared by `openlatch inventory ack`.
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct PendingAlert {
    pub alert_id: String,
    /// Session identifier the alert is bound to. Matches the CloudEvent
    /// `subject` on outbound hook envelopes.
    pub session_ref_id: String,
    /// One of `critical` / `high` / `medium` / `low` / `info`.
    pub severity: String,
    /// ≤120 chars — used as the alert headline in translator output.
    pub headline: String,
    /// ≤500 chars — body / explanation surfaced to the user.
    pub body: String,
    /// `{agent}:{kind}:{path_hash}` — opaque to the client; cloud resolves it.
    pub source_id: String,
    /// RFC 3339.
    pub created_at: String,
    /// Set true by `pop_for_session` once the translator has surfaced it.
    /// Persists in memory until ack so re-queries don't double-deliver.
    #[serde(default)]
    pub delivered: bool,
}

/// Bounded LRU pending-alert cache. `DashMap` for concurrent reads on the
/// hot path; a single `Mutex<VecDeque>` tracks LRU recency for eviction.
pub struct PendingAlerts {
    inner: DashMap<String, VecDeque<PendingAlert>>,
    lru: Mutex<VecDeque<String>>,
}

impl PendingAlerts {
    pub fn new() -> Self {
        Self {
            inner: DashMap::new(),
            lru: Mutex::new(VecDeque::with_capacity(MAX_SESSIONS)),
        }
    }

    /// Push an alert for its session. Drops the oldest entry if the
    /// per-session ring buffer is full; evicts the oldest session if the
    /// global session count exceeds `MAX_SESSIONS`.
    pub async fn push(&self, alert: PendingAlert) {
        let session_id = alert.session_ref_id.clone();
        {
            let mut entry = self.inner.entry(session_id.clone()).or_default();
            if entry.len() >= MAX_ALERTS_PER_SESSION {
                entry.pop_front();
            }
            entry.push_back(alert);
        }

        let mut lru = self.lru.lock().await;
        lru.retain(|s| s != &session_id);
        lru.push_back(session_id);
        while lru.len() > MAX_SESSIONS {
            if let Some(evicted) = lru.pop_front() {
                self.inner.remove(&evicted);
            }
        }
    }

    /// Take the next undelivered alert for `session_ref_id` and mark it
    /// delivered (so subsequent calls return the next one). Returns `None`
    /// when nothing is pending.
    pub fn pop_for_session(&self, session_ref_id: &str) -> Option<PendingAlert> {
        let mut entry = self.inner.get_mut(session_ref_id)?;
        entry.iter_mut().find(|a| !a.delivered).map(|a| {
            a.delivered = true;
            a.clone()
        })
    }

    /// Acknowledge by `alert_id`. Returns `true` if an entry was removed.
    /// When `alert_id` is `None`, clears every alert.
    pub fn ack(&self, alert_id: Option<&str>) -> usize {
        let mut count = 0usize;
        match alert_id {
            None => {
                for mut entry in self.inner.iter_mut() {
                    count += entry.len();
                    entry.clear();
                }
            }
            Some(id) => {
                for mut entry in self.inner.iter_mut() {
                    let before = entry.len();
                    entry.retain(|a| a.alert_id != id);
                    count += before - entry.len();
                }
            }
        }
        count
    }

    /// Total pending alerts across every session — exposed via
    /// `/admin/inventory/status`.
    pub fn pending_count(&self) -> usize {
        self.inner.iter().map(|e| e.len()).sum()
    }

    /// Fast path for the hot ingest loop — `true` when no alerts are
    /// queued for any session, so the handler can skip the per-event
    /// `pop_for_session` lookup entirely.
    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }

    /// Snapshot every pending alert (delivered + undelivered). Used by the
    /// `openlatch inventory inspect` admin path so users can see the queue
    /// before acking.
    pub fn snapshot(&self) -> Vec<PendingAlert> {
        let mut out = Vec::new();
        for entry in self.inner.iter() {
            out.extend(entry.iter().cloned());
        }
        out
    }
}

impl Default for PendingAlerts {
    fn default() -> Self {
        Self::new()
    }
}

/// Long-poll fetcher. Loops for the daemon's lifetime — the tokio runtime
/// drops the task on shutdown like every other detached worker. Failures
/// are logged at `debug!` and the loop continues; the cloud can be down
/// without disrupting the data plane.
pub async fn run_long_poll(
    pending: Arc<PendingAlerts>,
    cloud_state: CloudState,
    credentials: Arc<dyn CredentialProvider>,
    api_url: String,
    machine_id: String,
    http_client: reqwest::Client,
) {
    let url = format!("{}/api/v1/alerts/pending", api_url.trim_end_matches('/'));
    loop {
        let delay = std::cmp::max(cloud_state.next_health_delay(), MIN_POLL_DELAY);
        tokio::time::sleep(delay).await;

        let Some(token) = credentials.retrieve() else {
            continue;
        };
        let resp = http_client
            .get(&url)
            .bearer_auth(token.expose_secret())
            .header("X-OpenLatch-Machine-Id", &machine_id)
            .send()
            .await;
        match resp {
            Ok(r) if r.status().is_success() => match r.json::<Vec<PendingAlert>>().await {
                Ok(alerts) => {
                    for alert in alerts {
                        crate::telemetry::capture_global(
                            crate::telemetry::Event::config_pending_alert_received(
                                &alert.severity,
                                "claude-code",
                            ),
                        );
                        pending.push(alert).await;
                    }
                }
                Err(e) => {
                    tracing::debug!(
                        target: "alerts",
                        error = %e,
                        "pending-alerts response parse failed"
                    );
                }
            },
            Ok(r) => {
                tracing::debug!(
                    target: "alerts",
                    status = %r.status(),
                    "pending-alerts long-poll non-success"
                );
            }
            Err(e) => {
                tracing::debug!(
                    target: "alerts",
                    error = %e,
                    "pending-alerts long-poll failed"
                );
            }
        }
    }
}

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

    fn alert(session: &str, alert_id: &str, severity: &str) -> PendingAlert {
        PendingAlert {
            alert_id: alert_id.into(),
            session_ref_id: session.into(),
            severity: severity.into(),
            headline: "Configuration alert".into(),
            body: "An MCP server was added that needs review.".into(),
            source_id: "claude-code:mcp:abc123".into(),
            created_at: "2026-05-07T12:00:00Z".into(),
            delivered: false,
        }
    }

    #[tokio::test]
    async fn push_and_pop_returns_undelivered_then_marks_delivered() {
        let p = PendingAlerts::new();
        p.push(alert("sess_a", "alert_1", "high")).await;
        let got = p.pop_for_session("sess_a").unwrap();
        assert_eq!(got.alert_id, "alert_1");
        assert!(got.delivered);
        // Already delivered: next pop yields None.
        assert!(p.pop_for_session("sess_a").is_none());
    }

    #[tokio::test]
    async fn per_session_ring_buffer_drops_oldest() {
        let p = PendingAlerts::new();
        for i in 0..(MAX_ALERTS_PER_SESSION + 3) {
            p.push(alert("sess_a", &format!("a_{i}"), "low")).await;
        }
        let count: usize = p
            .snapshot()
            .iter()
            .filter(|a| a.session_ref_id == "sess_a")
            .count();
        assert_eq!(count, MAX_ALERTS_PER_SESSION);
        // Oldest 3 should have been evicted.
        let ids: Vec<String> = p.snapshot().into_iter().map(|a| a.alert_id).collect();
        assert!(!ids.contains(&"a_0".to_string()));
        assert!(ids.contains(&"a_3".to_string()));
    }

    #[tokio::test]
    async fn ack_by_id_removes_only_that_alert() {
        let p = PendingAlerts::new();
        p.push(alert("sess_a", "a", "low")).await;
        p.push(alert("sess_a", "b", "low")).await;
        let removed = p.ack(Some("a"));
        assert_eq!(removed, 1);
        let remaining: Vec<String> = p.snapshot().into_iter().map(|x| x.alert_id).collect();
        assert_eq!(remaining, vec!["b".to_string()]);
    }

    #[tokio::test]
    async fn ack_all_clears_everything() {
        let p = PendingAlerts::new();
        p.push(alert("sess_a", "a", "low")).await;
        p.push(alert("sess_b", "b", "low")).await;
        let removed = p.ack(None);
        assert_eq!(removed, 2);
        assert_eq!(p.pending_count(), 0);
    }

    #[tokio::test]
    async fn pending_count_sums_across_sessions() {
        let p = PendingAlerts::new();
        p.push(alert("sess_a", "a", "low")).await;
        p.push(alert("sess_b", "b", "low")).await;
        p.push(alert("sess_b", "c", "low")).await;
        assert_eq!(p.pending_count(), 3);
    }
}