billdogeng 1.0.0-beta.1

//! Analytics client: capture/identify/groupIdentify/alias + batching.
//!
//! Events accumulate in an in-memory queue and are flushed as a single batched
//! POST to `/ingest-events` when:
//!   - the queue reaches `flush_at`,
//!   - the background timer fires every `flush_interval_ms`, or
//!   - `flush()` / `shutdown()` is called.
//!
//! A failed flush re-queues its batch (front of the queue) so events are never
//! lost on a single transient failure; the [`crate::transport::Transport`] also
//! retries with backoff before a flush is considered failed.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use serde_json::{json, Map, Value};

use crate::error::Result;
use crate::transport::{RequestOptions, Transport};
use crate::types::{Groups, Properties, WireEvent};

struct Shared {
    queue: Mutex<Vec<WireEvent>>,
    /// Signals the background thread to wake (new flush due / shutdown).
    cvar: Condvar,
    /// Serializes flushes so re-queued batches don't interleave.
    flush_lock: Mutex<()>,
    shutdown: AtomicBool,
}

/// Analytics client.
pub struct Analytics {
    transport: Transport,
    api_key: String,
    flush_at: usize,
    max_queue_size: usize,
    enable_logging: bool,
    group_type_index: Option<std::collections::HashMap<String, u8>>,
    shared: Arc<Shared>,
    bg: Mutex<Option<JoinHandle<()>>>,
}

impl Analytics {
    /// Construct the analytics client and start its background flush thread.
    pub fn new(
        transport: Transport,
        api_key: String,
        flush_at: usize,
        flush_interval_ms: u64,
        max_queue_size: usize,
        enable_logging: bool,
        group_type_index: Option<std::collections::HashMap<String, u8>>,
    ) -> Self {
        let shared = Arc::new(Shared {
            queue: Mutex::new(Vec::new()),
            cvar: Condvar::new(),
            flush_lock: Mutex::new(()),
            shutdown: AtomicBool::new(false),
        });

        let bg = Self::start_timer(
            transport.clone(),
            api_key.clone(),
            flush_interval_ms,
            enable_logging,
            shared.clone(),
        );

        Self {
            transport,
            api_key,
            flush_at,
            max_queue_size,
            enable_logging,
            group_type_index,
            shared,
            bg: Mutex::new(Some(bg)),
        }
    }

    // ─── Public API ───────────────────────────────────────────────────────

    /// Capture an arbitrary event for a user. Batched; flushed per policy.
    pub fn capture(&self, distinct_id: &str, event: &str, properties: Properties, groups: Groups) {
        let props = self.with_groups(properties, &groups);
        self.enqueue(distinct_id, event, props);
    }

    /// Set person properties. Emits `$identify`.
    pub fn identify(&self, distinct_id: &str, properties: Properties) {
        let mut props: Map<String, Value> = properties.clone();
        props.insert("$set".to_string(), Value::Object(properties));
        self.enqueue(distinct_id, "$identify", props);
    }

    /// Set properties on a group. Emits `$groupidentify`.
    pub fn group_identify(&self, group_type: &str, group_key: &str, properties: Properties) {
        let mut props = Map::new();
        props.insert("$group_type".to_string(), json!(group_type));
        props.insert("$group_key".to_string(), json!(group_key));
        props.insert("$group_set".to_string(), Value::Object(properties));
        self.enqueue(group_key, "$groupidentify", props);
    }

    /// Alias one distinct id to another. Emits `$create_alias`.
    pub fn alias(&self, distinct_id: &str, alias: &str) {
        let mut props = Map::new();
        props.insert("alias".to_string(), json!(alias));
        props.insert("distinct_id".to_string(), json!(distinct_id));
        self.enqueue(distinct_id, "$create_alias", props);
    }

    /// Force-flush the queue immediately. Resolves when the in-flight batch settles.
    pub fn flush(&self) -> Result<()> {
        Self::do_flush(
            &self.transport,
            &self.api_key,
            self.enable_logging,
            &self.shared,
        )
    }

