levelops_sdk/
lib.rs

1use reqwest::Client as HttpClient;
2use serde::Serialize;
3use serde_json::Value;
4use std::collections::HashMap;
5use std::future::Future;
6use std::sync::{Arc, Mutex};
7use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
8use tokio::sync::Notify;
9use tokio::task::JoinHandle;
10use uuid::Uuid;
11
12// ---------------------------------------------------------------------------
13// Public types
14// ---------------------------------------------------------------------------
15
16#[derive(Clone)]
17pub struct Config {
18    pub api_key: String,
19    pub sli_id: String,
20    pub base_url: String,
21    pub batch_size: usize,
22    pub flush_interval: Duration,
23}
24
25impl Default for Config {
26    fn default() -> Self {
27        Config {
28            api_key: String::new(),
29            sli_id: String::new(),
30            base_url: "https://api.levelops.io".into(),
31            batch_size: 100,
32            flush_interval: Duration::from_secs(1),
33        }
34    }
35}
36
37#[derive(Default)]
38pub struct Event {
39    pub event_id: Option<String>,
40    pub timestamp: Option<String>,
41    pub fields: HashMap<String, Value>,
42    pub dimensions: HashMap<String, String>,
43}
44
45pub struct MeasureOpts {
46    pub fields: HashMap<String, Value>,
47    pub dimensions: HashMap<String, String>,
48    pub event_id: Option<String>,
49}
50
51impl MeasureOpts {
52    pub fn new(dims: impl IntoIterator<Item = (impl Into<String>, impl Into<String>)>) -> Self {
53        MeasureOpts {
54            fields: HashMap::new(),
55            dimensions: dims.into_iter().map(|(k, v)| (k.into(), v.into())).collect(),
56            event_id: None,
57        }
58    }
59
60    pub fn event_id(mut self, id: impl Into<String>) -> Self {
61        self.event_id = Some(id.into());
62        self
63    }
64
65    pub fn fields(mut self, f: HashMap<String, Value>) -> Self {
66        self.fields = f;
67        self
68    }
69}
70
71// ---------------------------------------------------------------------------
72// Internal wire format
73// ---------------------------------------------------------------------------
74
75#[derive(Serialize)]
76struct WireEvent<'a> {
77    event_id: &'a str,
78    sli_id: &'a str,
79    timestamp: &'a str,
80    fields: &'a HashMap<String, Value>,
81    dimensions: &'a HashMap<String, String>,
82}
83
84#[derive(Serialize)]
85struct Batch<'a> {
86    events: Vec<WireEvent<'a>>,
87}
88
89struct BufferedEvent {
90    event_id: String,
91    sli_id: String,
92    timestamp: String,
93    fields: HashMap<String, Value>,
94    dimensions: HashMap<String, String>,
95}
96
97// ---------------------------------------------------------------------------
98// Client
99// ---------------------------------------------------------------------------
100
101struct Inner {
102    api_key: String,
103    sli_id: String,
104    base_url: String,
105    batch_size: usize,
106    buffer: Mutex<Vec<BufferedEvent>>,
107    http: HttpClient,
108    notify: Notify,
109}
110
111#[derive(Clone)]
112pub struct Client {
113    inner: Arc<Inner>,
114    _bg: Arc<JoinHandle<()>>,
115}
116
117impl Client {
118    pub fn new(cfg: Config) -> Result<Self, Box<dyn std::error::Error>> {
119        let base_url = cfg.base_url.trim_end_matches('/').to_string();
120        let inner = Arc::new(Inner {
121            api_key: cfg.api_key,
122            sli_id: cfg.sli_id,
123            base_url,
124            batch_size: cfg.batch_size,
125            buffer: Mutex::new(Vec::new()),
126            http: HttpClient::builder().timeout(Duration::from_secs(10)).build()?,
127            notify: Notify::new(),
128        });
129
130        let bg_inner = Arc::clone(&inner);
131        let interval = cfg.flush_interval;
132        let handle = tokio::spawn(async move {
133            loop {
134                tokio::time::sleep(interval).await;
135                flush_inner(&bg_inner).await;
136            }
137        });
138
139        Ok(Client {
140            inner,
141            _bg: Arc::new(handle),
142        })
143    }
144
145    pub async fn record(&self, event: Event) -> Result<(), Box<dyn std::error::Error>> {
146        let event_id = event.event_id.unwrap_or_else(|| Uuid::new_v4().to_string());
147        let timestamp = event.timestamp.unwrap_or_else(utc_now);
148        let dimensions = event.dimensions;
149
150        let ev = BufferedEvent {
151            event_id,
152            sli_id: self.inner.sli_id.clone(),
153            timestamp,
154            fields: event.fields,
155            dimensions,
156        };
157
158        let full = {
159            let mut buf = self.inner.buffer.lock().unwrap();
160            buf.push(ev);
161            buf.len() >= self.inner.batch_size
162        };
163
164        if full {
165            self.flush().await?;
166        }
167        Ok(())
168    }
169
170    /// Measure execution time via an async closure.
171    ///
172    /// ```rust
173    /// client.measure(|ctx| async move {
174    ///     ctx.set_result("ok");
175    ///     Ok(())
176    /// }, MeasureOpts::new([("service", "api")])).await?;
177    /// ```
178    pub async fn measure<F, Fut>(&self, f: F, opts: MeasureOpts) -> Result<(), Box<dyn std::error::Error>>
179    where
180        F: FnOnce(MeasureContext) -> Fut,
181        Fut: Future<Output = Result<(), Box<dyn std::error::Error>>>,
182    {
183        let event_id = opts.event_id.unwrap_or_else(|| Uuid::new_v4().to_string());
184        let ctx = MeasureContext::new();
185        let start = Instant::now();
186
187        let result = f(ctx.clone()).await;
188
189        let latency_ms = start.elapsed().as_millis() as u64;
190        let res = ctx.result();
191        let mut fields = opts.fields;
192        fields.insert("result".into(), Value::String(res));
193        fields.insert("latency_ms".into(), Value::Number(latency_ms.into()));
194        self.record(Event {
195            event_id: Some(event_id),
196            fields,
197            dimensions: opts.dimensions,
198            ..Default::default()
199        })
200        .await?;
201
202        result
203    }
204
205    pub async fn flush(&self) -> Result<(), Box<dyn std::error::Error>> {
206        flush_inner(&self.inner).await;
207        Ok(())
208    }
209}
210
211// ---------------------------------------------------------------------------
212// MeasureContext
213// ---------------------------------------------------------------------------
214
215#[derive(Clone)]
216pub struct MeasureContext {
217    result: Arc<Mutex<String>>,
218}
219
220impl MeasureContext {
221    fn new() -> Self {
222        MeasureContext {
223            result: Arc::new(Mutex::new("ok".into())),
224        }
225    }
226
227    pub fn set_result(&self, r: &str) {
228        *self.result.lock().unwrap() = r.to_string();
229    }
230
231    fn result(&self) -> String {
232        self.result.lock().unwrap().clone()
233    }
234}
235
236// ---------------------------------------------------------------------------
237// Internal helpers
238// ---------------------------------------------------------------------------
239
240async fn flush_inner(inner: &Arc<Inner>) {
241    let batch: Vec<BufferedEvent> = {
242        let mut buf = inner.buffer.lock().unwrap();
243        if buf.is_empty() {
244            return;
245        }
246        std::mem::take(&mut *buf)
247    };
248
249    let wire: Vec<WireEvent> = batch
250        .iter()
251        .map(|e| WireEvent {
252            event_id: &e.event_id,
253            sli_id: &e.sli_id,
254            timestamp: &e.timestamp,
255            fields: &e.fields,
256            dimensions: &e.dimensions,
257        })
258        .collect();
259
260    let payload = match serde_json::to_string(&Batch { events: wire }) {
261        Ok(p) => p,
262        Err(_) => return,
263    };
264
265    for attempt in 0u32..3 {
266        match inner
267            .http
268            .post(format!("{}/events", inner.base_url))
269            .header("Content-Type", "application/json")
270            .header("Authorization", format!("Bearer {}", inner.api_key))
271            .body(payload.clone())
272            .send()
273            .await
274        {
275            Ok(resp) => {
276                if resp.status().as_u16() == 204 {
277                    return;
278                }
279                if resp.status().as_u16() == 429 {
280                    let wait = resp
281                        .headers()
282                        .get("Retry-After")
283                        .and_then(|v| v.to_str().ok())
284                        .and_then(|s| s.parse::<f64>().ok())
285                        .unwrap_or(2f64.powi(attempt as i32));
286                    tokio::time::sleep(Duration::from_secs_f64(wait)).await;
287                    continue;
288                }
289                return; // non-retryable
290            }
291            Err(_) => {
292                if attempt < 2 {
293                    tokio::time::sleep(Duration::from_secs(2u64.pow(attempt))).await;
294                }
295            }
296        }
297    }
298}
299
300fn utc_now() -> String {
301    let now = SystemTime::now()
302        .duration_since(UNIX_EPOCH)
303        .unwrap_or_default();
304    let secs = now.as_secs();
305    let millis = now.subsec_millis();
306    // Format: 2006-01-02T15:04:05.999Z
307    let (y, mo, d, h, mi, s) = epoch_to_parts(secs);
308    format!("{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:03}Z", y, mo, d, h, mi, s, millis)
309}
310
311fn epoch_to_parts(secs: u64) -> (u64, u64, u64, u64, u64, u64) {
312    let s = secs % 60;
313    let m = (secs / 60) % 60;
314    let h = (secs / 3600) % 24;
315    let days = secs / 86400;
316    // Days since 1970-01-01 → Gregorian
317    let (year, month, day) = days_to_ymd(days);
318    (year, month, day, h, m, s)
319}
320
321fn days_to_ymd(mut days: u64) -> (u64, u64, u64) {
322    let mut y = 1970u64;
323    loop {
324        let dy = if is_leap(y) { 366 } else { 365 };
325        if days < dy { break; }
326        days -= dy;
327        y += 1;
328    }
329    let months = if is_leap(y) {
330        [31u64, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
331    } else {
332        [31u64, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
333    };
334    let mut mo = 1u64;
335    for &dm in &months {
336        if days < dm { break; }
337        days -= dm;
338        mo += 1;
339    }
340    (y, mo, days + 1)
341}
342
343fn is_leap(y: u64) -> bool {
344    (y % 4 == 0 && y % 100 != 0) || y % 400 == 0
345}
levelops_sdk/lib.rs

levelops_sdk/
lib.rs
