tenant.openagents/omega

1mod event_coalescer;
2
3use crate::TelemetrySettings;
4use anyhow::{Context as _, Result};
5use clock::SystemClock;
6use fs::Fs;
7use futures::channel::mpsc;
8use futures::{Future, StreamExt};
9use gpui::{App, AppContext as _, BackgroundExecutor, Task};
10use http_client::{self, AsyncBody, HttpClient, HttpClientWithUrl, Method, Request};
11use parking_lot::Mutex;
12use regex::Regex;
13use release_channel::ReleaseChannel;
14use settings::{Settings, SettingsStore};
15use sha2::{Digest, Sha256};
16use std::collections::HashSet;
17use std::fs::File;
18use std::io::Write;
19use std::sync::LazyLock;
20use std::time::Instant;
21use std::{env, mem, path::PathBuf, sync::Arc, time::Duration};
22use telemetry_events::{AssistantEventData, AssistantPhase, Event, EventRequestBody, EventWrapper};
23
24pub struct TelemetrySubscription {
25    pub historical_events: Result<HistoricalEvents>,
26    pub queued_events: Vec<EventWrapper>,
27    pub live_events: mpsc::UnboundedReceiver<EventWrapper>,
28}
29
30pub struct HistoricalEvents {
31    pub events: Vec<EventWrapper>,
32    pub parse_error_count: usize,
33}
34use util::ResultExt as _;
35use worktree::{UpdatedEntriesSet, WorktreeId};
36
37use self::event_coalescer::EventCoalescer;
38
39pub struct Telemetry {
40    clock: Arc<dyn SystemClock>,
41    http_client: Arc<HttpClientWithUrl>,
42    executor: BackgroundExecutor,
43    state: Arc<Mutex<TelemetryState>>,
44}
45
46struct TelemetryState {
47    settings: TelemetrySettings,
48    system_id: Option<Arc<str>>,       // Per system
49    installation_id: Option<Arc<str>>, // Per app installation (different for dev, nightly, preview, and stable)
50    session_id: Option<String>,        // Per app launch
51    metrics_id: Option<Arc<str>>,      // Per logged-in user
52    release_channel: Option<ReleaseChannel>,
53    architecture: &'static str,
54    events_queue: Vec<EventWrapper>,
55    flush_events_task: Option<Task<()>>,
56
57    log_file: Option<File>,
58    is_staff: Option<bool>,
59    first_event_date_time: Option<Instant>,
60    event_coalescer: EventCoalescer,
61    max_queue_size: usize,
62    worktrees_with_project_type_events_sent: HashSet<WorktreeId>,
63
64    os_name: String,
65    app_version: String,
66    os_version: Option<String>,
67
68    subscribers: Vec<mpsc::UnboundedSender<EventWrapper>>,
69}
70
71#[cfg(debug_assertions)]
72const MAX_QUEUE_LEN: usize = 5;
73
74#[cfg(not(debug_assertions))]
75const MAX_QUEUE_LEN: usize = 50;
76
77#[cfg(debug_assertions)]
78const FLUSH_INTERVAL: Duration = Duration::from_secs(1);
79
80#[cfg(not(debug_assertions))]
81const FLUSH_INTERVAL: Duration = Duration::from_secs(60 * 5);
82static ZED_CLIENT_CHECKSUM_SEED: LazyLock<Option<Vec<u8>>> = LazyLock::new(|| {
83    option_env!("ZED_CLIENT_CHECKSUM_SEED")
84        .map(|s| s.as_bytes().into())
85        .or_else(|| {
86            env::var("ZED_CLIENT_CHECKSUM_SEED")
87                .ok()
88                .map(|s| s.as_bytes().into())
89        })
90});
91
92pub static MINIDUMP_ENDPOINT: LazyLock<Option<String>> = LazyLock::new(|| {
93    option_env!("ZED_MINIDUMP_ENDPOINT")
94        .map(str::to_string)
95        .or_else(|| env::var("ZED_MINIDUMP_ENDPOINT").ok())
96});
97
98pub fn should_install_crash_handler(channel: ReleaseChannel) -> bool {
99    matches!(
100        env::var("ZED_GENERATE_MINIDUMPS").as_deref(),
101        Ok("true" | "1")
102    ) || (channel != ReleaseChannel::Dev && MINIDUMP_ENDPOINT.is_some())
103}
104
105static DOTNET_PROJECT_FILES_REGEX: LazyLock<Regex> = LazyLock::new(|| {
106    Regex::new(r"^(global\.json|Directory\.Build\.props|.*\.(csproj|fsproj|vbproj|sln))$").unwrap()
107});
108
109pub fn os_name() -> String {
110    #[cfg(target_os = "macos")]
111    {
112        "macOS".to_string()
113    }
114    #[cfg(target_os = "linux")]
115    {
116        format!("Linux {}", gpui::guess_compositor())
117    }
118    #[cfg(target_os = "freebsd")]
119    {
120        format!("FreeBSD {}", gpui::guess_compositor())
121    }
122
123    #[cfg(target_os = "windows")]
124    {
125        "Windows".to_string()
126    }
127}
128
129/// Note: This might do blocking IO! Only call from background threads
130pub fn os_version() -> String {
131    cfg_select! {
132       feature = "test-support" => {
133           // MacOS branch in particular is quite slow, hence we ought to "avoid" it in tests.
134           "test binary".to_owned()
135       }
136       target_os = "macos" => {
137           static MACOS_VERSION_REGEX: LazyLock<Regex> = LazyLock::new(|| {
138               Regex::new(r"(\s*\(Build [^)]*[0-9]\))").unwrap()
139           });
140           use objc2_foundation::NSProcessInfo;
141           let process_info = NSProcessInfo::processInfo();
142           let version_nsstring = process_info.operatingSystemVersionString();
143           // "Version 15.6.1 (Build 24G90)" -> "15.6.1 (Build 24G90)"
144           let version_string = version_nsstring.to_string().replace("Version ", "");
145           // "15.6.1 (Build 24G90)" -> "15.6.1"
146           // "26.0.0 (Build 25A5349a)" -> unchanged (Beta or Rapid Security Response; ends with letter)
147           MACOS_VERSION_REGEX
148               .replace_all(&version_string, "")
149               .to_string()
150       }
151       any(target_os = "linux", target_os = "freebsd") => {
152           use std::path::Path;
153
154           let content = if let Ok(file) = std::fs::read_to_string(&Path::new("/etc/os-release")) {
155               file
156           } else if let Ok(file) = std::fs::read_to_string(&Path::new("/usr/lib/os-release")) {
157               file
158           } else if let Ok(file) = std::fs::read_to_string(&Path::new("/var/run/os-release")) {
159               file
160           } else {
161               log::error!(
162                   "Failed to load /etc/os-release, /usr/lib/os-release, or /var/run/os-release"
163               );
164               "".to_string()
165           };
166           util::parse_os_release(&content).unwrap_or_else(|| "unknown".to_string())
167       }
168       target_os = "windows" => {
169           let mut info = unsafe { std::mem::zeroed() };
170           let status = unsafe { windows::Wdk::System::SystemServices::RtlGetVersion(&mut info) };
171           if status.is_ok() {
172               semver::Version::new(
173                   info.dwMajorVersion as _,
174                   info.dwMinorVersion as _,
175                   info.dwBuildNumber as _,
176               )
177               .to_string()
178           } else {
179               "unknown".to_string()
180           }
181       }
182    }
183}
184
185impl Telemetry {
186    pub fn new(
187        clock: Arc<dyn SystemClock>,
188        client: Arc<HttpClientWithUrl>,
189        cx: &mut App,
190    ) -> Arc<Self> {
191        let state = Arc::new(Mutex::new(TelemetryState {
192            settings: *TelemetrySettings::get_global(cx),
193            architecture: env::consts::ARCH,
194            release_channel: ReleaseChannel::try_global(cx),
195            system_id: None,
196            installation_id: None,
197            session_id: None,
198            metrics_id: None,
199            events_queue: Vec::new(),
200            flush_events_task: None,
201            log_file: None,
202            is_staff: None,
203            first_event_date_time: None,
204            event_coalescer: EventCoalescer::new(clock.clone()),
205            max_queue_size: MAX_QUEUE_LEN,
206            worktrees_with_project_type_events_sent: HashSet::new(),
207
208            os_version: None,
209            os_name: os_name(),
210            app_version: release_channel::AppVersion::global(cx).to_string(),
211            subscribers: Vec::new(),
212        }));
213
214        cx.background_spawn({
215            let state = state.clone();
216            let os_version = os_version();
217            state.lock().os_version = Some(os_version);
218            async move {
219                if let Some(tempfile) = File::create(Self::log_file_path()).ok() {
220                    state.lock().log_file = Some(tempfile);
221                }
222            }
223        })
224        .detach();
225
226        cx.observe_global::<SettingsStore>({
227            let state = state.clone();
228
229            move |cx| {
230                let mut state = state.lock();
231                state.settings = *TelemetrySettings::get_global(cx);
232            }
233        })
234        .detach();
235
236        let this = Arc::new(Self {
237            clock,
238            http_client: client,
239            executor: cx.background_executor().clone(),
240            state,
241        });
242
243        let (tx, mut rx) = mpsc::unbounded();
244        ::telemetry::init(tx);
245
246        cx.background_spawn({
247            let this = Arc::downgrade(&this);
248            async move {
249                if cfg!(feature = "test-support") {
250                    return;
251                }
252                while let Some(event) = rx.next().await {
253                    let Some(state) = this.upgrade() else { break };
254                    state.report_event(Event::Flexible(event))
255                }
256            }
257        })
258        .detach();
259
260        // We should only ever have one instance of Telemetry, leak the subscription to keep it alive
261        // rather than store in TelemetryState, complicating spawn as subscriptions are not Send
262        std::mem::forget(cx.on_app_quit({
263            let this = this.clone();
264            move |_| this.shutdown_telemetry()
265        }));
266
267        this
268    }
269
270    #[cfg(any(test, feature = "test-support"))]
271    fn shutdown_telemetry(self: &Arc<Self>) -> impl Future<Output = ()> + use<> {
272        Task::ready(())
273    }
274
275    // Skip calling this function in tests.
276    // TestAppContext ends up calling this function on shutdown and it panics when trying to find the TelemetrySettings
277    #[cfg(not(any(test, feature = "test-support")))]
278    fn shutdown_telemetry(self: &Arc<Self>) -> impl Future<Output = ()> + use<> {
279        telemetry::event!("App Closed");
280        // TODO: close final edit period and make sure it's sent
281        Task::ready(())
282    }
283
284    pub fn log_file_path() -> PathBuf {
285        paths::logs_dir().join("telemetry.log")
286    }
287
288    pub async fn subscribe_with_history(
289        self: &Arc<Self>,
290        fs: Arc<dyn Fs>,
291    ) -> TelemetrySubscription {
292        let historical_events = self.read_log_file(fs).await;
293
294        let mut state = self.state.lock();
295        let queued_events: Vec<EventWrapper> = state.events_queue.clone();
296
297        let (tx, rx) = mpsc::unbounded();
298        state.subscribers.push(tx);
299
300        drop(state);
301
302        TelemetrySubscription {
303            historical_events,
304            queued_events,
305            live_events: rx,
306        }
307    }
308
309    async fn read_log_file(self: &Arc<Self>, fs: Arc<dyn Fs>) -> anyhow::Result<HistoricalEvents> {
310        const MAX_LOG_READ: usize = 5 * 1024 * 1024;
311
312        let path = Self::log_file_path();
313
314        let content = fs
315            .load_bytes(&path)
316            .await
317            .with_context(|| format!("failed to load telemetry log from {:?}", path))?;
318
319        let start_offset = if content.len() > MAX_LOG_READ {
320            let skip = content.len() - MAX_LOG_READ;
321            content[skip..]
322                .iter()
323                .position(|&b| b == b'\n')
324                .map(|pos| skip + pos + 1)
325                .unwrap_or(skip)
326        } else {
327            0
328        };
329
330        let content_str = std::str::from_utf8(&content[start_offset..])
331            .context("telemetry log file contains invalid UTF-8")?;
332
333        let mut events = Vec::new();
334        let mut parse_error_count = 0;
335
336        for line in content_str.lines() {
337            if line.trim().is_empty() {
338                continue;
339            }
340            match serde_json::from_str::<EventWrapper>(line) {
341                Ok(event) => events.push(event),
342                Err(_) => parse_error_count += 1,
343            }
344        }
345
346        Ok(HistoricalEvents {
347            events,
348            parse_error_count,
349        })
350    }
351
352    pub fn has_checksum_seed(&self) -> bool {
353        ZED_CLIENT_CHECKSUM_SEED.is_some()
354    }
355
356    pub fn start(
357        self: &Arc<Self>,
358        system_id: Option<String>,
359        installation_id: Option<String>,
360        session_id: String,
361        cx: &App,
362    ) {
363        let mut state = self.state.lock();
364        state.system_id = system_id.map(|id| id.into());
365        state.installation_id = installation_id.map(|id| id.into());
366        state.session_id = Some(session_id);
367        state.app_version = release_channel::AppVersion::global(cx).to_string();
368        state.os_name = os_name();
369    }
370
371    pub fn metrics_enabled(self: &Arc<Self>) -> bool {
372        self.state.lock().settings.metrics
373    }
374
375    pub fn diagnostics_enabled(self: &Arc<Self>) -> bool {
376        self.state.lock().settings.diagnostics
377    }
378
379    pub fn set_authenticated_user_info(
380        self: &Arc<Self>,
381        metrics_id: Option<String>,
382        is_staff: bool,
383    ) {
384        let mut state = self.state.lock();
385
386        if !state.settings.metrics {
387            return;
388        }
389
390        let metrics_id: Option<Arc<str>> = metrics_id.map(|id| id.into());
391        state.metrics_id.clone_from(&metrics_id);
392        state.is_staff = Some(is_staff);
393        drop(state);
394    }
395
396    pub fn report_assistant_event(self: &Arc<Self>, event: AssistantEventData) {
397        let event_type = match event.phase {
398            AssistantPhase::Response => "Assistant Responded",
399            AssistantPhase::Invoked => "Assistant Invoked",
400            AssistantPhase::Accepted => "Assistant Response Accepted",
401            AssistantPhase::Rejected => "Assistant Response Rejected",
402        };
403
404        telemetry::event!(
405            event_type,
406            conversation_id = event.conversation_id,
407            kind = event.kind,
408            phase = event.phase,
409            message_id = event.message_id,
410            model = event.model,
411            model_provider = event.model_provider,
412            response_latency = event.response_latency,
413            error_message = event.error_message,
414            language_name = event.language_name,
415        );
416    }
417
418    pub fn log_edit_event(self: &Arc<Self>, environment: &'static str, is_via_ssh: bool) {
419        static LAST_EVENT_TIME: Mutex<Option<Instant>> = Mutex::new(None);
420
421        let mut state = self.state.lock();
422        let period_data = state.event_coalescer.log_event(environment);
423        drop(state);
424
425        if let Some(mut last_event) = LAST_EVENT_TIME.try_lock() {
426            let current_time = std::time::Instant::now();
427            let last_time = last_event.get_or_insert(current_time);
428
429            if current_time.duration_since(*last_time) > Duration::from_secs(60 * 10) {
430                *last_time = current_time;
431            } else {
432                return;
433            }
434
435            if let Some((start, end, environment)) = period_data {
436                let duration = end
437                    .saturating_duration_since(start)
438                    .min(Duration::from_secs(60 * 60 * 24))
439                    .as_millis() as i64;
440
441                telemetry::event!(
442                    "Editor Edited",
443                    duration = duration,
444                    environment = environment,
445                    is_via_ssh = is_via_ssh
446                );
447            }
448        }
449    }
450
451    pub fn report_discovered_project_type_events(
452        self: &Arc<Self>,
453        worktree_id: WorktreeId,
454        updated_entries_set: &UpdatedEntriesSet,
455    ) {
456        let Some(project_types) = self.detect_project_types(worktree_id, updated_entries_set)
457        else {
458            return;
459        };
460
461        for project_type in project_types {
462            telemetry::event!("Project Opened", project_type = project_type);
463        }
464    }
465
466    fn detect_project_types(
467        self: &Arc<Self>,
468        worktree_id: WorktreeId,
469        updated_entries_set: &UpdatedEntriesSet,
470    ) -> Option<Vec<String>> {
471        let mut state = self.state.lock();
472
473        if state
474            .worktrees_with_project_type_events_sent
475            .contains(&worktree_id)
476        {
477            return None;
478        }
479
480        let mut project_types: HashSet<&str> = HashSet::new();
481
482        for (path, _, _) in updated_entries_set.iter() {
483            let Some(file_name) = path.file_name() else {
484                continue;
485            };
486
487            let project_type = match file_name {
488                "pnpm-lock.yaml" => Some("pnpm"),
489                "yarn.lock" => Some("yarn"),
490                "package.json" => Some("node"),
491                _ if DOTNET_PROJECT_FILES_REGEX.is_match(file_name) => Some("dotnet"),
492                _ => None,
493            };
494
495            if let Some(project_type) = project_type {
496                project_types.insert(project_type);
497            };
498        }
499
500        if !project_types.is_empty() {
501            state
502                .worktrees_with_project_type_events_sent
503                .insert(worktree_id);
504        }
505
506        let mut project_types: Vec<_> = project_types.into_iter().map(String::from).collect();
507        project_types.sort();
508        Some(project_types)
509    }
510
511    /// Report a telemetry event that originated on a remote server.
512    ///
513    /// The remote server cannot upload telemetry itself, so it forwards events
514    /// (as a JSON-serialized [`Event`]) to the client. Since the OS metadata in
515    /// [`EventRequestBody`] is batch-level (describing the uploading client),
516    /// the remote server's OS is attached as event properties instead, so the
517    /// origin can still be distinguished downstream.
518    pub fn report_remote_event(
519        self: &Arc<Self>,
520        event_json: &str,
521        connection_type: &str,
522        os_name: String,
523        os_version: Option<String>,
524        architecture: String,
525    ) -> Result<()> {
526        // The remote server forwards a bare `telemetry_events::FlexibleEvent`
527        // (the type behind `telemetry::event!`), not the tagged `Event` enum.
528        let mut flexible: telemetry_events::FlexibleEvent =
529            serde_json::from_str(event_json).context("invalid remote telemetry event")?;
530        flexible
531            .event_properties
532            .insert("remote".into(), true.into());
533        flexible
534            .event_properties
535            .insert("remote_connection_type".into(), connection_type.into());
536        flexible
537            .event_properties
538            .insert("remote_os_name".into(), os_name.into());
539        flexible
540            .event_properties
541            .insert("remote_architecture".into(), architecture.into());
542        if let Some(os_version) = os_version {
543            flexible
544                .event_properties
545                .insert("remote_os_version".into(), os_version.into());
546        }
547        self.report_event(Event::Flexible(flexible));
548        Ok(())
549    }
550
551    /// Returns a snapshot of the currently queued (not-yet-flushed) telemetry
552    /// events, for use in tests.
553    #[cfg(any(test, feature = "test-support"))]
554    pub fn queued_events(self: &Arc<Self>) -> Vec<telemetry_events::FlexibleEvent> {
555        self.state
556            .lock()
557            .events_queue
558            .iter()
559            .map(|wrapper| {
560                let Event::Flexible(event) = &wrapper.event;
561                event.clone()
562            })
563            .collect()
564    }
565
566    fn report_event(self: &Arc<Self>, mut event: Event) {
567        let mut state = self.state.lock();
568        // RUST_LOG=telemetry=trace to debug telemetry events
569        log::trace!(target: "telemetry", "{:?}", event);
570
571        if !state.settings.metrics {
572            return;
573        }
574
575        match &mut event {
576            Event::Flexible(event) => event
577                .event_properties
578                .insert("event_source".into(), "zed".into()),
579        };
580
581        if state.flush_events_task.is_none() {
582            let this = self.clone();
583            state.flush_events_task = Some(self.executor.spawn(async move {
584                this.executor.timer(FLUSH_INTERVAL).await;
585                this.flush_events().detach();
586            }));
587        }
588
589        let date_time = self.clock.utc_now();
590
591        let milliseconds_since_first_event = match state.first_event_date_time {
592            Some(first_event_date_time) => date_time
593                .saturating_duration_since(first_event_date_time)
594                .min(Duration::from_secs(60 * 60 * 24))
595                .as_millis() as i64,
596            None => {
597                state.first_event_date_time = Some(date_time);
598                0
599            }
600        };
601
602        let signed_in = state.metrics_id.is_some();
603        let event_wrapper = EventWrapper {
604            signed_in,
605            milliseconds_since_first_event,
606            event,
607        };
608
609        state
610            .subscribers
611            .retain(|tx| tx.unbounded_send(event_wrapper.clone()).is_ok());
612
613        state.events_queue.push(event_wrapper);
614
615        if state.installation_id.is_some() && state.events_queue.len() >= state.max_queue_size {
616            drop(state);
617            self.flush_events().detach();
618        }
619    }
620
621    pub fn metrics_id(self: &Arc<Self>) -> Option<Arc<str>> {
622        self.state.lock().metrics_id.clone()
623    }
624
625    pub fn system_id(self: &Arc<Self>) -> Option<Arc<str>> {
626        self.state.lock().system_id.clone()
627    }
628
629    pub fn installation_id(self: &Arc<Self>) -> Option<Arc<str>> {
630        self.state.lock().installation_id.clone()
631    }
632
633    pub fn is_staff(self: &Arc<Self>) -> Option<bool> {
634        self.state.lock().is_staff
635    }
636
637    fn build_request(
638        self: &Arc<Self>,
639        // We take in the JSON bytes buffer so we can reuse the existing allocation.
640        mut json_bytes: Vec<u8>,
641        event_request: &EventRequestBody,
642    ) -> Result<Request<AsyncBody>> {
643        json_bytes.clear();
644        serde_json::to_writer(&mut json_bytes, event_request)?;
645
646        let checksum = calculate_json_checksum(&json_bytes).unwrap_or_default();
647
648        Ok(Request::builder()
649            .method(Method::POST)
650            .uri(
651                self.http_client
652                    .build_zed_api_url("/telemetry/events", &[])?
653                    .as_ref(),
654            )
655            .header("Content-Type", "application/json")
656            .header("x-zed-checksum", checksum)
657            .body(json_bytes.into())?)
658    }
659
660    pub async fn flush_events_inner(self: &Arc<Self>) -> Result<()> {
661        let (json_bytes, request_body) = {
662            let mut state = self.state.lock();
663            state.first_event_date_time = None;
664            let events = mem::take(&mut state.events_queue);
665            state.flush_events_task.take();
666            if events.is_empty() {
667                return Ok(());
668            }
669
670            let mut json_bytes = Vec::new();
671
672            if let Some(file) = &mut state.log_file {
673                for event in &events {
674                    json_bytes.clear();
675                    serde_json::to_writer(&mut json_bytes, event)?;
676                    file.write_all(&json_bytes)?;
677                    file.write_all(b"\n")?;
678                }
679            }
680
681            (
682                json_bytes,
683                EventRequestBody {
684                    system_id: state.system_id.as_deref().map(Into::into),
685                    installation_id: state.installation_id.as_deref().map(Into::into),
686                    session_id: state.session_id.clone(),
687                    metrics_id: state.metrics_id.as_deref().map(Into::into),
688                    is_staff: state.is_staff,
689                    app_version: state.app_version.clone(),
690                    os_name: state.os_name.clone(),
691                    os_version: state.os_version.clone(),
692                    architecture: state.architecture.to_string(),
693
694                    release_channel: state
695                        .release_channel
696                        .map(|channel| channel.display_name().to_owned()),
697                    events,
698                },
699            )
700        };
701
702        let request = self.build_request(json_bytes, &request_body)?;
703        let response = self.http_client.send(request).await?;
704        if response.status() != 200 {
705            log::error!("Failed to send events: HTTP {:?}", response.status());
706        }
707
708        anyhow::Ok(())
709    }
710
711    pub fn flush_events(self: &Arc<Self>) -> Task<()> {
712        let this = self.clone();
713        self.executor.spawn(async move {
714            this.flush_events_inner().await.log_err();
715        })
716    }
717}
718
719pub fn calculate_json_checksum(json: &impl AsRef<[u8]>) -> Option<String> {
720    let checksum_seed = ZED_CLIENT_CHECKSUM_SEED.as_ref()?;
721
722    let mut summer = Sha256::new();
723    summer.update(checksum_seed);
724    summer.update(json);
725    summer.update(checksum_seed);
726    let mut checksum = String::new();
727    for byte in summer.finalize().as_slice() {
728        use std::fmt::Write;
729        write!(&mut checksum, "{:02x}", byte).unwrap();
730    }
731
732    Some(checksum)
733}
734
735#[cfg(test)]
736mod tests {
737    use super::*;
738    use clock::FakeSystemClock;
739
740    use gpui::TestAppContext;
741    use http_client::FakeHttpClient;
742    use std::collections::HashMap;
743    use telemetry_events::FlexibleEvent;
744    use util::rel_path::RelPath;
745    use worktree::{PathChange, ProjectEntryId, WorktreeId};
746
747    #[gpui::test]
748    async fn test_telemetry_flush_on_max_queue_size(
749        executor: BackgroundExecutor,
750        cx: &mut TestAppContext,
751    ) {
752        init_test(cx);
753        let clock = Arc::new(FakeSystemClock::new());
754        let http = FakeHttpClient::with_200_response();
755        let system_id = Some("system_id".to_string());
756        let installation_id = Some("installation_id".to_string());
757        let session_id = "session_id".to_string();
758
759        let (telemetry, first_date_time, event) = cx.update(|cx| {
760            let telemetry = Telemetry::new(clock.clone(), http, cx);
761
762            telemetry.state.lock().max_queue_size = 4;
763            telemetry.start(system_id, installation_id, session_id, cx);
764
765            assert!(is_empty_state(&telemetry));
766
767            let first_date_time = clock.utc_now();
768            let event_properties = HashMap::from_iter([(
769                "test_key".to_string(),
770                serde_json::Value::String("test_value".to_string()),
771            )]);
772
773            let event = FlexibleEvent {
774                event_type: "test".to_string(),
775                event_properties,
776            };
777
778            (telemetry, first_date_time, event)
779        });
780
781        cx.update(|_cx| {
782            telemetry.report_event(Event::Flexible(event.clone()));
783            assert_eq!(telemetry.state.lock().events_queue.len(), 1);
784            assert!(telemetry.state.lock().flush_events_task.is_some());
785            assert_eq!(
786                telemetry.state.lock().first_event_date_time,
787                Some(first_date_time)
788            );
789
790            clock.advance(Duration::from_millis(100));
791
792            telemetry.report_event(Event::Flexible(event.clone()));
793            assert_eq!(telemetry.state.lock().events_queue.len(), 2);
794            assert!(telemetry.state.lock().flush_events_task.is_some());
795            assert_eq!(
796                telemetry.state.lock().first_event_date_time,
797                Some(first_date_time)
798            );
799
800            clock.advance(Duration::from_millis(100));
801
802            telemetry.report_event(Event::Flexible(event.clone()));
803            assert_eq!(telemetry.state.lock().events_queue.len(), 3);
804            assert!(telemetry.state.lock().flush_events_task.is_some());
805            assert_eq!(
806                telemetry.state.lock().first_event_date_time,
807                Some(first_date_time)
808            );
809
810            clock.advance(Duration::from_millis(100));
811
812            // Adding a 4th event should cause a flush
813            telemetry.report_event(Event::Flexible(event));
814        });
815
816        // Run the spawned flush task to completion
817        executor.run_until_parked();
818
819        cx.update(|_cx| {
820            assert!(is_empty_state(&telemetry));
821        });
822    }
823
824    #[gpui::test]
825    async fn test_telemetry_flush_on_flush_interval(
826        executor: BackgroundExecutor,
827        cx: &mut TestAppContext,
828    ) {
829        init_test(cx);
830        let clock = Arc::new(FakeSystemClock::new());
831        let http = FakeHttpClient::with_200_response();
832        let system_id = Some("system_id".to_string());
833        let installation_id = Some("installation_id".to_string());
834        let session_id = "session_id".to_string();
835
836        cx.update(|cx| {
837            let telemetry = Telemetry::new(clock.clone(), http, cx);
838            telemetry.state.lock().max_queue_size = 4;
839            telemetry.start(system_id, installation_id, session_id, cx);
840
841            assert!(is_empty_state(&telemetry));
842            let first_date_time = clock.utc_now();
843
844            let event_properties = HashMap::from_iter([(
845                "test_key".to_string(),
846                serde_json::Value::String("test_value".to_string()),
847            )]);
848
849            let event = FlexibleEvent {
850                event_type: "test".to_string(),
851                event_properties,
852            };
853
854            telemetry.report_event(Event::Flexible(event));
855            assert_eq!(telemetry.state.lock().events_queue.len(), 1);
856            assert!(telemetry.state.lock().flush_events_task.is_some());
857            assert_eq!(
858                telemetry.state.lock().first_event_date_time,
859                Some(first_date_time)
860            );
861
862            let duration = Duration::from_millis(1);
863
864            // Test 1 millisecond before the flush interval limit is met
865            executor.advance_clock(FLUSH_INTERVAL - duration);
866
867            assert!(!is_empty_state(&telemetry));
868
869            // Test the exact moment the flush interval limit is met
870            executor.advance_clock(duration);
871
872            assert!(is_empty_state(&telemetry));
873        });
874    }
875
876    #[gpui::test]
877    async fn test_report_remote_event_tags_origin(cx: &mut TestAppContext) {
878        init_test(cx);
879        let clock = Arc::new(FakeSystemClock::new());
880        let http = FakeHttpClient::with_200_response();
881
882        let telemetry = cx.update(|cx| {
883            let telemetry = Telemetry::new(clock.clone(), http, cx);
884            telemetry.start(
885                Some("system_id".to_string()),
886                Some("installation_id".to_string()),
887                "session_id".to_string(),
888                cx,
889            );
890            telemetry
891        });
892
893        // Mirror what the remote server forwards: a bare `FlexibleEvent`, which
894        // is the type produced by `telemetry::event!` / sent over the queue.
895        let event_json = serde_json::to_string(&FlexibleEvent {
896            event_type: "fs_watcher_poll".to_string(),
897            event_properties: HashMap::from_iter([(
898                "path".to_string(),
899                serde_json::Value::String("/code/project".to_string()),
900            )]),
901        })
902        .unwrap();
903
904        cx.update(|_| {
905            telemetry
906                .report_remote_event(
907                    &event_json,
908                    "ssh",
909                    "Linux".to_string(),
910                    Some("ubuntu 24.04".to_string()),
911                    "aarch64".to_string(),
912                )
913                .unwrap();
914        });
915
916        let queue = telemetry.state.lock().events_queue.clone();
917        assert_eq!(queue.len(), 1);
918        let Event::Flexible(event) = &queue[0].event;
919        assert_eq!(event.event_type, "fs_watcher_poll");
920        // Original properties are preserved.
921        assert_eq!(
922            event.event_properties.get("path"),
923            Some(&serde_json::Value::String("/code/project".to_string()))
924        );
925        // The remote server's OS is attached as properties, since the batch-level
926        // OS describes the uploading client rather than the remote host.
927        assert_eq!(
928            event.event_properties.get("remote"),
929            Some(&serde_json::Value::Bool(true))
930        );
931        assert_eq!(
932            event.event_properties.get("remote_connection_type"),
933            Some(&serde_json::Value::String("ssh".to_string()))
934        );
935        assert_eq!(
936            event.event_properties.get("remote_os_name"),
937            Some(&serde_json::Value::String("Linux".to_string()))
938        );
939        assert_eq!(
940            event.event_properties.get("remote_os_version"),
941            Some(&serde_json::Value::String("ubuntu 24.04".to_string()))
942        );
943        assert_eq!(
944            event.event_properties.get("remote_architecture"),
945            Some(&serde_json::Value::String("aarch64".to_string()))
946        );
947    }
948
949    #[gpui::test]
950    fn test_project_discovery_does_not_double_report(cx: &mut gpui::TestAppContext) {
951        init_test(cx);
952
953        let clock = Arc::new(FakeSystemClock::new());
954        let http = FakeHttpClient::with_200_response();
955        let telemetry = cx.update(|cx| Telemetry::new(clock.clone(), http, cx));
956        let worktree_id = 1;
957
958        // Scan of empty worktree finds nothing
959        test_project_discovery_helper(telemetry.clone(), vec![], Some(vec![]), worktree_id);
960
961        // Files added, second scan of worktree 1 finds project type
962        test_project_discovery_helper(
963            telemetry.clone(),
964            vec!["package.json"],
965            Some(vec!["node"]),
966            worktree_id,
967        );
968
969        // Third scan of worktree does not double report, as we already reported
970        test_project_discovery_helper(telemetry, vec!["package.json"], None, worktree_id);
971    }
972
973    #[gpui::test]
974    fn test_pnpm_project_discovery(cx: &mut gpui::TestAppContext) {
975        init_test(cx);
976
977        let clock = Arc::new(FakeSystemClock::new());
978        let http = FakeHttpClient::with_200_response();
979        let telemetry = cx.update(|cx| Telemetry::new(clock.clone(), http, cx));
980
981        test_project_discovery_helper(
982            telemetry,
983            vec!["package.json", "pnpm-lock.yaml"],
984            Some(vec!["node", "pnpm"]),
985            1,
986        );
987    }
988
989    #[gpui::test]
990    fn test_yarn_project_discovery(cx: &mut gpui::TestAppContext) {
991        init_test(cx);
992
993        let clock = Arc::new(FakeSystemClock::new());
994        let http = FakeHttpClient::with_200_response();
995        let telemetry = cx.update(|cx| Telemetry::new(clock.clone(), http, cx));
996
997        test_project_discovery_helper(
998            telemetry,
999            vec!["package.json", "yarn.lock"],
1000            Some(vec!["node", "yarn"]),
1001            1,
1002        );
1003    }
1004
1005    #[gpui::test]
1006    fn test_dotnet_project_discovery(cx: &mut gpui::TestAppContext) {
1007        init_test(cx);
1008
1009        let clock = Arc::new(FakeSystemClock::new());
1010        let http = FakeHttpClient::with_200_response();
1011        let telemetry = cx.update(|cx| Telemetry::new(clock.clone(), http, cx));
1012
1013        // Using different worktrees, as production code blocks from reporting a
1014        // project type for the same worktree multiple times
1015
1016        test_project_discovery_helper(
1017            telemetry.clone(),
1018            vec!["global.json"],
1019            Some(vec!["dotnet"]),
1020            1,
1021        );
1022        test_project_discovery_helper(
1023            telemetry.clone(),
1024            vec!["Directory.Build.props"],
1025            Some(vec!["dotnet"]),
1026            2,
1027        );
1028        test_project_discovery_helper(
1029            telemetry.clone(),
1030            vec!["file.csproj"],
1031            Some(vec!["dotnet"]),
1032            3,
1033        );
1034        test_project_discovery_helper(
1035            telemetry.clone(),
1036            vec!["file.fsproj"],
1037            Some(vec!["dotnet"]),
1038            4,
1039        );
1040        test_project_discovery_helper(
1041            telemetry.clone(),
1042            vec!["file.vbproj"],
1043            Some(vec!["dotnet"]),
1044            5,
1045        );
1046        test_project_discovery_helper(telemetry.clone(), vec!["file.sln"], Some(vec!["dotnet"]), 6);
1047
1048        // Each worktree should only send a single project type event, even when
1049        // encountering multiple files associated with that project type
1050        test_project_discovery_helper(
1051            telemetry,
1052            vec!["global.json", "Directory.Build.props"],
1053            Some(vec!["dotnet"]),
1054            7,
1055        );
1056    }
1057
1058    // TODO:
1059    // Test settings
1060    // Update FakeHTTPClient to keep track of the number of requests and assert on it
1061
1062    fn init_test(cx: &mut TestAppContext) {
1063        cx.update(|cx| {
1064            let settings_store = SettingsStore::test(cx);
1065            cx.set_global(settings_store);
1066        });
1067    }
1068
1069    fn is_empty_state(telemetry: &Telemetry) -> bool {
1070        telemetry.state.lock().events_queue.is_empty()
1071            && telemetry.state.lock().flush_events_task.is_none()
1072            && telemetry.state.lock().first_event_date_time.is_none()
1073    }
1074
1075    fn test_project_discovery_helper(
1076        telemetry: Arc<Telemetry>,
1077        file_paths: Vec<&str>,
1078        expected_project_types: Option<Vec<&str>>,
1079        worktree_id_num: usize,
1080    ) {
1081        let worktree_id = WorktreeId::from_usize(worktree_id_num);
1082        let entries: Vec<_> = file_paths
1083            .into_iter()
1084            .enumerate()
1085            .filter_map(|(i, path)| {
1086                Some((
1087                    Arc::from(RelPath::from_unix_str(path).ok()?),
1088                    ProjectEntryId::from_proto(i as u64 + 1),
1089                    PathChange::Added,
1090                ))
1091            })
1092            .collect();
1093        let updated_entries: UpdatedEntriesSet = Arc::from(entries.as_slice());
1094
1095        let detected_project_types = telemetry.detect_project_types(worktree_id, &updated_entries);
1096
1097        let expected_project_types =
1098            expected_project_types.map(|types| types.iter().map(|&t| t.to_string()).collect());
1099
1100        assert_eq!(detected_project_types, expected_project_types);
1101    }
1102}
1103