codesynapse_core/
telemetry.rs

1use std::collections::HashMap;
2use std::io::Write as IoWrite;
3use std::path::{Path, PathBuf};
4use std::sync::Mutex;
5use std::time::{SystemTime, UNIX_EPOCH};
6
7pub const ENDPOINT: &str = "https://codesynapse-telemetry.sohilladhani.workers.dev/v1/events";
8const SCHEMA_VERSION: u32 = 1;
9const MAX_QUEUE_BYTES: u64 = 256 * 1024;
10
11// ── config ───────────────────────────────────────────────────────────────────
12
13#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
14pub struct TelemetryConfig {
15    pub enabled: bool,
16    pub machine_id: String,
17    pub consent_source: String,
18    pub updated_at: String,
19}
20
21#[derive(Debug)]
22pub struct TelemetryStatus {
23    pub enabled: bool,
24    pub decided_by: &'static str,
25    pub machine_id: Option<String>,
26    pub config_path: PathBuf,
27}
28
29// ── in-memory aggregation ────────────────────────────────────────────────────
30
31#[derive(Default)]
32struct CountEntry {
33    calls: u64,
34    errors: u64,
35    saved_chars: u64,
36}
37
38#[derive(Default)]
39struct Inner {
40    // key: (utc_day, tool_name)
41    counts: HashMap<(String, String), CountEntry>,
42    lifecycle: Vec<serde_json::Value>,
43}
44
45// ── public handle ────────────────────────────────────────────────────────────
46
47pub struct Telemetry {
48    global_dir: PathBuf,
49    inner: Mutex<Inner>,
50}
51
52impl Telemetry {
53    pub fn new(global_dir: PathBuf) -> Self {
54        Self {
55            global_dir,
56            inner: Mutex::new(Inner::default()),
57        }
58    }
59
60    // ── consent ──────────────────────────────────────────────────────────────
61
62    pub fn status(&self) -> TelemetryStatus {
63        let config = self.read_config();
64        let machine_id = config.as_ref().map(|c| c.machine_id.clone());
65        let config_path = self.config_path();
66
67        if env_flag("DO_NOT_TRACK") {
68            return TelemetryStatus {
69                enabled: false,
70                decided_by: "DO_NOT_TRACK",
71                machine_id,
72                config_path,
73            };
74        }
75        if let Ok(v) = std::env::var("CODESYNAPSE_TELEMETRY") {
76            let on = !matches!(v.as_str(), "0" | "false" | "off");
77            return TelemetryStatus {
78                enabled: on,
79                decided_by: "CODESYNAPSE_TELEMETRY",
80                machine_id,
81                config_path,
82            };
83        }
84        match config {
85            Some(c) => TelemetryStatus {
86                enabled: c.enabled,
87                decided_by: "config",
88                machine_id: Some(c.machine_id),
89                config_path,
90            },
91            None => TelemetryStatus {
92                enabled: false,
93                decided_by: "default",
94                machine_id: None,
95                config_path,
96            },
97        }
98    }
99
100    pub fn is_enabled(&self) -> bool {
101        self.status().enabled
102    }
103
104    pub fn set_enabled(&self, enabled: bool) {
105        let existing = self.read_config();
106        let machine_id = existing
107            .as_ref()
108            .map(|c| c.machine_id.clone())
109            .unwrap_or_else(new_uuid);
110        let config = TelemetryConfig {
111            enabled,
112            machine_id,
113            consent_source: "cli".into(),
114            updated_at: utc_now_iso(),
115        };
116        self.write_config(&config);
117        if !enabled {
118            let _ = std::fs::remove_file(self.queue_path());
119        }
120    }
121
122    // ── recording (hot path — just in-memory) ────────────────────────────────
123
124    pub fn record_usage(&self, tool: &str, saved_chars: usize, ok: bool) {
125        if !self.is_enabled() {
126            return;
127        }
128        let day = utc_day();
129        let mut inner = self.inner.lock().unwrap();
130        let entry = inner.counts.entry((day, tool.to_string())).or_default();
131        entry.calls += 1;
132        if !ok {
133            entry.errors += 1;
134        }
135        entry.saved_chars += saved_chars as u64;
136    }
137
138    pub fn record_lifecycle(&self, event: &str, props: serde_json::Value) {
139        if !self.is_enabled() {
140            return;
141        }
142        let line = serde_json::json!({
143            "type": "lifecycle",
144            "event": event,
145            "ts": utc_now_iso(),
146            "props": props,
147        });
148        self.inner.lock().unwrap().lifecycle.push(line);
149    }
150
151    // ── persist (sync, call at process exit) ────────────────────────────────
152
153    pub fn persist_sync(&self) {
154        if !self.is_enabled() {
155            return;
156        }
157        let mut inner = self.inner.lock().unwrap();
158        if inner.counts.is_empty() && inner.lifecycle.is_empty() {
159            return;
160        }
161
162        let mut lines: Vec<serde_json::Value> = Vec::new();
163
164        for ((day, tool), entry) in inner.counts.drain() {
165            lines.push(serde_json::json!({
166                "type": "rollup",
167                "day": day,
168                "tool": tool,
169                "calls": entry.calls,
170                "errors": entry.errors,
171                "saved_chars": entry.saved_chars,
172            }));
173        }
174        for ev in inner.lifecycle.drain(..) {
175            lines.push(ev);
176        }
177
178        let _ = self.append_to_queue(&lines);
179    }
180
181    // ── flush (background thread, fire-and-forget) ───────────────────────────
182
183    pub fn flush_bg(&self) {
184        if !self.is_enabled() {
185            return;
186        }
187        let config = match self.read_config() {
188            Some(c) if c.enabled => c,
189            _ => return,
190        };
191        let queue_path = self.queue_path();
192        if !queue_path.exists() {
193            return;
194        }
195
196        let endpoint = std::env::var("CODESYNAPSE_TELEMETRY_ENDPOINT")
197            .unwrap_or_else(|_| ENDPOINT.to_string());
198
199        std::thread::spawn(move || {
200            let _ = do_flush(&queue_path, &config.machine_id, &endpoint);
201        });
202    }
203
204    // ── paths ────────────────────────────────────────────────────────────────
205
206    fn config_path(&self) -> PathBuf {
207        self.global_dir.join("telemetry.json")
208    }
209
210    fn queue_path(&self) -> PathBuf {
211        self.global_dir.join("telemetry-queue.jsonl")
212    }
213
214    fn read_config(&self) -> Option<TelemetryConfig> {
215        let text = std::fs::read_to_string(self.config_path()).ok()?;
216        serde_json::from_str(&text).ok()
217    }
218
219    fn write_config(&self, config: &TelemetryConfig) {
220        let _ = std::fs::create_dir_all(&self.global_dir);
221        if let Ok(text) = serde_json::to_string_pretty(config) {
222            let _ = std::fs::write(self.config_path(), text + "\n");
223        }
224    }
225
226    fn append_to_queue(&self, lines: &[serde_json::Value]) -> std::io::Result<()> {
227        std::fs::create_dir_all(&self.global_dir)?;
228        let path = self.queue_path();
229
230        // Cap at MAX_QUEUE_BYTES — drop oldest lines if needed.
231        let payload = lines
232            .iter()
233            .filter_map(|l| serde_json::to_string(l).ok())
234            .collect::<Vec<_>>()
235            .join("\n")
236            + "\n";
237
238        let existing_len = std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0);
239        if existing_len + payload.len() as u64 > MAX_QUEUE_BYTES {
240            let existing = std::fs::read_to_string(&path).unwrap_or_default();
241            let combined = existing + &payload;
242            let trimmed = trim_queue(&combined, MAX_QUEUE_BYTES as usize);
243            return std::fs::write(&path, trimmed);
244        }
245
246        let mut f = std::fs::OpenOptions::new()
247            .create(true)
248            .append(true)
249            .open(&path)?;
250        f.write_all(payload.as_bytes())?;
251        Ok(())
252    }
253}
254
255// ── flush logic (runs in background thread) ──────────────────────────────────
256
257fn do_flush(queue_path: &Path, machine_id: &str, endpoint: &str) -> Option<()> {
258    let text = std::fs::read_to_string(queue_path).ok()?;
259    let today = utc_day();
260
261    let mut sendable: Vec<serde_json::Value> = Vec::new();
262    let mut keep: Vec<serde_json::Value> = Vec::new();
263
264    for line in text.lines() {
265        if line.trim().is_empty() {
266            continue;
267        }
268        let Ok(v) = serde_json::from_str::<serde_json::Value>(line) else {
269            continue;
270        };
271        let is_lifecycle = v.get("type").and_then(|t| t.as_str()) == Some("lifecycle");
272        let day = v.get("day").and_then(|d| d.as_str()).unwrap_or("");
273
274        if is_lifecycle || (!day.is_empty() && day < today.as_str()) {
275            sendable.push(v);
276        } else {
277            keep.push(v);
278        }
279    }
280
281    if sendable.is_empty() {
282        return Some(());
283    }
284
285    let events: Vec<serde_json::Value> = sendable
286        .iter()
287        .map(|v| {
288            if v.get("type").and_then(|t| t.as_str()) == Some("rollup") {
289                let tool = v.get("tool").and_then(|t| t.as_str()).unwrap_or("");
290                let calls = v.get("calls").and_then(|c| c.as_u64()).unwrap_or(0);
291                let errors = v.get("errors").and_then(|e| e.as_u64()).unwrap_or(0);
292                let saved = v.get("saved_chars").and_then(|s| s.as_u64()).unwrap_or(0);
293                let day = v.get("day").and_then(|d| d.as_str()).unwrap_or("");
294                serde_json::json!({
295                    "event": "usage_rollup",
296                    "day": day,
297                    "tool": tool,
298                    "call_count": calls,
299                    "error_count": errors,
300                    "saved_chars_bucket": bucket_chars(saved),
301                })
302            } else {
303                let event = v.get("event").and_then(|e| e.as_str()).unwrap_or("unknown");
304                let ts = v.get("ts").and_then(|t| t.as_str()).unwrap_or("");
305                let props = v.get("props").cloned().unwrap_or(serde_json::json!({}));
306                serde_json::json!({
307                    "event": event,
308                    "ts": ts,
309                    "props": props,
310                })
311            }
312        })
313        .collect();
314
315    let envelope = serde_json::json!({
316        "schema_version": SCHEMA_VERSION,
317        "machine_id": machine_id,
318        "codesynapse_version": env!("CARGO_PKG_VERSION"),
319        "os": std::env::consts::OS,
320        "arch": std::env::consts::ARCH,
321        "events": events,
322    });
323
324    let body = serde_json::to_string(&envelope).ok()?;
325    let result = ureq::post(endpoint)
326        .set("Content-Type", "application/json")
327        .timeout(std::time::Duration::from_secs(5))
328        .send_string(&body);
329
330    // Any response or error → remove sent entries from queue.
331    // We treat failure as "drop and move on" to avoid retrying indefinitely.
332    let _ = result;
333
334    let new_queue = keep
335        .iter()
336        .filter_map(|v| serde_json::to_string(v).ok())
337        .collect::<Vec<_>>()
338        .join("\n");
339
340    if new_queue.is_empty() {
341        let _ = std::fs::remove_file(queue_path);
342    } else {
343        let _ = std::fs::write(queue_path, new_queue + "\n");
344    }
345
346    Some(())
347}
348
349// ── helpers ───────────────────────────────────────────────────────────────────
350
351fn utc_day() -> String {
352    let secs = SystemTime::now()
353        .duration_since(UNIX_EPOCH)
354        .unwrap_or_default()
355        .as_secs();
356    let days = secs / 86400;
357    // Convert day count since epoch to YYYY-MM-DD.
358    // Using a simple algorithm (Tomohiko Sakamoto / civil date).
359    let z = days as i64 + 719468;
360    let era = if z >= 0 { z } else { z - 146096 } / 146097;
361    let doe = z - era * 146097;
362    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
363    let y = yoe + era * 400;
364    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
365    let mp = (5 * doy + 2) / 153;
366    let d = doy - (153 * mp + 2) / 5 + 1;
367    let m = if mp < 10 { mp + 3 } else { mp - 9 };
368    let y = if m <= 2 { y + 1 } else { y };
369    format!("{:04}-{:02}-{:02}", y, m, d)
370}
371
372fn utc_now_iso() -> String {
373    let secs = SystemTime::now()
374        .duration_since(UNIX_EPOCH)
375        .unwrap_or_default()
376        .as_secs();
377    let day = utc_day();
378    let h = (secs % 86400) / 3600;
379    let m = (secs % 3600) / 60;
380    let s = secs % 60;
381    format!("{}T{:02}:{:02}:{:02}Z", day, h, m, s)
382}
383
384fn new_uuid() -> String {
385    // UUID v4 using random bytes from /dev/urandom or a fallback.
386    let mut bytes = [0u8; 16];
387    if let Ok(mut f) = std::fs::File::open("/dev/urandom") {
388        use std::io::Read;
389        let _ = f.read_exact(&mut bytes);
390    } else {
391        // Fallback: mix of timestamp + pid.
392        let ts = SystemTime::now()
393            .duration_since(UNIX_EPOCH)
394            .unwrap_or_default()
395            .subsec_nanos();
396        let pid = std::process::id();
397        bytes[..4].copy_from_slice(&ts.to_le_bytes());
398        bytes[4..8].copy_from_slice(&pid.to_le_bytes());
399    }
400    // Set version (4) and variant bits.
401    bytes[6] = (bytes[6] & 0x0f) | 0x40;
402    bytes[8] = (bytes[8] & 0x3f) | 0x80;
403    format!(
404        "{:02x}{:02x}{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}",
405        bytes[0], bytes[1], bytes[2], bytes[3],
406        bytes[4], bytes[5],
407        bytes[6], bytes[7],
408        bytes[8], bytes[9],
409        bytes[10], bytes[11], bytes[12], bytes[13], bytes[14], bytes[15],
410    )
411}
412
413fn env_flag(name: &str) -> bool {
414    !matches!(
415        std::env::var(name).ok().as_deref(),
416        Some("") | Some("0") | Some("false") | None
417    )
418}
419
420fn trim_queue(combined: &str, max_bytes: usize) -> String {
421    if combined.len() <= max_bytes {
422        return combined.to_string();
423    }
424    let excess = combined.len() - max_bytes;
425    let trimmed = &combined[excess..];
426    // Skip partial first line.
427    if let Some(pos) = trimmed.find('\n') {
428        trimmed[pos + 1..].to_string()
429    } else {
430        String::new()
431    }
432}
433
434pub fn bucket_chars(n: u64) -> &'static str {
435    if n < 1_000 {
436        "<1k"
437    } else if n < 10_000 {
438        "1k-10k"
439    } else if n < 100_000 {
440        "10k-100k"
441    } else {
442        "100k+"
443    }
444}
445
446pub fn bucket_count(n: usize) -> &'static str {
447    if n < 100 {
448        "<100"
449    } else if n < 1_000 {
450        "100-1k"
451    } else if n < 10_000 {
452        "1k-10k"
453    } else if n < 100_000 {
454        "10k-100k"
455    } else {
456        "100k+"
457    }
458}
459
460pub fn bucket_duration_ms(ms: u64) -> &'static str {
461    if ms < 10_000 {
462        "<10s"
463    } else if ms < 60_000 {
464        "10-60s"
465    } else if ms < 300_000 {
466        "1-5m"
467    } else {
468        "5m+"
469    }
470}
471
472#[cfg(test)]
473mod tests {
474    use super::*;
475
476    #[test]
477    fn utc_day_format() {
478        let d = utc_day();
479        assert_eq!(d.len(), 10);
480        assert!(d.contains('-'));
481    }
482
483    #[test]
484    fn uuid_format() {
485        let u = new_uuid();
486        let parts: Vec<&str> = u.split('-').collect();
487        assert_eq!(parts.len(), 5);
488        assert_eq!(parts[0].len(), 8);
489    }
490
491    #[test]
492    fn bucket_chars_boundaries() {
493        assert_eq!(bucket_chars(0), "<1k");
494        assert_eq!(bucket_chars(999), "<1k");
495        assert_eq!(bucket_chars(1_000), "1k-10k");
496        assert_eq!(bucket_chars(100_000), "100k+");
497    }
498
499    #[test]
500    fn disabled_by_default_no_config() {
501        let dir = tempfile::tempdir().unwrap();
502        let t = Telemetry::new(dir.path().to_path_buf());
503        assert!(!t.is_enabled());
504    }
505
506    #[test]
507    fn set_enabled_writes_config() {
508        let dir = tempfile::tempdir().unwrap();
509        let t = Telemetry::new(dir.path().to_path_buf());
510        t.set_enabled(true);
511        assert!(t.is_enabled());
512        let config = t.read_config().unwrap();
513        assert!(config.enabled);
514        assert_eq!(config.consent_source, "cli");
515    }
516
517    #[test]
518    fn record_and_persist() {
519        let dir = tempfile::tempdir().unwrap();
520        let t = Telemetry::new(dir.path().to_path_buf());
521        t.set_enabled(true);
522        t.record_usage("codesynapse_context", 5000, true);
523        t.record_usage("codesynapse_context", 3000, false);
524        t.persist_sync();
525        let queue = std::fs::read_to_string(t.queue_path()).unwrap();
526        assert!(queue.contains("codesynapse_context"));
527        assert!(queue.contains("\"calls\":2"));
528        assert!(queue.contains("\"errors\":1"));
529    }
530
531    #[test]
532    fn disabled_no_record() {
533        let dir = tempfile::tempdir().unwrap();
534        let t = Telemetry::new(dir.path().to_path_buf());
535        // default disabled — record_usage is a no-op
536        t.record_usage("codesynapse_context", 5000, true);
537        t.persist_sync();
538        assert!(!t.queue_path().exists());
539    }
540}
codesynapse_core/telemetry.rs

codesynapse_core/
telemetry.rs
