bctx 0.1.39

//! Local-first usage telemetry. Every command is recorded locally; the cloud receives at
//! most one per-day snapshot every SYNC_INTERVAL_HOURS, sent by a detached
//! `bctx sync --telemetry` process, so no command ever waits on the network.

use anyhow::{Context, Result};
use chrono::{DateTime, Duration, Utc};
use serde::{Deserialize, Serialize};
use std::path::PathBuf;

const SYNC_INTERVAL_HOURS: i64 = 6;
const SNAPSHOT_WINDOW_DAYS: i64 = 35;

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct TelemetryState {
    pub device_id: String,
    /// The account this device reports to; another account starts a new device.
    pub user_id: String,
    /// Executions before this instant were sent one by one by older clients;
    /// snapshots never include them, so nothing is counted twice.
    pub epoch: String,
    pub last_attempt: Option<String>,
    pub last_success: Option<String>,
}

fn bctx_dir() -> PathBuf {
    PathBuf::from(crate::commands::home_dir()).join(".bctx")
}

fn state_path() -> PathBuf {
    bctx_dir().join("telemetry.json")
}

/// Readers see the old file or the new one, never a half-written one: the contents go to a
/// temporary file beside `path`, which then replaces it in one rename.
fn write_atomically(path: &std::path::Path, contents: &[u8]) -> std::io::Result<()> {
    let tmp = path.with_extension(format!("tmp-{}", uuid::Uuid::new_v4().simple()));
    std::fs::write(&tmp, contents)?;
    std::fs::rename(&tmp, path).inspect_err(|_| {
        let _ = std::fs::remove_file(&tmp);
    })
}

fn save_state(state: &TelemetryState) -> Result<()> {
    std::fs::create_dir_all(bctx_dir())?;
    write_atomically(
        &state_path(),
        serde_json::to_string_pretty(state)?.as_bytes(),
    )?;
    Ok(())
}

pub fn load_or_init_state(user_id: &str) -> Result<TelemetryState> {
    if let Ok(raw) = std::fs::read_to_string(state_path()) {
        if let Ok(state) = serde_json::from_str::<TelemetryState>(&raw) {
            if state.user_id == user_id {
                return Ok(state);
            }
        }
    }
    let state = TelemetryState {
        device_id: uuid::Uuid::new_v4().to_string(),
        user_id: user_id.to_string(),
        epoch: Utc::now().to_rfc3339(),
        last_attempt: None,
        last_success: None,
    };
    save_state(&state)?;
    Ok(state)
}

pub fn sync_due(state: &TelemetryState, now: DateTime<Utc>) -> bool {
    match state
        .last_attempt
        .as_deref()
        .and_then(|s| DateTime::parse_from_rfc3339(s).ok())
    {
        // A future attempt means the clock was ahead when it was stamped; waiting for that
        // moment could stop telemetry for years.
        Some(t) => {
            let t = t.with_timezone(&Utc);
            t > now || now - t >= Duration::hours(SYNC_INTERVAL_HOURS)
        }
        None => true,
    }
}

/// Lower bound for snapshot rows: the later of the device epoch and the window start. The
/// window starts at 00:00 UTC, so every day sent is whole: the server replaces a day's total,
/// and a mid-day start would overwrite the oldest day with part of it on every sync.
pub fn snapshot_since(state: &TelemetryState, now: DateTime<Utc>) -> String {
    let window = (now - Duration::days(SNAPSHOT_WINDOW_DAYS - 1))
        .date_naive()
        .and_hms_opt(0, 0, 0)
        .expect("midnight exists")
        .and_utc();
    match DateTime::parse_from_rfc3339(&state.epoch) {
        Ok(epoch) if epoch.with_timezone(&Utc) > window => epoch.with_timezone(&Utc).to_rfc3339(),
        _ => window.to_rfc3339(),
    }
}

/// Upload rows for the local rollup, summed again where names collapse to the same program.
fn snapshot_rows(daily: Vec<forge::tracker::store::DailyUsage>) -> Vec<cloud::usage::UsageRow> {
    let mut sums: std::collections::BTreeMap<(String, String), (i64, i64, i64)> =
        std::collections::BTreeMap::new();
    for d in daily {
        let e = sums
            .entry((d.day, cloud::usage::skill_name(&d.program)))
            .or_default();
        e.0 += d.tokens_sent;
        e.1 += d.tokens_saved;
        e.2 += d.runs;
    }
    sums.into_iter()
        .map(
            |((day, skill), (tokens_sent, tokens_saved, runs))| cloud::usage::UsageRow {
                day,
                skill,
                tokens_sent,
                tokens_saved,
                runs,
            },
        )
        .collect()
}

