errsight 0.1.1

//! The [`Client`]: an in-memory queue drained by a background OS thread.
//!
//! Capture is non-blocking — `enqueue` just pushes onto a channel and returns.
//! A single worker thread batches events, ships them with the configured
//! transport, and handles the awkward parts: bounded backpressure, payload
//! splitting under the 512 KB ingestion limit, non-blocking 429 backoff, and a
//! bounded graceful drain on shutdown. Running on its own thread (rather than
//! an async runtime) is what lets the same client work in sync binaries, CLIs,
//! and async servers without pulling in tokio.

use std::panic::AssertUnwindSafe;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::mpsc::{self, Receiver, Sender, SyncSender};
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use crate::config::Config;
use crate::event::Event;
use crate::transport::{SendOutcome, Transport, UreqTransport};

/// Stay comfortably under the backend's 512 KB hard limit so headers and any
/// server-side overhead never tip a batch over.
const MAX_PAYLOAD_BYTES: usize = 490 * 1024;

/// The backend rejects a batch of more than 100 events with 422. We cap chunks
/// at the send site so an operator who sets a larger `batch_size` (or builds a
/// `Config` directly) can't cause silent, permanent event loss.
const MAX_BATCH_EVENTS: usize = 100;

/// Messages from app threads to the worker.
enum Msg {
    Event(Box<Event>),
    /// Flush now; ack with `true` if the buffer was emptied (everything
    /// delivered), `false` if events remain (e.g. paused by a 429 backoff).
    Flush(SyncSender<bool>),
    /// Drain and exit; ack when done.
    Shutdown(SyncSender<()>),
}

/// The capture/transport client. Held behind an `Arc`; cheap to share.
pub struct Client {
    config: Arc<Config>,
    tx: Sender<Msg>,
    /// Approximate count of events queued-but-not-yet-pulled, for backpressure.
    queue_len: Arc<AtomicUsize>,
    /// Worker handle, taken by `close` to join.
    worker: Mutex<Option<JoinHandle<()>>>,
    closed: AtomicBool,
}

impl Client {
    /// Build a client and spawn its worker thread.
    pub fn new(config: Config) -> Arc<Client> {
        let config = Arc::new(config);

        let transport: Arc<dyn Transport> = match &config.transport {
            Some(t) => t.clone(),
            None => {
                // The default transport carries the API key and PII. Warn loudly
                // (regardless of `debug`) if it would go over cleartext http to a
                // non-local host, unless the operator explicitly opted in.
                if config.is_insecure_remote_host() && !config.allow_insecure_transport {
                    eprintln!(
                        "[errsight] WARNING: sending events (including your API key) over \
                         cleartext HTTP to {} — use https, or call \
                         allow_insecure_transport(true) to silence this warning.",
                        config.host
                    );
                }
                Arc::new(UreqTransport::new(
                    config.events_endpoint(),
                    config.api_key.clone().unwrap_or_default(),
                    config.timeout,
                ))
            }
        };

        let (tx, rx) = mpsc::channel::<Msg>();
        let queue_len = Arc::new(AtomicUsize::new(0));

        let worker = {
            let config = config.clone();
            let queue_len = queue_len.clone();
            std::thread::Builder::new()
                .name("errsight-worker".to_string())
                .spawn(move || worker_loop(rx, config, transport, queue_len))
                .ok()
        };

        if worker.is_none() && config.debug {
            eprintln!("[errsight] failed to spawn worker thread; events will not be delivered");
        }

        Arc::new(Client {
            config,
            tx,
            queue_len,
            worker: Mutex::new(worker),
            closed: AtomicBool::new(false),
        })
    }

    pub fn config(&self) -> &Config {
        &self.config
    }

