1use crate::http::HttpClient;
2use db::kvp::KEY_VALUE_STORE;
3use gpui::{
4 executor::Background,
5 serde_json::{self, value::Map, Value},
6 AppContext, Task,
7};
8use isahc::Request;
9use lazy_static::lazy_static;
10use parking_lot::Mutex;
11use serde::Serialize;
12use serde_json::json;
13use std::{
14 io::Write,
15 mem,
16 path::PathBuf,
17 sync::Arc,
18 time::{Duration, SystemTime, UNIX_EPOCH},
19};
20use tempfile::NamedTempFile;
21use util::{channel::ReleaseChannel, post_inc, ResultExt, TryFutureExt};
22use uuid::Uuid;
23
24pub struct Telemetry {
25 http_client: Arc<dyn HttpClient>,
26 executor: Arc<Background>,
27 state: Mutex<TelemetryState>,
28}
29
30#[derive(Default)]
31struct TelemetryState {
32 metrics_id: Option<Arc<str>>,
33 device_id: Option<Arc<str>>,
34 app_version: Option<Arc<str>>,
35 release_channel: Option<&'static str>,
36 os_version: Option<Arc<str>>,
37 os_name: &'static str,
38 queue: Vec<MixpanelEvent>,
39 next_event_id: usize,
40 flush_task: Option<Task<()>>,
41 log_file: Option<NamedTempFile>,
42}
43
44const MIXPANEL_EVENTS_URL: &'static str = "https://api.mixpanel.com/track";
45const MIXPANEL_ENGAGE_URL: &'static str = "https://api.mixpanel.com/engage#profile-set";
46
47lazy_static! {
48 static ref MIXPANEL_TOKEN: Option<String> = std::env::var("ZED_MIXPANEL_TOKEN")
49 .ok()
50 .or_else(|| option_env!("ZED_MIXPANEL_TOKEN").map(|key| key.to_string()));
51}
52
53#[derive(Serialize, Debug)]
54struct MixpanelEvent {
55 event: String,
56 properties: MixpanelEventProperties,
57}
58
59#[derive(Serialize, Debug)]
60struct MixpanelEventProperties {
61 // Mixpanel required fields
62 #[serde(skip_serializing_if = "str::is_empty")]
63 token: &'static str,
64 time: u128,
65 distinct_id: Option<Arc<str>>,
66 #[serde(rename = "$insert_id")]
67 insert_id: usize,
68 // Custom fields
69 #[serde(skip_serializing_if = "Option::is_none", flatten)]
70 event_properties: Option<Map<String, Value>>,
71 #[serde(rename = "OS Name")]
72 os_name: &'static str,
73 #[serde(rename = "OS Version")]
74 os_version: Option<Arc<str>>,
75 #[serde(rename = "Release Channel")]
76 release_channel: Option<&'static str>,
77 #[serde(rename = "App Version")]
78 app_version: Option<Arc<str>>,
79 #[serde(rename = "Signed In")]
80 signed_in: bool,
81}
82
83#[derive(Serialize)]
84struct MixpanelEngageRequest {
85 #[serde(rename = "$token")]
86 token: &'static str,
87 #[serde(rename = "$distinct_id")]
88 distinct_id: Arc<str>,
89 #[serde(rename = "$set")]
90 set: Value,
91}
92
93#[cfg(debug_assertions)]
94const MAX_QUEUE_LEN: usize = 1;
95
96#[cfg(not(debug_assertions))]
97const MAX_QUEUE_LEN: usize = 10;
98
99#[cfg(debug_assertions)]
100const DEBOUNCE_INTERVAL: Duration = Duration::from_secs(1);
101
102#[cfg(not(debug_assertions))]
103const DEBOUNCE_INTERVAL: Duration = Duration::from_secs(30);
104
105impl Telemetry {
106 pub fn new(client: Arc<dyn HttpClient>, cx: &AppContext) -> Arc<Self> {
107 let platform = cx.platform();
108 let release_channel = if cx.has_global::<ReleaseChannel>() {
109 Some(cx.global::<ReleaseChannel>().display_name())
110 } else {
111 None
112 };
113 let this = Arc::new(Self {
114 http_client: client,
115 executor: cx.background().clone(),
116 state: Mutex::new(TelemetryState {
117 os_version: platform.os_version().ok().map(|v| v.to_string().into()),
118 os_name: platform.os_name().into(),
119 app_version: platform.app_version().ok().map(|v| v.to_string().into()),
120 release_channel,
121 device_id: None,
122 metrics_id: None,
123 queue: Default::default(),
124 flush_task: Default::default(),
125 next_event_id: 0,
126 log_file: None,
127 }),
128 });
129
130 if MIXPANEL_TOKEN.is_some() {
131 this.executor
132 .spawn({
133 let this = this.clone();
134 async move {
135 if let Some(tempfile) = NamedTempFile::new().log_err() {
136 this.state.lock().log_file = Some(tempfile);
137 }
138 }
139 })
140 .detach();
141 }
142
143 this
144 }
145
146 pub fn log_file_path(&self) -> Option<PathBuf> {
147 Some(self.state.lock().log_file.as_ref()?.path().to_path_buf())
148 }
149
150 pub fn start(self: &Arc<Self>) {
151 let this = self.clone();
152 self.executor
153 .spawn(
154 async move {
155 let device_id =
156 if let Ok(Some(device_id)) = KEY_VALUE_STORE.read_kvp("device_id") {
157 device_id
158 } else {
159 let device_id = Uuid::new_v4().to_string();
160 KEY_VALUE_STORE
161 .write_kvp("device_id".to_string(), device_id.clone())
162 .await?;
163 device_id
164 };
165
166 let device_id: Arc<str> = device_id.into();
167 let mut state = this.state.lock();
168 state.device_id = Some(device_id.clone());
169 for event in &mut state.queue {
170 event
171 .properties
172 .distinct_id
173 .get_or_insert_with(|| device_id.clone());
174 }
175 if !state.queue.is_empty() {
176 drop(state);
177 this.flush();
178 }
179
180 anyhow::Ok(())
181 }
182 .log_err(),
183 )
184 .detach();
185 }
186
187 pub fn set_authenticated_user_info(
188 self: &Arc<Self>,
189 metrics_id: Option<String>,
190 is_staff: bool,
191 ) {
192 let this = self.clone();
193 let mut state = self.state.lock();
194 let device_id = state.device_id.clone();
195 let metrics_id: Option<Arc<str>> = metrics_id.map(|id| id.into());
196 state.metrics_id = metrics_id.clone();
197 drop(state);
198
199 if let Some((token, device_id)) = MIXPANEL_TOKEN.as_ref().zip(device_id) {
200 self.executor
201 .spawn(
202 async move {
203 let json_bytes = serde_json::to_vec(&[MixpanelEngageRequest {
204 token,
205 distinct_id: device_id,
206 set: json!({
207 "Staff": is_staff,
208 "ID": metrics_id,
209 "App": true
210 }),
211 }])?;
212 let request = Request::post(MIXPANEL_ENGAGE_URL)
213 .header("Content-Type", "application/json")
214 .body(json_bytes.into())?;
215 this.http_client.send(request).await?;
216 Ok(())
217 }
218 .log_err(),
219 )
220 .detach();
221 }
222 }
223
224 pub fn report_event(self: &Arc<Self>, kind: &str, properties: Value) {
225 let mut state = self.state.lock();
226 let event = MixpanelEvent {
227 event: kind.to_string(),
228 properties: MixpanelEventProperties {
229 token: "",
230 time: SystemTime::now()
231 .duration_since(UNIX_EPOCH)
232 .unwrap()
233 .as_millis(),
234 distinct_id: state.device_id.clone(),
235 insert_id: post_inc(&mut state.next_event_id),
236 event_properties: if let Value::Object(properties) = properties {
237 Some(properties)
238 } else {
239 None
240 },
241 os_name: state.os_name,
242 os_version: state.os_version.clone(),
243 release_channel: state.release_channel,
244 app_version: state.app_version.clone(),
245 signed_in: state.metrics_id.is_some(),
246 },
247 };
248 state.queue.push(event);
249 if state.device_id.is_some() {
250 if state.queue.len() >= MAX_QUEUE_LEN {
251 drop(state);
252 self.flush();
253 } else {
254 let this = self.clone();
255 let executor = self.executor.clone();
256 state.flush_task = Some(self.executor.spawn(async move {
257 executor.timer(DEBOUNCE_INTERVAL).await;
258 this.flush();
259 }));
260 }
261 }
262 }
263
264 fn flush(self: &Arc<Self>) {
265 let mut state = self.state.lock();
266 let mut events = mem::take(&mut state.queue);
267 state.flush_task.take();
268 drop(state);
269
270 if let Some(token) = MIXPANEL_TOKEN.as_ref() {
271 let this = self.clone();
272 self.executor
273 .spawn(
274 async move {
275 let mut json_bytes = Vec::new();
276
277 if let Some(file) = &mut this.state.lock().log_file {
278 let file = file.as_file_mut();
279 for event in &mut events {
280 json_bytes.clear();
281 serde_json::to_writer(&mut json_bytes, event)?;
282 file.write_all(&json_bytes)?;
283 file.write(b"\n")?;
284
285 event.properties.token = token;
286 }
287 }
288
289 json_bytes.clear();
290 serde_json::to_writer(&mut json_bytes, &events)?;
291 let request = Request::post(MIXPANEL_EVENTS_URL)
292 .header("Content-Type", "application/json")
293 .body(json_bytes.into())?;
294 this.http_client.send(request).await?;
295 Ok(())
296 }
297 .log_err(),
298 )
299 .detach();
300 }
301 }
302}