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 std::{
14 io::Write,
15 mem,
16 path::PathBuf,
17 sync::Arc,
18 time::{Duration, SystemTime, UNIX_EPOCH},
19};
20use tempfile::NamedTempFile;
21use util::{post_inc, ResultExt, TryFutureExt};
22use uuid::Uuid;
23
24pub struct Telemetry {
25 http_client: Arc<dyn HttpClient>,
26 executor: Arc<Background>,
27 session_id: u128,
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 os_version: Option<Arc<str>>,
37 os_name: &'static str,
38 queue: Vec<AmplitudeEvent>,
39 next_event_id: usize,
40 flush_task: Option<Task<()>>,
41 log_file: Option<NamedTempFile>,
42}
43
44const AMPLITUDE_EVENTS_URL: &'static str = "https://api2.amplitude.com/batch";
45
46lazy_static! {
47 static ref AMPLITUDE_API_KEY: Option<String> = std::env::var("ZED_AMPLITUDE_API_KEY")
48 .ok()
49 .or_else(|| option_env!("ZED_AMPLITUDE_API_KEY").map(|key| key.to_string()));
50}
51
52#[derive(Serialize)]
53struct AmplitudeEventBatch {
54 api_key: &'static str,
55 events: Vec<AmplitudeEvent>,
56}
57
58#[derive(Serialize)]
59struct AmplitudeEvent {
60 #[serde(skip_serializing_if = "Option::is_none")]
61 user_id: Option<Arc<str>>,
62 device_id: Option<Arc<str>>,
63 event_type: String,
64 #[serde(skip_serializing_if = "Option::is_none")]
65 event_properties: Option<Map<String, Value>>,
66 #[serde(skip_serializing_if = "Option::is_none")]
67 user_properties: Option<Map<String, Value>>,
68 os_name: &'static str,
69 os_version: Option<Arc<str>>,
70 app_version: Option<Arc<str>>,
71 platform: &'static str,
72 event_id: usize,
73 session_id: u128,
74 time: u128,
75}
76
77#[cfg(debug_assertions)]
78const MAX_QUEUE_LEN: usize = 1;
79
80#[cfg(not(debug_assertions))]
81const MAX_QUEUE_LEN: usize = 10;
82
83#[cfg(debug_assertions)]
84const DEBOUNCE_INTERVAL: Duration = Duration::from_secs(1);
85
86#[cfg(not(debug_assertions))]
87const DEBOUNCE_INTERVAL: Duration = Duration::from_secs(30);
88
89impl Telemetry {
90 pub fn new(client: Arc<dyn HttpClient>, cx: &AppContext) -> Arc<Self> {
91 let platform = cx.platform();
92 let this = Arc::new(Self {
93 http_client: client,
94 executor: cx.background().clone(),
95 session_id: SystemTime::now()
96 .duration_since(UNIX_EPOCH)
97 .unwrap()
98 .as_millis(),
99 state: Mutex::new(TelemetryState {
100 os_version: platform
101 .os_version()
102 .log_err()
103 .map(|v| v.to_string().into()),
104 os_name: platform.os_name().into(),
105 app_version: platform
106 .app_version()
107 .log_err()
108 .map(|v| v.to_string().into()),
109 device_id: None,
110 queue: Default::default(),
111 flush_task: Default::default(),
112 next_event_id: 0,
113 log_file: None,
114 metrics_id: None,
115 }),
116 });
117
118 if AMPLITUDE_API_KEY.is_some() {
119 this.executor
120 .spawn({
121 let this = this.clone();
122 async move {
123 if let Some(tempfile) = NamedTempFile::new().log_err() {
124 this.state.lock().log_file = Some(tempfile);
125 }
126 }
127 })
128 .detach();
129 }
130
131 this
132 }
133
134 pub fn log_file_path(&self) -> Option<PathBuf> {
135 Some(self.state.lock().log_file.as_ref()?.path().to_path_buf())
136 }
137
138 pub fn start(self: &Arc<Self>, db: Arc<Db>) {
139 let this = self.clone();
140 self.executor
141 .spawn(
142 async move {
143 let device_id = if let Some(device_id) = db
144 .read(["device_id"])?
145 .into_iter()
146 .flatten()
147 .next()
148 .and_then(|bytes| String::from_utf8(bytes).ok())
149 {
150 device_id
151 } else {
152 let device_id = Uuid::new_v4().to_string();
153 db.write([("device_id", device_id.as_bytes())])?;
154 device_id
155 };
156
157 let device_id = Some(Arc::from(device_id));
158 let mut state = this.state.lock();
159 state.device_id = device_id.clone();
160 for event in &mut state.queue {
161 event.device_id = device_id.clone();
162 }
163 if !state.queue.is_empty() {
164 drop(state);
165 this.flush();
166 }
167
168 anyhow::Ok(())
169 }
170 .log_err(),
171 )
172 .detach();
173 }
174
175 pub fn set_authenticated_user_info(
176 self: &Arc<Self>,
177 metrics_id: Option<String>,
178 is_staff: bool,
179 ) {
180 let is_signed_in = metrics_id.is_some();
181 self.state.lock().metrics_id = metrics_id.map(|s| s.into());
182 if is_signed_in {
183 self.report_event_with_user_properties(
184 "$identify",
185 Default::default(),
186 json!({ "$set": { "staff": is_staff } }),
187 )
188 }
189 }
190
191 pub fn report_event(self: &Arc<Self>, kind: &str, properties: Value) {
192 self.report_event_with_user_properties(kind, properties, Default::default());
193 }
194
195 fn report_event_with_user_properties(
196 self: &Arc<Self>,
197 kind: &str,
198 properties: Value,
199 user_properties: Value,
200 ) {
201 if AMPLITUDE_API_KEY.is_none() {
202 return;
203 }
204
205 let mut state = self.state.lock();
206 let event = AmplitudeEvent {
207 event_type: kind.to_string(),
208 time: SystemTime::now()
209 .duration_since(UNIX_EPOCH)
210 .unwrap()
211 .as_millis(),
212 session_id: self.session_id,
213 event_properties: if let Value::Object(properties) = properties {
214 Some(properties)
215 } else {
216 None
217 },
218 user_properties: if let Value::Object(user_properties) = user_properties {
219 Some(user_properties)
220 } else {
221 None
222 },
223 user_id: state.metrics_id.clone(),
224 device_id: state.device_id.clone(),
225 os_name: state.os_name,
226 platform: "Zed",
227 os_version: state.os_version.clone(),
228 app_version: state.app_version.clone(),
229 event_id: post_inc(&mut state.next_event_id),
230 };
231 state.queue.push(event);
232 if state.device_id.is_some() {
233 if state.queue.len() >= MAX_QUEUE_LEN {
234 drop(state);
235 self.flush();
236 } else {
237 let this = self.clone();
238 let executor = self.executor.clone();
239 state.flush_task = Some(self.executor.spawn(async move {
240 executor.timer(DEBOUNCE_INTERVAL).await;
241 this.flush();
242 }));
243 }
244 }
245 }
246
247 fn flush(self: &Arc<Self>) {
248 let mut state = self.state.lock();
249 let events = mem::take(&mut state.queue);
250 state.flush_task.take();
251 drop(state);
252
253 if let Some(api_key) = AMPLITUDE_API_KEY.as_ref() {
254 let this = self.clone();
255 self.executor
256 .spawn(
257 async move {
258 let mut json_bytes = Vec::new();
259
260 if let Some(file) = &mut this.state.lock().log_file {
261 let file = file.as_file_mut();
262 for event in &events {
263 json_bytes.clear();
264 serde_json::to_writer(&mut json_bytes, event)?;
265 file.write_all(&json_bytes)?;
266 file.write(b"\n")?;
267 }
268 }
269
270 let batch = AmplitudeEventBatch { api_key, events };
271 json_bytes.clear();
272 serde_json::to_writer(&mut json_bytes, &batch)?;
273 let request =
274 Request::post(AMPLITUDE_EVENTS_URL).body(json_bytes.into())?;
275 this.http_client.send(request).await?;
276 Ok(())
277 }
278 .log_err(),
279 )
280 .detach();
281 }
282 }
283}