    /// Run `before_send` (on the caller's thread), then enqueue. Returns the
    /// event's `ingestion_id` if it was queued, `None` if dropped.
    pub fn capture(&self, event: Event) -> Option<String> {
        if self.closed.load(Ordering::Acquire) {
            return None;
        }

        let event = match &self.config.before_send {
            None => event,
            Some(cb) => {
                // Clone so the original survives if the filter panics.
                let cb = cb.clone();
                let probe = event.clone();
                match std::panic::catch_unwind(AssertUnwindSafe(move || cb(probe))) {
                    Ok(Some(filtered)) => filtered,
                    Ok(None) => return None, // intentional drop
                    Err(_) => {
                        // The filter panicked. Default is fail-closed: drop the
                        // event, because before_send is the PII/secret scrubber
                        // and shipping the un-scrubbed original would be a leak.
                        // Opt into fail-open (send unmodified, availability over
                        // confidentiality) with `before_send_fail_open(true)`.
                        if self.config.before_send_fail_open {
                            self.debug_log(
                                "before_send panicked; sending event unmodified (fail-open)",
                            );
                            event
                        } else {
                            self.debug_log("before_send panicked; dropping event (fail-closed)");
                            return None;
                        }
                    }
                }
            }
        };

        let id = event.ingestion_id.clone();
        self.enqueue(event);
        Some(id)
    }

    /// Push an event onto the queue, dropping it (with a debug log) if the
    /// queue is full. Non-blocking.
    fn enqueue(&self, event: Event) {
        if self.queue_len.load(Ordering::Relaxed) >= self.config.max_queue_size {
            self.debug_log(&format!(
                "queue full ({}), dropping event",
                self.config.max_queue_size
            ));
            return;
        }
        self.queue_len.fetch_add(1, Ordering::Relaxed);
        if self.tx.send(Msg::Event(Box::new(event))).is_err() {
            // Worker gone (closed/crashed) — undo the count.
            self.queue_len.fetch_sub(1, Ordering::Relaxed);
        }
    }

    /// Block until queued events are delivered, or `timeout` elapses. Returns
    /// `true` only if everything queued was actually sent — `false` if the
    /// timeout elapsed first, or events remain because the worker is in a 429
    /// backoff window. Default timeout is the configured shutdown timeout.
    pub fn flush(&self, timeout: Option<Duration>) -> bool {
        if self.closed.load(Ordering::Acquire) {
            return false;
        }
        let (ack_tx, ack_rx) = mpsc::sync_channel::<bool>(0);
        if self.tx.send(Msg::Flush(ack_tx)).is_err() {
            return false;
        }
        let to = timeout.unwrap_or(self.config.shutdown_timeout);
        // `Ok(true)` = drained; `Ok(false)` = events remain (paused); timeout = false.
        ack_rx.recv_timeout(to).unwrap_or(false)
    }

    /// Drain the queue and stop the worker. Idempotent; safe to call from a
    /// `Drop` guard and again manually. Bounded by `shutdown_timeout` plus a
    /// small margin so a hung send can't wedge process exit.
    pub fn close(&self, timeout: Option<Duration>) -> bool {
        if self.closed.swap(true, Ordering::AcqRel) {
            return true; // already closed
        }
        let (ack_tx, ack_rx) = mpsc::sync_channel::<()>(0);
        let sent = self.tx.send(Msg::Shutdown(ack_tx)).is_ok();
        let to = timeout.unwrap_or(self.config.shutdown_timeout + Duration::from_secs(1));
        let acked = sent && ack_rx.recv_timeout(to).is_ok();

        // Join only once the worker acked — it returns immediately after. If it
        // didn't ack in time (a hung send), leave it detached rather than
        // blocking process exit on join.
        if let Ok(mut guard) = self.worker.lock() {
            if let Some(handle) = guard.take() {
                if acked {
                    let _ = handle.join();
                }
            }
        }
        acked
    }

    fn debug_log(&self, msg: &str) {
        if self.config.debug {
            eprintln!("[errsight] {msg}");
        }
    }
}

impl Drop for Client {
    fn drop(&mut self) {
        // Last-resort drain if the client is dropped without an explicit close
        // (e.g. the init guard was discarded). `close` is idempotent.
        if !self.closed.load(Ordering::Acquire) {
            self.close(None);
        }
    }
}

// ---------------------------------------------------------------------------
// Worker thread
// ---------------------------------------------------------------------------

