bctx 0.1.38

//! 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")
}

fn save_state(state: &TelemetryState) -> Result<()> {
    std::fs::create_dir_all(bctx_dir())?;
    std::fs::write(state_path(), serde_json::to_string_pretty(state)?)?;
    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())
    {
        Some(t) => now - t.with_timezone(&Utc) >= 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(),
    }
}

/// The name a row is uploaded under: the program's basename, never its arguments. Some
/// records hold whole shell lines (paths, tokens, URLs) that must stay on this machine.
fn skill_name(program: &str) -> String {
    let first = program.split_whitespace().next().unwrap_or("");
    let path: String = first
        .chars()
        .take_while(|&c| c == '/' || cloud::usage::is_skill_char(c))
        .collect();
    let mut name: String = path.rsplit('/').next().unwrap_or("").to_string();
    name.truncate(cloud::usage::MAX_SKILL_LEN);
    if name.is_empty() {
        "other".into()
    } else {
        name
    }
}

/// 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, 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 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 _ = spawn_detached(&exe, &["sync", "--telemetry"]);
    }
}

/// Starts `exe` outside the terminal's process group (Unix) or console and job (Windows),
/// so Ctrl-C or closing the window while the command finishes cannot cut a sync short.
fn spawn_detached(exe: &std::path::Path, args: &[&str]) -> std::io::Result<std::process::Child> {
    let command = || {
        let mut cmd = std::process::Command::new(exe);
        cmd.args(args)
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null());
        cmd
    };
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        command().process_group(0).spawn()
    }
    #[cfg(windows)]
    {
        use std::os::windows::process::CommandExt;
        const DETACHED_PROCESS: u32 = 0x0000_0008;
        const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200;
        const CREATE_BREAKAWAY_FROM_JOB: u32 = 0x0100_0000;
        let flags = DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP;
        // A job that forbids breakaway refuses the flag; detaching from the console still helps.
        command()
            .creation_flags(flags | CREATE_BREAKAWAY_FROM_JOB)
            .spawn()
            .or_else(|_| command().creation_flags(flags).spawn())
    }
    #[cfg(not(any(unix, windows)))]
    {
        command().spawn()
    }
}

/// `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
        ));
    }

    #[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)
        );
    }

    #[cfg(unix)]
    #[test]
    fn the_sync_process_is_out_of_reach_of_the_terminals_ctrl_c() {
        // Ctrl-C signals the terminal's foreground process group; the sync must not be in it.
        let mut child = spawn_detached(std::path::Path::new("sleep"), &["5"]).unwrap();
        let pgid_of = |pid: u32| {
            let out = std::process::Command::new("ps")
                .args(["-o", "pgid=", "-p", &pid.to_string()])
                .output()
                .unwrap();
            String::from_utf8_lossy(&out.stdout).trim().to_string()
        };
        let child_group = pgid_of(child.id());
        let _ = child.kill();
        let _ = child.wait();
        assert_eq!(child_group, child.id().to_string(), "leads its own group");
        assert_ne!(child_group, pgid_of(std::process::id()));
    }

    #[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));
    }
}