use std::io::{IsTerminal, Write};
use std::process::ExitCode;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Mutex, PoisonError};
use std::time::{Duration, Instant};
use cerulion_cli_engine::auth;
use cerulion_cli_engine::error::{CliError, CliResult};
use cerulion_cli_engine::login_cmd::LoginOutcome;
use cerulion_telemetry::consent::{self, Source};
use cerulion_telemetry::{guard, Client, Common, EventSpec, Props, DEFAULT_SHUTDOWN_BUDGET};
use crate::cli::TelemetryAction;
pub const CLI_COMMAND_RUN: EventSpec = EventSpec {
name: "cli_command_run",
allowlist: &[
"verb",
"subverb",
"exit_code",
"duration_bucket",
"install_method",
],
};
pub const CLI_LOGIN_COMPLETED: EventSpec = EventSpec {
name: "cli_login_completed",
allowlist: &["is_account_switch"],
};
const BAKED_KEY: Option<&str> = option_env!("CERULION_POSTHOG_KEY");
static SENDING: AtomicBool = AtomicBool::new(false);
static NOTICE_RUN: AtomicBool = AtomicBool::new(false);
static SWITCH_RUN: AtomicBool = AtomicBool::new(false);
static UNCARRIED: AtomicBool = AtomicBool::new(false);
static CLIENT: Mutex<Option<Client>> = Mutex::new(None);
static CI_RUN: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
fn ci_run() -> bool {
*CI_RUN.get_or_init(cerulion_telemetry::ci::detected)
}
#[cfg(feature = "telemetry")]
const PENDING_ALIAS_FILE: &str = "telemetry_alias_pending";
#[cfg(feature = "telemetry")]
const ANON_ACCOUNT_FILE: &str = "telemetry_anon_account";
pub const NOTICE: &str = "Cerulion sends usage telemetry; turn it off with \
`cerulion telemetry off` (details: https://cerulion.com/telemetry).";
const DETAILS_URL: &str = "https://cerulion.com/telemetry";
const UNRECORDED_VERBS: &[&str] = &["telemetry", "completions", "bagd"];
const UNRECORDED_SUBVERBS: &[&str] = &["run-worker", "run-gateway"];
pub fn common() -> Common {
Common {
surface: "cli".into(),
env: if cfg!(debug_assertions) {
"dev"
} else {
"prod"
}
.into(),
app_version: env!("CARGO_PKG_VERSION").into(),
channel: None,
ci: ci_run(),
}
}
pub fn duration_bucket(elapsed: Duration) -> &'static str {
match elapsed.as_secs() {
0 => "lt_1s",
1..=9 => "1s_10s",
10..=59 => "10s_1m",
60..=599 => "1m_10m",
_ => "gte_10m",
}
}
pub fn exit_code_number(code: ExitCode) -> i64 {
(0..=u8::MAX)
.find(|n| ExitCode::from(*n) == code)
.map_or(-1, i64::from)
}
pub fn command_run_props(
verb: &str,
subverb: Option<&str>,
code: ExitCode,
elapsed: Duration,
install_method: Option<&str>,
) -> Props {
let mut props: Props = vec![("verb".into(), verb.into())];
if let Some(subverb) = subverb {
props.push(("subverb".into(), subverb.into()));
}
props.push(("exit_code".into(), exit_code_number(code).into()));
props.push(("duration_bucket".into(), duration_bucket(elapsed).into()));
if let Some(method) = install_method {
props.push(("install_method".into(), method.into()));
}
props
}
pub fn is_recorded(verb: &str, subverb: Option<&str>) -> bool {
!UNRECORDED_VERBS.contains(&verb) && !subverb.is_some_and(|s| UNRECORDED_SUBVERBS.contains(&s))
}
pub struct CommandRun {
verb: String,
subverb: Option<String>,
started: Instant,
}
impl CommandRun {
pub fn start(
matches: &clap::ArgMatches,
switch: &crate::network_switch::SwitchEnv,
) -> Option<CommandRun> {
let (verb, sub) = matches.subcommand()?;
let subverb = sub.subcommand_name();
if !is_recorded(verb, subverb) {
return None;
}
if crate::network_switch::engaged(matches, switch) {
#[cfg(feature = "telemetry")]
if key_present() && consent::status().enabled {
SWITCH_RUN.store(true, Ordering::Relaxed);
}
return None;
}
let client = Client::from_env_or_key(BAKED_KEY, common())?;
match consent::notice_shown() {
Ok(true) => {}
Ok(false) | Err(cerulion_telemetry::Error::Json(_)) if ci_run() => {}
Ok(false) | Err(cerulion_telemetry::Error::Json(_))
if std::io::stderr().is_terminal() =>
{
if writeln!(std::io::stderr(), "{NOTICE}").is_ok() {
let _ = consent::mark_notice_shown();
}
NOTICE_RUN.store(true, Ordering::Relaxed);
return None;
}
Ok(false) | Err(cerulion_telemetry::Error::Json(_)) => {}
Err(_) => return None,
}
if !settle_anon_account() {
return None;
}
merge_pending_alias(&client);
*CLIENT.lock().unwrap_or_else(PoisonError::into_inner) = Some(client);
SENDING.store(true, Ordering::Relaxed);
Some(CommandRun {
verb: verb.to_owned(),
subverb: subverb.map(str::to_owned),
started: Instant::now(),
})
}
pub fn finish(self, code: ExitCode) {
let props = command_run_props(
&self.verb,
self.subverb.as_deref(),
code,
self.started.elapsed(),
crate::install_marker::current_method(),
);
emit(CLI_COMMAND_RUN, props);
SENDING.store(false, Ordering::Relaxed);
let client = CLIENT.lock().unwrap_or_else(PoisonError::into_inner).take();
if let Some(mut client) = client {
client.shutdown(DEFAULT_SHUTDOWN_BUDGET);
}
}
}
pub fn emit(spec: EventSpec, props: Props) {
if !SENDING.load(Ordering::Relaxed) || !consent::status().enabled {
return;
}
if !settle_anon_account() {
SENDING.store(false, Ordering::Relaxed);
return;
}
let guard = CLIENT.lock().unwrap_or_else(PoisonError::into_inner);
let Some(client) = guard.as_ref() else {
return;
};
let account = if ci_run() {
None
} else {
auth::load().state().and_then(|s| hosted_sub(&s.account_id))
};
match account {
Some(sub) => {
consent::while_enabled(|| client.capture(spec, &sub, props));
}
None => {
if let Ok(Some(anon_id)) = consent::event_anon_id() {
consent::while_enabled(|| client.capture_anonymous(spec, &anon_id, props));
}
}
}
}
#[cfg(feature = "telemetry")]
fn pending_alias_path() -> Option<std::path::PathBuf> {
Some(
consent::file_path()
.ok()?
.with_file_name(PENDING_ALIAS_FILE),
)
}
#[cfg(feature = "telemetry")]
fn anon_account_path() -> Option<std::path::PathBuf> {
Some(consent::file_path().ok()?.with_file_name(ANON_ACCOUNT_FILE))
}
#[cfg(feature = "telemetry")]
enum AnonAccount {
Unclaimed,
Bound(String),
Unknown,
}
#[cfg(feature = "telemetry")]
fn anon_account() -> AnonAccount {
let Some(path) = anon_account_path() else {
return AnonAccount::Unknown;
};
match std::fs::read_to_string(path) {
Ok(account) => AnonAccount::Bound(account),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => AnonAccount::Unclaimed,
Err(_) => AnonAccount::Unknown,
}
}
#[cfg(feature = "telemetry")]
fn bind_anon_account(account: &str) -> bool {
let Some(path) = anon_account_path() else {
return false;
};
match std::fs::write(path, account) {
Ok(()) => true,
Err(e) => {
tracing::warn!(error = %e, "telemetry anonymous id account record not written");
false
}
}
}
#[cfg(feature = "telemetry")]
fn claim_anon_id() -> bool {
let Some(path) = anon_account_path() else {
return false;
};
match std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(path)
{
Ok(_) => true,
Err(e) => {
if e.kind() != std::io::ErrorKind::AlreadyExists {
tracing::warn!(error = %e, "telemetry anonymous id not claimed for the login");
}
false
}
}
}
#[cfg(feature = "telemetry")]
fn remove_pending_alias(path: &std::path::Path) {
if let Err(e) = std::fs::remove_file(path) {
if e.kind() != std::io::ErrorKind::NotFound {
tracing::warn!(error = %e, "telemetry alias marker not removed");
}
}
}
#[cfg(feature = "telemetry")]
fn rotate_existing_anon_id() -> bool {
if !consent::file_path().is_ok_and(|p| p.exists()) {
return true;
}
if let Some(path) = pending_alias_path() {
remove_pending_alias(&path);
}
consent::rotate_anon_id().is_ok()
}
fn settle_anon_account() -> bool {
#[cfg(feature = "telemetry")]
{
let Some(account) = auth::load().state().map(|s| s.account_id.clone()) else {
return true;
};
match anon_account() {
AnonAccount::Bound(bound) if bound == account => return true,
AnonAccount::Unclaimed => {}
AnonAccount::Bound(_) | AnonAccount::Unknown => {
if !rotate_existing_anon_id() {
return false;
}
}
}
bind_anon_account(&account)
}
#[cfg(not(feature = "telemetry"))]
true
}
fn merge_pending_alias(client: &Client) {
#[cfg(feature = "telemetry")]
{
if ci_run() {
return;
}
let Some(path) = pending_alias_path().filter(|p| p.exists()) else {
return;
};
let sub = auth::load().state().and_then(|s| hosted_sub(&s.account_id));
let Ok(anon_id) = consent::anon_id() else {
return;
};
consent::while_enabled(|| {
if !path.exists() {
return;
}
if let (Some(sub), Some(anon_id)) = (sub, anon_id) {
client.alias(&sub, &anon_id);
}
remove_pending_alias(&path);
});
}
#[cfg(not(feature = "telemetry"))]
let _ = client;
}
fn hosted_sub(account_id: &str) -> Option<String> {
guard::check_sub(account_id)
.ok()
.map(|()| account_id.to_owned())
}
pub fn login_anon_id() -> Option<String> {
if ci_run() {
return None;
}
match auth::load() {
auth::LoadedAuth::Absent => {}
auth::LoadedAuth::Present(state) if !state.logged_in_ever => {}
auth::LoadedAuth::Present(_)
| auth::LoadedAuth::SignedOut { .. }
| auth::LoadedAuth::Corrupt(_) => return None,
}
#[cfg(feature = "telemetry")]
{
if !matches!(anon_account(), AnonAccount::Unclaimed) {
return None;
}
}
if SENDING.load(Ordering::Relaxed) {
if !consent::status().enabled {
return None;
}
let anon_id = consent::anon_id()
.ok()
.flatten()
.filter(|id| guard::check_anon_id(id).is_ok())?;
#[cfg(feature = "telemetry")]
{
if !claim_anon_id() {
return None;
}
}
return Some(anon_id);
}
if (NOTICE_RUN.load(Ordering::Relaxed) || SWITCH_RUN.load(Ordering::Relaxed))
&& consent::status().enabled
&& consent::anon_id().ok().flatten().is_some()
{
UNCARRIED.store(true, Ordering::Relaxed);
}
None
}
pub fn login_completed(outcome: &LoginOutcome) {
#[cfg(feature = "telemetry")]
let carried = settle_login(outcome);
#[cfg(not(feature = "telemetry"))]
let carried = outcome.telemetry_anon_id.as_deref();
if !SENDING.load(Ordering::Relaxed) || !consent::status().enabled {
return;
}
if ci_run() {
emit(CLI_LOGIN_COMPLETED, login_props(outcome.switched_account));
return;
}
let Some(sub) = hosted_sub(&outcome.state.account_id) else {
return;
};
let guard = CLIENT.lock().unwrap_or_else(PoisonError::into_inner);
let Some(client) = guard.as_ref() else {
return;
};
consent::while_enabled(|| {
if let Some(anon_id) = carried {
client.alias(&sub, anon_id);
}
client.capture(
CLI_LOGIN_COMPLETED,
&sub,
login_props(outcome.switched_account),
);
});
}
#[cfg(feature = "telemetry")]
fn settle_login(outcome: &LoginOutcome) -> Option<&str> {
let mut carried = outcome.telemetry_anon_id.as_deref();
if !consent::status().enabled {
return None;
}
let deferred =
UNCARRIED.load(Ordering::Relaxed) && hosted_sub(&outcome.state.account_id).is_some();
if consent::file_path().is_ok_and(|p| p.exists()) {
let account = &outcome.state.account_id;
let claimed_here = carried.is_some();
let anothers = match anon_account() {
AnonAccount::Unclaimed => false,
AnonAccount::Bound(bound) if bound.is_empty() && claimed_here => false,
AnonAccount::Bound(bound) => bound != *account,
AnonAccount::Unknown => true,
};
if anothers {
carried = None;
}
let owed = outcome.switched_account || anothers;
let in_play = SENDING.load(Ordering::Relaxed) || claimed_here || deferred;
if owed && !rotate_existing_anon_id() {
SENDING.store(false, Ordering::Relaxed);
bind_anon_account("");
} else if in_play && !bind_anon_account(account) {
SENDING.store(false, Ordering::Relaxed);
}
}
if deferred && !outcome.switched_account {
if let Some(path) = pending_alias_path() {
if let Err(e) = std::fs::write(path, b"") {
tracing::warn!(error = %e, "telemetry alias marker not written");
}
}
}
carried
}
pub fn login_props(is_account_switch: bool) -> Props {
vec![("is_account_switch".into(), is_account_switch.into())]
}
fn source_label(source: Source) -> &'static str {
match source {
Source::NotCompiled => "not compiled into this build",
Source::DoNotTrack => "set by DO_NOT_TRACK",
Source::EnvVar => "set by CERULION_TELEMETRY",
Source::File => "set by `cerulion telemetry on|off`",
Source::Default => "default",
}
}
pub fn run_verb(action: TelemetryAction, out: &mut impl Write) -> CliResult<()> {
let io = |e: std::io::Error| CliError::Io(e);
if !matches!(action, TelemetryAction::Status) {
let enabled = matches!(action, TelemetryAction::On);
if consent::status().source == Source::NotCompiled {
writeln!(
out,
"telemetry is not compiled into this build; it never sends anything"
)
.map_err(io)?;
return Ok(());
}
consent::set_enabled(enabled).map_err(|e| {
CliError::Validation(format!("could not update the telemetry consent file: {e}"))
})?;
}
let status = consent::status();
let state = if status.enabled { "on" } else { "off" };
writeln!(out, "telemetry: {state} ({})", source_label(status.source)).map_err(io)?;
if matches!(status.source, Source::DoNotTrack | Source::EnvVar) {
writeln!(
out,
"the environment variable overrides the consent file until it is unset"
)
.map_err(io)?;
}
#[cfg(feature = "telemetry")]
{
if status.enabled && !key_present() {
writeln!(out, "no telemetry key in this build: nothing is sent").map_err(io)?;
}
if let Ok(path) = consent::file_path() {
writeln!(out, "consent file: {}", path.display()).map_err(io)?;
}
}
writeln!(out, "what is sent: {DETAILS_URL}").map_err(io)?;
Ok(())
}
#[cfg(feature = "telemetry")]
fn key_present() -> bool {
let present = |k: &str| !k.trim().is_empty();
match std::env::var("POSTHOG_API_KEY") {
Ok(key) if present(&key) => true,
Err(std::env::VarError::NotUnicode(_)) => false,
_ => BAKED_KEY.is_some_and(present),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn durations_fall_into_coarse_buckets() {
let b = |s| duration_bucket(Duration::from_secs(s));
assert_eq!(duration_bucket(Duration::from_millis(999)), "lt_1s");
assert_eq!(b(1), "1s_10s");
assert_eq!(b(59), "10s_1m");
assert_eq!(b(60), "1m_10m");
assert_eq!(b(600), "gte_10m");
assert_eq!(b(u64::MAX), "gte_10m");
}
#[test]
fn exit_codes_map_to_their_numbers() {
assert_eq!(exit_code_number(ExitCode::SUCCESS), 0);
assert_eq!(exit_code_number(ExitCode::FAILURE), 1);
assert_eq!(exit_code_number(ExitCode::from(7)), 7);
assert_eq!(exit_code_number(ExitCode::from(255)), 255);
}
#[test]
fn command_run_props_stay_inside_the_allowlist_and_pass_the_guard() {
for (subverb, method) in [(None, None), (Some("run"), Some("install.sh"))] {
let props =
command_run_props("graph", subverb, ExitCode::FAILURE, Duration::ZERO, method);
let (kept, dropped) = guard::filter(props.clone(), CLI_COMMAND_RUN.allowlist);
assert!(dropped.is_empty(), "{dropped:?}");
assert_eq!(kept, props);
}
guard::check_event_name(CLI_COMMAND_RUN.name).expect("event name");
guard::check_event_name(CLI_LOGIN_COMPLETED.name).expect("event name");
for switch in [false, true] {
let (_, dropped) = guard::filter(login_props(switch), CLI_LOGIN_COMPLETED.allowlist);
assert!(dropped.is_empty(), "{dropped:?}");
}
let c = common();
for value in [&c.surface, &c.env, &c.app_version] {
guard::check_str(value).expect(value);
}
}
#[test]
fn consent_completion_daemon_and_subprocess_verbs_are_unrecorded() {
assert!(!is_recorded("telemetry", Some("off")));
assert!(!is_recorded("completions", None));
assert!(!is_recorded("bagd", None));
assert!(!is_recorded("graph", Some("run-worker")));
assert!(!is_recorded("graph", Some("run-gateway")));
assert!(is_recorded("graph", Some("run")));
assert!(is_recorded("login", None));
}
#[test]
fn every_unrecorded_name_is_a_real_subcommand() {
let cli = crate::families::cli_command();
let names: Vec<&str> = cli.get_subcommands().map(|c| c.get_name()).collect();
for verb in UNRECORDED_VERBS {
assert!(names.contains(verb), "{verb}");
}
let graph = cli.find_subcommand("graph").expect("graph");
for sub in UNRECORDED_SUBVERBS {
assert!(graph.find_subcommand(sub).is_some(), "{sub}");
}
}
}