fn worker_loop(
    rx: Receiver<Msg>,
    config: Arc<Config>,
    transport: Arc<dyn Transport>,
    queue_len: Arc<AtomicUsize>,
) {
    let mut buffer: Vec<Event> = Vec::new();
    // When set, we're backing off from a 429 until this instant.
    let mut rate_limited_until: Option<Instant> = None;
    // When the oldest buffered event arrived, for partial-batch flush latency.
    let mut oldest_at: Option<Instant> = None;

    loop {
        match rx.recv_timeout(config.flush_interval) {
            Ok(Msg::Event(event)) => {
                queue_len.fetch_sub(1, Ordering::Relaxed);
                buffer.push(*event);
                // Bound the worker's own buffer, not just the channel. While
                // we're backing off from a 429 (or sends are slow) the worker
                // keeps draining the channel — which frees producer slots and
                // lets more events in — so without a cap here the buffer grows
                // without limit. Shed oldest beyond the cap.
                while buffer.len() > config.max_queue_size {
                    buffer.remove(0);
                    if config.debug {
                        eprintln!("[errsight] buffer at capacity; dropping oldest event");
                    }
                }
                if oldest_at.is_none() {
                    oldest_at = Some(Instant::now());
                }
                let batch_full = buffer.len() >= config.batch_size;
                let interval_elapsed = oldest_at
                    .map(|t| t.elapsed() >= config.flush_interval)
                    .unwrap_or(false);
                if batch_full || interval_elapsed {
                    flush_buffer(
                        &mut buffer,
                        &transport,
                        &config,
                        &mut rate_limited_until,
                        None,
                    );
                    if buffer.is_empty() {
                        oldest_at = None;
                    }
                }
            }
            Ok(Msg::Flush(ack)) => {
                flush_buffer(
                    &mut buffer,
                    &transport,
                    &config,
                    &mut rate_limited_until,
                    None,
                );
                if buffer.is_empty() {
                    oldest_at = None;
                }
                // Report honestly: a non-empty buffer means a 429 backoff (or a
                // single oversized event) left events undelivered.
                let _ = ack.send(buffer.is_empty());
            }
            Ok(Msg::Shutdown(ack)) => {
                shutdown_drain(
                    &rx,
                    &mut buffer,
                    &queue_len,
                    &transport,
                    &config,
                    &mut rate_limited_until,
                );
                let _ = ack.send(());
                return;
            }
            Err(mpsc::RecvTimeoutError::Timeout) => {
                // Periodic tick: flush anything buffered and retry after a 429
                // window expires.
                flush_buffer(
                    &mut buffer,
                    &transport,
                    &config,
                    &mut rate_limited_until,
                    None,
                );
                if buffer.is_empty() {
                    oldest_at = None;
                }
            }
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                // All senders dropped without an explicit shutdown. Best-effort
                // final drain, then exit.
                flush_buffer(
                    &mut buffer,
                    &transport,
                    &config,
                    &mut rate_limited_until,
                    None,
                );
                return;
            }
        }
    }
}

/// On shutdown: pull everything still queued, then flush with a hard deadline.
fn shutdown_drain(
    rx: &Receiver<Msg>,
    buffer: &mut Vec<Event>,
    queue_len: &AtomicUsize,
    transport: &Arc<dyn Transport>,
    config: &Config,
    rate_limited_until: &mut Option<Instant>,
) {
    loop {
        match rx.try_recv() {
            Ok(Msg::Event(event)) => {
                queue_len.fetch_sub(1, Ordering::Relaxed);
                buffer.push(*event);
            }
            // Ack any stray control messages so their callers don't block. We
            // report the flush as undelivered (false) here — shutdown_drain
            // hasn't sent yet — though a concurrent flush during shutdown is a
            // corner case.
            Ok(Msg::Flush(ack)) => {
                let _ = ack.send(false);
            }
            Ok(Msg::Shutdown(ack)) => {
                let _ = ack.send(());
            }
            Err(_) => break, // Empty or Disconnected
        }
    }

    // Drain within a hard deadline. If we hit a 429, honour the backoff but
    // only as far as the deadline allows, then retry — a single flush attempt
    // would otherwise drop everything the moment it's rate limited.
    let deadline = Instant::now() + config.shutdown_timeout;
    while !buffer.is_empty() {
        let now = Instant::now();
        if now >= deadline {
            break;
        }
        if let Some(until) = *rate_limited_until {
            if now < until {
                std::thread::sleep((until - now).min(deadline - now));
            }
        }
        flush_buffer(
            buffer,
            transport,
            config,
            rate_limited_until,
            Some(deadline),
        );
        // After a flush, a non-empty buffer means either we're rate limited
        // (loop and wait it out) or the deadline was hit mid-drain (stop).
        if rate_limited_until.is_none() && !buffer.is_empty() {
            break;
        }
    }

    if !buffer.is_empty() && config.debug {
        eprintln!(
            "[errsight] dropping {} undelivered event(s) on shutdown",
            buffer.len()
        );
    }
}

