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