    /// Current number of queued (not-yet-flushed) events.
    pub fn queue_length(&self) -> usize {
        self.shared.queue.lock().unwrap().len()
    }

    /// Stop the background timer and flush remaining events.
    pub fn shutdown(&self) -> Result<()> {
        self.shared.shutdown.store(true, Ordering::SeqCst);
        self.shared.cvar.notify_all();
        if let Some(handle) = self.bg.lock().unwrap().take() {
            let _ = handle.join();
        }
        self.flush()
    }

    // ─── Internals ──────────────────────────────────────────────────────────

    fn start_timer(
        transport: Transport,
        api_key: String,
        flush_interval_ms: u64,
        enable_logging: bool,
        shared: Arc<Shared>,
    ) -> JoinHandle<()> {
        thread::spawn(move || {
            let interval = Duration::from_millis(flush_interval_ms);
            loop {
                let guard = shared.queue.lock().unwrap();
                let (_g, _timeout) = shared
                    .cvar
                    .wait_timeout(guard, interval)
                    .unwrap();
                drop(_g);
                if shared.shutdown.load(Ordering::SeqCst) {
                    break;
                }
                let _ = Self::do_flush(&transport, &api_key, enable_logging, &shared);
            }
        })
    }

    fn with_groups(&self, properties: Properties, groups: &Groups) -> Properties {
        if groups.is_empty() {
            return properties;
        }
        let mut out = properties;
        let mut group_map = Map::new();
        for (k, v) in groups {
            group_map.insert(k.clone(), json!(v));
        }
        out.insert("$groups".to_string(), Value::Object(group_map));
        if let Some(index_map) = &self.group_type_index {
            for (gtype, gkey) in groups {
                if let Some(&idx) = index_map.get(gtype) {
                    if idx <= 4 {
                        out.insert(format!("$group_{idx}"), json!(gkey));
                    }
                }
            }
        }
        out
    }

    fn enqueue(&self, distinct_id: &str, event_name: &str, properties: Properties) {
        let wire = WireEvent {
            event_name: event_name.to_string(),
            event_timestamp: now_ms(),
            properties,
            user_id: distinct_id.to_string(),
        };

        let should_flush = {
            let mut q = self.shared.queue.lock().unwrap();
            q.push(wire);
            if q.len() > self.max_queue_size {
                let dropped = q.len() - self.max_queue_size;
                q.drain(0..dropped);
                if self.enable_logging {
                    eprintln!("[BilldogEng:analytics] max queue size exceeded, dropped {dropped} oldest event(s)");
                }
            }
            q.len() >= self.flush_at
        };

        if should_flush {
            let _ = self.flush();
        }
    }

    fn do_flush(
        transport: &Transport,
        api_key: &str,
        enable_logging: bool,
        shared: &Arc<Shared>,
    ) -> Result<()> {
        // Serialize flushes.
        let _serialize = shared.flush_lock.lock().unwrap();

        let batch = {
            let mut q = shared.queue.lock().unwrap();
            if q.is_empty() {
                return Ok(());
            }
            std::mem::take(&mut *q)
        };

        let body = json!({
            "api_key": api_key,
            "events": batch,
        });

        let opts = RequestOptions::post("/ingest-events", body)
            .header("x-api-key", api_key);

        match transport.request(&opts) {
            Ok(_) => {
                if enable_logging {
                    eprintln!("[BilldogEng:analytics] flushed {} event(s)", batch.len());
                }
                Ok(())
            }
            Err(e) => {
                // Re-queue the batch at the front so it is retried; never lose events.
                let mut q = shared.queue.lock().unwrap();
                let mut requeued = batch;
                requeued.append(&mut *q);
                *q = requeued;
                if enable_logging {
                    eprintln!(
                        "[BilldogEng:analytics] flush failed, re-queued event(s): {}",
                        e.message
                    );
                }
                Err(e)
            }
        }
    }
}

fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}