/// Send as much of `buffer` as possible, oldest-first, in chunks that fit both
/// `batch_size` and the payload limit. On a 429, set the backoff window and
/// leave the unsent events in `buffer` for the next tick. On other errors, drop
/// the chunk (the bounded queue is the backpressure mechanism — we don't
/// retry, matching the Ruby SDK). `deadline`, if set, stops sending once
/// passed.
fn flush_buffer(
    buffer: &mut Vec<Event>,
    transport: &Arc<dyn Transport>,
    config: &Config,
    rate_limited_until: &mut Option<Instant>,
    deadline: Option<Instant>,
) {
    // Honour an active 429 backoff.
    if let Some(until) = *rate_limited_until {
        if Instant::now() < until {
            return;
        }
        *rate_limited_until = None;
    }

    while !buffer.is_empty() {
        if let Some(dl) = deadline {
            if Instant::now() > dl {
                break;
            }
        }

        // Largest prefix whose serialized form fits the payload limit, never
        // more than `batch_size` or the backend's 100-event hard cap. A single
        // oversized event is sent anyway and rejected server-side rather than
        // wedging the queue.
        let mut take = buffer.len().min(config.batch_size).min(MAX_BATCH_EVENTS);
        let body = loop {
            let body = serde_json::to_vec(&buffer[0..take]).unwrap_or_else(|_| b"[]".to_vec());
            if body.len() <= MAX_PAYLOAD_BYTES || take == 1 {
                break body;
            }
            take /= 2;
        };

        // A custom Transport that panics must not take down the worker (which
        // would silently stop all delivery). Contain it and treat as an error.
        let outcome = std::panic::catch_unwind(AssertUnwindSafe(|| transport.send(&body)))
            .unwrap_or(SendOutcome::Error("transport panicked".to_string()));

        match outcome {
            SendOutcome::Success => {
                buffer.drain(0..take);
            }
            SendOutcome::RateLimited(d) => {
                *rate_limited_until = Some(Instant::now() + d);
                if config.debug {
                    eprintln!(
                        "[errsight] rate limited; pausing sends for {}s ({} event(s) re-queued)",
                        d.as_secs(),
                        buffer.len()
                    );
                }
                break; // leave events in the buffer for retry
            }
            SendOutcome::Error(reason) => {
                if config.debug {
                    eprintln!("[errsight] dropping {take} event(s): {reason}");
                }
                buffer.drain(0..take);
            }
        }
    }

    // If a 429 re-queue (or slow drain) pushed us over the cap, shed oldest.
    while buffer.len() > config.max_queue_size {
        buffer.remove(0);
        if config.debug {
            eprintln!("[errsight] queue over capacity during backoff; dropping oldest event");
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::ConfigBuilder;
    use crate::level::Level;
    use std::sync::Mutex as StdMutex;

    /// Shared store of the raw bodies a transport was asked to send.
    type SentBodies = Arc<StdMutex<Vec<Vec<u8>>>>;

    /// A transport that records every body it's asked to send and can be told
    /// to 429 the first N attempts.
    struct RecordingTransport {
        bodies: SentBodies,
        rate_limit_first: AtomicUsize,
    }

    impl RecordingTransport {
        fn new() -> (Arc<Self>, SentBodies) {
            let bodies = Arc::new(StdMutex::new(Vec::new()));
            let t = Arc::new(RecordingTransport {
                bodies: bodies.clone(),
                rate_limit_first: AtomicUsize::new(0),
            });
            (t, bodies)
        }
    }

    impl Transport for RecordingTransport {
        fn send(&self, body: &[u8]) -> SendOutcome {
            if self
                .rate_limit_first
                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| {
                    if n > 0 { Some(n - 1) } else { None }
                })
                .is_ok()
            {
                return SendOutcome::RateLimited(Duration::from_millis(50));
            }
            self.bodies.lock().unwrap().push(body.to_vec());
            SendOutcome::Success
        }
    }

    fn test_config(transport: Arc<dyn Transport>) -> Config {
        ConfigBuilder::from_default()
            .api_key("elp_test")
            .min_level(Level::Debug)
            .batch_size(5)
            .flush_interval(Duration::from_millis(20))
            .transport(transport)
            .build()
    }

    fn count_events(bodies: &[Vec<u8>]) -> usize {
        bodies
            .iter()
            .map(|b| serde_json::from_slice::<serde_json::Value>(b).unwrap())
            .map(|v| v.as_array().map(|a| a.len()).unwrap_or(0))
            .sum()
    }

    #[test]
    fn flushes_all_events_on_close() {
        let (transport, bodies) = RecordingTransport::new();
        let client = Client::new(test_config(transport));
        for i in 0..23 {
            client.capture(Event::new(Level::Error, format!("e{i}")));
        }
        assert!(client.close(Some(Duration::from_secs(5))));
        let bodies = bodies.lock().unwrap();
        assert_eq!(count_events(&bodies), 23, "every event must be delivered");
    }

    #[test]
    fn flush_waits_for_delivery() {
        let (transport, bodies) = RecordingTransport::new();
        let client = Client::new(test_config(transport));
        for i in 0..3 {
            client.capture(Event::new(Level::Warning, format!("e{i}")));
        }
        assert!(client.flush(Some(Duration::from_secs(5))));
        assert_eq!(count_events(&bodies.lock().unwrap()), 3);
        client.close(None);
    }

    #[test]
    fn rate_limit_requeues_then_delivers() {
        let (transport, bodies) = RecordingTransport::new();
        transport.rate_limit_first.store(2, Ordering::SeqCst);
        let client = Client::new(test_config(transport));
        for i in 0..4 {
            client.capture(Event::new(Level::Error, format!("e{i}")));
        }
        // Closing clears the backoff and makes a final delivery attempt.
        assert!(client.close(Some(Duration::from_secs(5))));
        assert_eq!(
            count_events(&bodies.lock().unwrap()),
            4,
            "events held during 429 must still be delivered"
        );
    }

    #[test]
    fn queue_full_drops_without_blocking() {
        let (transport, bodies) = RecordingTransport::new();
        // Block all sends so the queue can actually fill.
        transport
            .rate_limit_first
            .store(usize::MAX, Ordering::SeqCst);
        let cfg = ConfigBuilder::from_default()
            .api_key("elp_test")
            .min_level(Level::Debug)
            .max_queue_size(10)
            .flush_interval(Duration::from_millis(10))
            .transport(transport.clone())
            .build();
        let client = Client::new(cfg);
        // Far more than the cap; must not block or panic.
        for i in 0..1000 {
            client.capture(Event::new(Level::Error, format!("e{i}")));
        }
        // Let backoff lapse and allow sends.
        transport.rate_limit_first.store(0, Ordering::SeqCst);
        client.close(Some(Duration::from_secs(5)));
        // We can't assert an exact count, but it must be bounded well below 1000.
        let delivered = count_events(&bodies.lock().unwrap());
        assert!(
            delivered <= 60,
            "queue cap should bound retained events, got {delivered}"
        );
    }

    fn max_body_events(bodies: &[Vec<u8>]) -> usize {
        bodies
            .iter()
            .map(|b| serde_json::from_slice::<serde_json::Value>(b).unwrap())
            .map(|v| v.as_array().map(|a| a.len()).unwrap_or(0))
            .max()
            .unwrap_or(0)
    }

    #[test]
    fn batch_never_exceeds_backend_limit_even_when_misconfigured() {
        let (transport, bodies) = RecordingTransport::new();
        // Construct a Config directly with an over-limit batch_size (pub field,
        // bypassing the builder clamp) to prove the send-site cap holds.
        let mut cfg = test_config(transport);
        cfg.batch_size = 500;
        cfg.max_queue_size = 10_000;
        let client = Client::new(cfg);
        for i in 0..250 {
            client.capture(Event::new(Level::Error, format!("e{i}")));
        }
        assert!(client.close(Some(Duration::from_secs(5))));
        let bodies = bodies.lock().unwrap();
        assert_eq!(count_events(&bodies), 250, "all delivered");
        assert!(
            max_body_events(&bodies) <= 100,
            "no single request may carry more than 100 events, got {}",
            max_body_events(&bodies)
        );
    }

    #[test]
    fn flush_returns_false_while_rate_limited() {
        let (transport, _bodies) = RecordingTransport::new();
        transport
            .rate_limit_first
            .store(usize::MAX, Ordering::SeqCst); // always 429
        let client = Client::new(test_config(transport));
        for i in 0..3 {
            client.capture(Event::new(Level::Error, format!("e{i}")));
        }
        // Events can't be delivered during the backoff, so flush must report false.
        assert!(
            !client.flush(Some(Duration::from_secs(2))),
            "flush must not claim success while events remain buffered"
        );
        client.close(Some(Duration::from_secs(1)));
    }

    struct PanicTransport {
        panic_first: AtomicUsize,
        bodies: SentBodies,
    }
    impl Transport for PanicTransport {
        fn send(&self, body: &[u8]) -> SendOutcome {
            if self
                .panic_first
                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| {
                    if n > 0 { Some(n - 1) } else { None }
                })
                .is_ok()
            {
                panic!("simulated transport panic");
            }
            self.bodies.lock().unwrap().push(body.to_vec());
            SendOutcome::Success
        }
    }

    #[test]
    fn panicking_transport_does_not_kill_worker() {
        let bodies: SentBodies = Arc::new(StdMutex::new(Vec::new()));
        let transport = Arc::new(PanicTransport {
            panic_first: AtomicUsize::new(1), // panic on the first send only
            bodies: bodies.clone(),
        });
        let client = Client::new(test_config(transport));
        // First batch triggers the panic (caught → dropped). Worker must survive.
        for i in 0..5 {
            client.capture(Event::new(Level::Error, format!("a{i}")));
        }
        client.flush(Some(Duration::from_secs(2)));
        // Subsequent events must still be delivered — proof the worker is alive.
        for i in 0..5 {
            client.capture(Event::new(Level::Error, format!("b{i}")));
        }
        assert!(client.close(Some(Duration::from_secs(5))));
        assert!(
            count_events(&bodies.lock().unwrap()) >= 5,
            "worker should keep delivering after a transport panic"
        );
    }

    #[test]
    fn before_send_panic_fails_closed_by_default() {
        let (transport, bodies) = RecordingTransport::new();
        let cfg = ConfigBuilder::from_default()
            .api_key("elp_test")
            .min_level(Level::Debug)
            .flush_interval(Duration::from_millis(10))
            .transport(transport)
            .before_send(|_e| panic!("scrubber bug"))
            .build();
        let client = Client::new(cfg);
        assert!(
            client
                .capture(Event::new(Level::Error, "has-pii"))
                .is_none(),
            "a panicking scrubber must drop the event (fail-closed)"
        );
        client.flush(Some(Duration::from_secs(2)));
        assert_eq!(
            count_events(&bodies.lock().unwrap()),
            0,
            "no unscrubbed event may be delivered"
        );
        client.close(None);
    }

    #[test]
    fn before_send_panic_sends_when_fail_open() {
        let (transport, bodies) = RecordingTransport::new();
        let cfg = ConfigBuilder::from_default()
            .api_key("elp_test")
            .min_level(Level::Debug)
            .flush_interval(Duration::from_millis(10))
            .transport(transport)
            .before_send(|_e| panic!("scrubber bug"))
            .before_send_fail_open(true)
            .build();
        let client = Client::new(cfg);
        assert!(client.capture(Event::new(Level::Error, "boom")).is_some());
        client.flush(Some(Duration::from_secs(2)));
        assert_eq!(count_events(&bodies.lock().unwrap()), 1);
        client.close(None);
    }
}