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