/// Logging out ends this device: the next login, to whichever account, starts afresh.
pub fn forget_device() -> Result<()> {
    match std::fs::remove_file(state_path()) {
        Err(e) if e.kind() != std::io::ErrorKind::NotFound => Err(e.into()),
        _ => Ok(()),
    }
}

/// Called before a command runs: a new device's epoch must precede the command's record,
/// or the first command after a login would never be sent.
pub fn start_device_before_recording() {
    if let Some(token) = cloud::client::auth::load_token() {
        let _ = load_or_init_state(&token.user_id);
    }
}

/// Called after every recorded command. Returns immediately in every case.
pub fn maybe_spawn_sync() {
    let Some(token) = cloud::client::auth::load_token() else {
        return;
    };
    let Ok(mut state) = load_or_init_state(&token.user_id) else {
        return;
    };
    let now = Utc::now();
    if !sync_due(&state, now) {
        return;
    }
    state.last_attempt = Some(now.to_rfc3339());
    if save_state(&state).is_err() {
        return;
    }
    if let Ok(exe) = std::env::current_exe() {
        let _ = crate::detach::spawn(|| {
            let mut cmd = std::process::Command::new(&exe);
            cmd.args(["sync", "--telemetry"])
                .stdin(std::process::Stdio::null())
                .stdout(std::process::Stdio::null())
                .stderr(std::process::Stdio::null());
            cmd
        });
    }
}

