levelops-sdk 0.1.0

use reqwest::Client as HttpClient;
use serde::Serialize;
use serde_json::Value;
use std::collections::HashMap;
use std::future::Future;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::Notify;
use tokio::task::JoinHandle;
use uuid::Uuid;

// ---------------------------------------------------------------------------
// Public types
// ---------------------------------------------------------------------------

#[derive(Clone)]
pub struct Config {
    pub api_key: String,
    pub sli_id: String,
    pub base_url: String,
    pub batch_size: usize,
    pub flush_interval: Duration,
}

impl Default for Config {
    fn default() -> Self {
        Config {
            api_key: String::new(),
            sli_id: String::new(),
            base_url: "https://api.levelops.io".into(),
            batch_size: 100,
            flush_interval: Duration::from_secs(1),
        }
    }
}

#[derive(Default)]
pub struct Event {
    pub event_id: Option<String>,
    pub timestamp: Option<String>,
    pub fields: HashMap<String, Value>,
    pub dimensions: HashMap<String, String>,
}

pub struct MeasureOpts {
    pub fields: HashMap<String, Value>,
    pub dimensions: HashMap<String, String>,
    pub event_id: Option<String>,
}

impl MeasureOpts {
    pub fn new(dims: impl IntoIterator<Item = (impl Into<String>, impl Into<String>)>) -> Self {
        MeasureOpts {
            fields: HashMap::new(),
            dimensions: dims.into_iter().map(|(k, v)| (k.into(), v.into())).collect(),
            event_id: None,
        }
    }

    pub fn event_id(mut self, id: impl Into<String>) -> Self {
        self.event_id = Some(id.into());
        self
    }

    pub fn fields(mut self, f: HashMap<String, Value>) -> Self {
        self.fields = f;
        self
    }
}

// ---------------------------------------------------------------------------
// Internal wire format
// ---------------------------------------------------------------------------

#[derive(Serialize)]
struct WireEvent<'a> {
    event_id: &'a str,
    sli_id: &'a str,
    timestamp: &'a str,
    fields: &'a HashMap<String, Value>,
    dimensions: &'a HashMap<String, String>,
}

#[derive(Serialize)]
struct Batch<'a> {
    events: Vec<WireEvent<'a>>,
}

struct BufferedEvent {
    event_id: String,
    sli_id: String,
    timestamp: String,
    fields: HashMap<String, Value>,
    dimensions: HashMap<String, String>,
}

// ---------------------------------------------------------------------------
// Client
// ---------------------------------------------------------------------------

struct Inner {
    api_key: String,
    sli_id: String,
    base_url: String,
    batch_size: usize,
    buffer: Mutex<Vec<BufferedEvent>>,
    http: HttpClient,
    notify: Notify,
}

#[derive(Clone)]
pub struct Client {
    inner: Arc<Inner>,
    _bg: Arc<JoinHandle<()>>,
}

impl Client {
    pub fn new(cfg: Config) -> Result<Self, Box<dyn std::error::Error>> {
        let base_url = cfg.base_url.trim_end_matches('/').to_string();
        let inner = Arc::new(Inner {
            api_key: cfg.api_key,
            sli_id: cfg.sli_id,
            base_url,
            batch_size: cfg.batch_size,
            buffer: Mutex::new(Vec::new()),
            http: HttpClient::builder().timeout(Duration::from_secs(10)).build()?,
            notify: Notify::new(),
        });

        let bg_inner = Arc::clone(&inner);
        let interval = cfg.flush_interval;
        let handle = tokio::spawn(async move {
            loop {
                tokio::time::sleep(interval).await;
                flush_inner(&bg_inner).await;
            }
        });

        Ok(Client {
            inner,
            _bg: Arc::new(handle),
        })
    }

    pub async fn record(&self, event: Event) -> Result<(), Box<dyn std::error::Error>> {
        let event_id = event.event_id.unwrap_or_else(|| Uuid::new_v4().to_string());
        let timestamp = event.timestamp.unwrap_or_else(utc_now);
        let dimensions = event.dimensions;

        let ev = BufferedEvent {
            event_id,
            sli_id: self.inner.sli_id.clone(),
            timestamp,
            fields: event.fields,
            dimensions,
        };

        let full = {
            let mut buf = self.inner.buffer.lock().unwrap();
            buf.push(ev);
            buf.len() >= self.inner.batch_size
        };

        if full {
            self.flush().await?;
        }
        Ok(())
    }