/// `bctx sync --telemetry`: build and send the snapshot now. Returns the rows sent.
pub fn sync_now() -> Result<usize> {
    let token = cloud::client::auth::load_token().context("not logged in (run `bctx login`)")?;
    let mut state = load_or_init_state(&token.user_id)?;
    let now = Utc::now();
    let store = forge::tracker::store::ExecutionStore::open(&bctx_dir().join("executions.db"))?;
    let mut rows = snapshot_rows(store.daily_rollup(&snapshot_since(&state, now))?);
    if rows.len() > cloud::usage::MAX_SNAPSHOT_ROWS {
        rows.sort_by(|a, b| b.day.cmp(&a.day));
        rows.truncate(cloud::usage::MAX_SNAPSHOT_ROWS);
    }
    let sent = rows.len();
    let snapshot = cloud::usage::UsageSnapshot {
        device_id: state.device_id.clone(),
        client_version: env!("CARGO_PKG_VERSION").to_string(),
        rows,
    };
    cloud::client::sync::push_usage_snapshot(&token.endpoint, &token.access_token, &snapshot)?;
    state.last_success = Some(now.to_rfc3339());
    save_state(&state)?;
    Ok(sent)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn state(epoch: &str, last_attempt: Option<&str>) -> TelemetryState {
        TelemetryState {
            device_id: "d".into(),
            user_id: "u".into(),
            epoch: epoch.into(),
            last_attempt: last_attempt.map(String::from),
            last_success: None,
        }
    }

    fn at(s: &str) -> DateTime<Utc> {
        DateTime::parse_from_rfc3339(s).unwrap().with_timezone(&Utc)
    }

    #[test]
    fn sync_is_due_without_attempts_and_after_the_interval() {
        let now = at("2026-10-18T12:00:00+00:00");
        assert!(sync_due(&state("2026-10-01T00:00:00+00:00", None), now));
        assert!(!sync_due(
            &state(
                "2026-10-01T00:00:00+00:00",
                Some("2026-10-18T07:00:00+00:00")
            ),
            now
        ));
        assert!(sync_due(
            &state(
                "2026-10-01T00:00:00+00:00",
                Some("2026-10-18T06:00:00+00:00")
            ),
            now
        ));
        assert!(
            sync_due(
                &state(
                    "2026-10-01T00:00:00+00:00",
                    Some("2026-10-18T13:00:00+00:00")
                ),
                now
            ),
            "an attempt stamped in the future is due"
        );
    }

    #[test]
    fn snapshot_never_reaches_before_the_epoch_or_the_window() {
        let now = at("2026-10-18T12:00:00+00:00");
        assert_eq!(
            snapshot_since(&state("2026-10-17T09:30:00+00:00", None), now),
            "2026-10-17T09:30:00+00:00"
        );
        assert_eq!(
            snapshot_since(&state("2026-01-01T00:00:00+00:00", None), now),
            "2026-09-14T00:00:00+00:00"
        );
    }

    #[test]
    fn the_oldest_day_is_always_sent_whole() {
        // Snapshots replace a day's total, so a window starting mid-day would overwrite the
        // oldest day with a partial sum on every sync, and that value would then freeze.
        let old = state("2026-01-01T00:00:00+00:00", None);
        let morning = snapshot_since(&old, at("2026-10-18T09:00:00+00:00"));
        let evening = snapshot_since(&old, at("2026-10-18T15:00:00+00:00"));
        assert_eq!(morning, evening);
        assert!(morning.ends_with("T00:00:00+00:00"), "{morning}");
    }

    fn usage(day: &str, program: &str, sent: i64, saved: i64) -> forge::tracker::store::DailyUsage {
        forge::tracker::store::DailyUsage {
            day: day.into(),
            program: program.into(),
            tokens_sent: sent,
            tokens_saved: saved,
            runs: 1,
        }
    }

    #[test]
    fn only_the_program_name_leaves_the_machine() {
        // The scout skill records whole shell lines; their arguments must never be uploaded.
        let rows = snapshot_rows(vec![
            usage(
                "2026-10-18",
                "cargo test --workspace --locked 2>&1 | grep -E 'test result'",
                10,
                90,
            ),
            usage("2026-10-18", "/usr/bin/git", 5, 5),
            usage("2026-10-18", "ls;<img src=x onerror=alert(1)>", 1, 1),
            usage(
                "2026-10-18",
                "export TOKEN=secret; curl https://example.com",
                2,
                2,
            ),
        ]);
        let names: Vec<&str> = rows.iter().map(|r| r.skill.as_str()).collect();
        assert_eq!(names, vec!["cargo", "export", "git", "ls"]);
    }

    #[test]
    fn rows_that_end_up_with_the_same_name_are_added_together() {
        let rows = snapshot_rows(vec![
            usage("2026-10-18", "git", 10, 40),
            usage("2026-10-18", "git status | head", 5, 5),
        ]);
        assert_eq!(rows.len(), 1);
        assert_eq!(
            (rows[0].tokens_sent, rows[0].tokens_saved, rows[0].runs),
            (15, 45, 2)
        );
    }

    #[test]
    fn a_reader_never_sees_a_half_written_state_file() {
        // Parallel commands read telemetry.json while one of them saves it; a torn read
        // would mint a new device and epoch.
        let dir = std::env::temp_dir().join(format!("bctx-atomic-{}", uuid::Uuid::new_v4()));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("telemetry.json");
        let doc = |n: usize| serde_json::json!({ "n": n, "pad": "x".repeat(200_000) }).to_string();
        write_atomically(&path, doc(0).as_bytes()).unwrap();
        let writer = {
            let path = path.clone();
            std::thread::spawn(move || {
                for n in 1..=300 {
                    write_atomically(&path, doc(n).as_bytes()).unwrap();
                }
            })
        };
        let mut torn = 0;
        while !writer.is_finished() {
            let raw = std::fs::read_to_string(&path).unwrap_or_default();
            if serde_json::from_str::<serde_json::Value>(&raw).is_err() {
                torn += 1;
            }
        }
        writer.join().unwrap();
        let leftovers = std::fs::read_dir(&dir).unwrap().count();
        let _ = std::fs::remove_dir_all(&dir);
        assert_eq!(torn, 0, "{torn} reads saw a half-written file");
        assert_eq!(leftovers, 1, "temporary files were left behind");
    }

    #[test]
    fn a_name_is_never_longer_than_the_server_accepts() {
        let long = "a".repeat(100);
        let rows = snapshot_rows(vec![
            usage("2026-10-18", &long, 1, 1),
            usage("2026-10-18", ";;", 1, 1),
        ]);
        assert!(rows
            .iter()
            .all(|r| !r.skill.is_empty() && r.skill.len() <= 64));
    }
}