    /// Measure execution time via an async closure.
    ///
    /// ```rust
    /// client.measure(|ctx| async move {
    ///     ctx.set_result("ok");
    ///     Ok(())
    /// }, MeasureOpts::new([("service", "api")])).await?;
    /// ```
    pub async fn measure<F, Fut>(&self, f: F, opts: MeasureOpts) -> Result<(), Box<dyn std::error::Error>>
    where
        F: FnOnce(MeasureContext) -> Fut,
        Fut: Future<Output = Result<(), Box<dyn std::error::Error>>>,
    {
        let event_id = opts.event_id.unwrap_or_else(|| Uuid::new_v4().to_string());
        let ctx = MeasureContext::new();
        let start = Instant::now();

        let result = f(ctx.clone()).await;

        let latency_ms = start.elapsed().as_millis() as u64;
        let res = ctx.result();
        let mut fields = opts.fields;
        fields.insert("result".into(), Value::String(res));
        fields.insert("latency_ms".into(), Value::Number(latency_ms.into()));
        self.record(Event {
            event_id: Some(event_id),
            fields,
            dimensions: opts.dimensions,
            ..Default::default()
        })
        .await?;

        result
    }

    pub async fn flush(&self) -> Result<(), Box<dyn std::error::Error>> {
        flush_inner(&self.inner).await;
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// MeasureContext
// ---------------------------------------------------------------------------

#[derive(Clone)]
pub struct MeasureContext {
    result: Arc<Mutex<String>>,
}

impl MeasureContext {
    fn new() -> Self {
        MeasureContext {
            result: Arc::new(Mutex::new("ok".into())),
        }
    }

    pub fn set_result(&self, r: &str) {
        *self.result.lock().unwrap() = r.to_string();
    }

    fn result(&self) -> String {
        self.result.lock().unwrap().clone()
    }
}

// ---------------------------------------------------------------------------
// Internal helpers
// ---------------------------------------------------------------------------

async fn flush_inner(inner: &Arc<Inner>) {
    let batch: Vec<BufferedEvent> = {
        let mut buf = inner.buffer.lock().unwrap();
        if buf.is_empty() {
            return;
        }
        std::mem::take(&mut *buf)
    };

    let wire: Vec<WireEvent> = batch
        .iter()
        .map(|e| WireEvent {
            event_id: &e.event_id,
            sli_id: &e.sli_id,
            timestamp: &e.timestamp,
            fields: &e.fields,
            dimensions: &e.dimensions,
        })
        .collect();

    let payload = match serde_json::to_string(&Batch { events: wire }) {
        Ok(p) => p,
        Err(_) => return,
    };

    for attempt in 0u32..3 {
        match inner
            .http
            .post(format!("{}/events", inner.base_url))
            .header("Content-Type", "application/json")
            .header("Authorization", format!("Bearer {}", inner.api_key))
            .body(payload.clone())
            .send()
            .await
        {
            Ok(resp) => {
                if resp.status().as_u16() == 204 {
                    return;
                }
                if resp.status().as_u16() == 429 {
                    let wait = resp
                        .headers()
                        .get("Retry-After")
                        .and_then(|v| v.to_str().ok())
                        .and_then(|s| s.parse::<f64>().ok())
                        .unwrap_or(2f64.powi(attempt as i32));
                    tokio::time::sleep(Duration::from_secs_f64(wait)).await;
                    continue;
                }
                return; // non-retryable
            }
            Err(_) => {
                if attempt < 2 {
                    tokio::time::sleep(Duration::from_secs(2u64.pow(attempt))).await;
                }
            }
        }
    }
}

fn utc_now() -> String {
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default();
    let secs = now.as_secs();
    let millis = now.subsec_millis();
    // Format: 2006-01-02T15:04:05.999Z
    let (y, mo, d, h, mi, s) = epoch_to_parts(secs);
    format!("{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:03}Z", y, mo, d, h, mi, s, millis)
}

fn epoch_to_parts(secs: u64) -> (u64, u64, u64, u64, u64, u64) {
    let s = secs % 60;
    let m = (secs / 60) % 60;
    let h = (secs / 3600) % 24;
    let days = secs / 86400;
    // Days since 1970-01-01 → Gregorian
    let (year, month, day) = days_to_ymd(days);
    (year, month, day, h, m, s)
}

fn days_to_ymd(mut days: u64) -> (u64, u64, u64) {
    let mut y = 1970u64;
    loop {
        let dy = if is_leap(y) { 366 } else { 365 };
        if days < dy { break; }
        days -= dy;
        y += 1;
    }
    let months = if is_leap(y) {
        [31u64, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    } else {
        [31u64, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    };
    let mut mo = 1u64;
    for &dm in &months {
        if days < dm { break; }
        days -= dm;
        mo += 1;
    }
    (y, mo, days + 1)
}

fn is_leap(y: u64) -> bool {
    (y % 4 == 0 && y % 100 != 0) || y % 400 == 0
}