1use super::ips_file::IpsFile;
2use crate::api::CloudflareIpCountryHeader;
3use crate::{AppState, Error, Result, api::slack};
4use anyhow::anyhow;
5use aws_sdk_s3::primitives::ByteStream;
6use axum::{
7 Extension, Router, TypedHeader,
8 body::Bytes,
9 headers::Header,
10 http::{HeaderMap, HeaderName, StatusCode},
11 routing::post,
12};
13use chrono::Duration;
14use semantic_version::SemanticVersion;
15use serde::{Deserialize, Serialize};
16use serde_json::json;
17use sha2::{Digest, Sha256};
18use std::sync::{Arc, OnceLock};
19use telemetry_events::{Event, EventRequestBody, Panic};
20use util::ResultExt;
21use uuid::Uuid;
22
23const CRASH_REPORTS_BUCKET: &str = "zed-crash-reports";
24
25pub fn router() -> Router {
26 Router::new()
27 .route("/telemetry/events", post(post_events))
28 .route("/telemetry/crashes", post(post_crash))
29 .route("/telemetry/panics", post(post_panic))
30 .route("/telemetry/hangs", post(post_hang))
31}
32
33pub struct ZedChecksumHeader(Vec<u8>);
34
35impl Header for ZedChecksumHeader {
36 fn name() -> &'static HeaderName {
37 static ZED_CHECKSUM_HEADER: OnceLock<HeaderName> = OnceLock::new();
38 ZED_CHECKSUM_HEADER.get_or_init(|| HeaderName::from_static("x-zed-checksum"))
39 }
40
41 fn decode<'i, I>(values: &mut I) -> Result<Self, axum::headers::Error>
42 where
43 Self: Sized,
44 I: Iterator<Item = &'i axum::http::HeaderValue>,
45 {
46 let checksum = values
47 .next()
48 .ok_or_else(axum::headers::Error::invalid)?
49 .to_str()
50 .map_err(|_| axum::headers::Error::invalid())?;
51
52 let bytes = hex::decode(checksum).map_err(|_| axum::headers::Error::invalid())?;
53 Ok(Self(bytes))
54 }
55
56 fn encode<E: Extend<axum::http::HeaderValue>>(&self, _values: &mut E) {
57 unimplemented!()
58 }
59}
60
61pub async fn post_crash(
62 Extension(app): Extension<Arc<AppState>>,
63 headers: HeaderMap,
64 body: Bytes,
65) -> Result<()> {
66 let report = IpsFile::parse(&body)?;
67 let version_threshold = SemanticVersion::new(0, 123, 0);
68
69 let bundle_id = &report.header.bundle_id;
70 let app_version = &report.app_version();
71
72 if bundle_id == "dev.zed.Zed-Dev" {
73 log::error!("Crash uploads from {} are ignored.", bundle_id);
74 return Ok(());
75 }
76
77 if app_version.is_none() || app_version.unwrap() < version_threshold {
78 log::error!(
79 "Crash uploads from {} are ignored.",
80 report.header.app_version
81 );
82 return Ok(());
83 }
84 let app_version = app_version.unwrap();
85
86 if let Some(blob_store_client) = app.blob_store_client.as_ref() {
87 let response = blob_store_client
88 .head_object()
89 .bucket(CRASH_REPORTS_BUCKET)
90 .key(report.header.incident_id.clone() + ".ips")
91 .send()
92 .await;
93
94 if response.is_ok() {
95 log::info!("We've already uploaded this crash");
96 return Ok(());
97 }
98
99 blob_store_client
100 .put_object()
101 .bucket(CRASH_REPORTS_BUCKET)
102 .key(report.header.incident_id.clone() + ".ips")
103 .acl(aws_sdk_s3::types::ObjectCannedAcl::PublicRead)
104 .body(ByteStream::from(body.to_vec()))
105 .send()
106 .await
107 .map_err(|e| log::error!("Failed to upload crash: {}", e))
108 .ok();
109 }
110
111 let recent_panic_on: Option<i64> = headers
112 .get("x-zed-panicked-on")
113 .and_then(|h| h.to_str().ok())
114 .and_then(|s| s.parse().ok());
115
116 let installation_id = headers
117 .get("x-zed-installation-id")
118 .and_then(|h| h.to_str().ok())
119 .map(|s| s.to_string())
120 .unwrap_or_default();
121
122 let mut recent_panic = None;
123
124 if let Some(recent_panic_on) = recent_panic_on {
125 let crashed_at = match report.timestamp() {
126 Ok(t) => Some(t),
127 Err(e) => {
128 log::error!("Can't parse {}: {}", report.header.timestamp, e);
129 None
130 }
131 };
132 if crashed_at.is_some_and(|t| (t.timestamp_millis() - recent_panic_on).abs() <= 30000) {
133 recent_panic = headers.get("x-zed-panic").and_then(|h| h.to_str().ok());
134 }
135 }
136
137 let description = report.description(recent_panic);
138 let summary = report.backtrace_summary();
139
140 tracing::error!(
141 service = "client",
142 version = %report.header.app_version,
143 os_version = %report.header.os_version,
144 bundle_id = %report.header.bundle_id,
145 incident_id = %report.header.incident_id,
146 installation_id = %installation_id,
147 description = %description,
148 backtrace = %summary,
149 "crash report"
150 );
151
152 if let Some(kinesis_client) = app.kinesis_client.clone() {
153 if let Some(stream) = app.config.kinesis_stream.clone() {
154 let properties = json!({
155 "app_version": report.header.app_version,
156 "os_version": report.header.os_version,
157 "os_name": "macOS",
158 "bundle_id": report.header.bundle_id,
159 "incident_id": report.header.incident_id,
160 "installation_id": installation_id,
161 "description": description,
162 "backtrace": summary,
163 });
164 SnowflakeRow::new(
165 "Crash Report",
166 None,
167 false,
168 Some(installation_id),
169 properties,
170 )
171 .write(&Some(kinesis_client), &Some(stream))
172 .await
173 .log_err();
174 }
175 }
176
177 if let Some(slack_panics_webhook) = app.config.slack_panics_webhook.clone() {
178 let payload = slack::WebhookBody::new(|w| {
179 w.add_section(|s| s.text(slack::Text::markdown(description)))
180 .add_section(|s| {
181 s.add_field(slack::Text::markdown(format!(
182 "*Version:*\n{} ({})",
183 bundle_id, app_version
184 )))
185 .add_field({
186 let hostname = app.config.blob_store_url.clone().unwrap_or_default();
187 let hostname = hostname.strip_prefix("https://").unwrap_or_else(|| {
188 hostname.strip_prefix("http://").unwrap_or_default()
189 });
190
191 slack::Text::markdown(format!(
192 "*Incident:*\n<https://{}.{}/{}.ips|{}…>",
193 CRASH_REPORTS_BUCKET,
194 hostname,
195 report.header.incident_id,
196 report
197 .header
198 .incident_id
199 .chars()
200 .take(8)
201 .collect::<String>(),
202 ))
203 })
204 })
205 .add_rich_text(|r| r.add_preformatted(|p| p.add_text(summary)))
206 });
207 let payload_json = serde_json::to_string(&payload).map_err(|err| {
208 log::error!("Failed to serialize payload to JSON: {err}");
209 Error::Internal(anyhow!(err))
210 })?;
211
212 reqwest::Client::new()
213 .post(slack_panics_webhook)
214 .header("Content-Type", "application/json")
215 .body(payload_json)
216 .send()
217 .await
218 .map_err(|err| {
219 log::error!("Failed to send payload to Slack: {err}");
220 Error::Internal(anyhow!(err))
221 })?;
222 }
223
224 Ok(())
225}
226
227pub async fn post_hang(
228 Extension(app): Extension<Arc<AppState>>,
229 TypedHeader(ZedChecksumHeader(checksum)): TypedHeader<ZedChecksumHeader>,
230 body: Bytes,
231) -> Result<()> {
232 let Some(expected) = calculate_json_checksum(app.clone(), &body) else {
233 return Err(Error::http(
234 StatusCode::INTERNAL_SERVER_ERROR,
235 "events not enabled".into(),
236 ))?;
237 };
238
239 if checksum != expected {
240 return Err(Error::http(
241 StatusCode::BAD_REQUEST,
242 "invalid checksum".into(),
243 ))?;
244 }
245
246 let incident_id = Uuid::new_v4().to_string();
247
248 // dump JSON into S3 so we can get frame offsets if we need to.
249 if let Some(blob_store_client) = app.blob_store_client.as_ref() {
250 blob_store_client
251 .put_object()
252 .bucket(CRASH_REPORTS_BUCKET)
253 .key(incident_id.clone() + ".hang.json")
254 .acl(aws_sdk_s3::types::ObjectCannedAcl::PublicRead)
255 .body(ByteStream::from(body.to_vec()))
256 .send()
257 .await
258 .map_err(|e| log::error!("Failed to upload crash: {}", e))
259 .ok();
260 }
261
262 let report: telemetry_events::HangReport = serde_json::from_slice(&body).map_err(|err| {
263 log::error!("can't parse report json: {err}");
264 Error::Internal(anyhow!(err))
265 })?;
266
267 let mut backtrace = "Possible hang detected on main thread:".to_string();
268 let unknown = "<unknown>".to_string();
269 for frame in report.backtrace.iter() {
270 backtrace.push_str(&format!("\n{}", frame.symbols.first().unwrap_or(&unknown)));
271 }
272
273 tracing::error!(
274 service = "client",
275 version = %report.app_version.unwrap_or_default().to_string(),
276 os_name = %report.os_name,
277 os_version = report.os_version.unwrap_or_default().to_string(),
278 incident_id = %incident_id,
279 installation_id = %report.installation_id.unwrap_or_default(),
280 backtrace = %backtrace,
281 "hang report");
282
283 Ok(())
284}
285
286pub async fn post_panic(
287 Extension(app): Extension<Arc<AppState>>,
288 TypedHeader(ZedChecksumHeader(checksum)): TypedHeader<ZedChecksumHeader>,
289 body: Bytes,
290) -> Result<()> {
291 let Some(expected) = calculate_json_checksum(app.clone(), &body) else {
292 return Err(Error::http(
293 StatusCode::INTERNAL_SERVER_ERROR,
294 "events not enabled".into(),
295 ))?;
296 };
297
298 if checksum != expected {
299 return Err(Error::http(
300 StatusCode::BAD_REQUEST,
301 "invalid checksum".into(),
302 ))?;
303 }
304
305 let report: telemetry_events::PanicRequest = serde_json::from_slice(&body)
306 .map_err(|_| Error::http(StatusCode::BAD_REQUEST, "invalid json".into()))?;
307 let incident_id = uuid::Uuid::new_v4().to_string();
308 let panic = report.panic;
309
310 if panic.os_name == "Linux" && panic.os_version == Some("1.0.0".to_string()) {
311 return Err(Error::http(
312 StatusCode::BAD_REQUEST,
313 "invalid os version".into(),
314 ))?;
315 }
316
317 if let Some(blob_store_client) = app.blob_store_client.as_ref() {
318 let response = blob_store_client
319 .head_object()
320 .bucket(CRASH_REPORTS_BUCKET)
321 .key(incident_id.clone() + ".json")
322 .send()
323 .await;
324
325 if response.is_ok() {
326 log::info!("We've already uploaded this crash");
327 return Ok(());
328 }
329
330 blob_store_client
331 .put_object()
332 .bucket(CRASH_REPORTS_BUCKET)
333 .key(incident_id.clone() + ".json")
334 .acl(aws_sdk_s3::types::ObjectCannedAcl::PublicRead)
335 .body(ByteStream::from(body.to_vec()))
336 .send()
337 .await
338 .map_err(|e| log::error!("Failed to upload crash: {}", e))
339 .ok();
340 }
341
342 let backtrace = panic.backtrace.join("\n");
343
344 tracing::error!(
345 service = "client",
346 version = %panic.app_version,
347 os_name = %panic.os_name,
348 os_version = %panic.os_version.clone().unwrap_or_default(),
349 incident_id = %incident_id,
350 installation_id = %panic.installation_id.clone().unwrap_or_default(),
351 description = %panic.payload,
352 backtrace = %backtrace,
353 "panic report"
354 );
355
356 if let Some(kinesis_client) = app.kinesis_client.clone() {
357 if let Some(stream) = app.config.kinesis_stream.clone() {
358 let properties = json!({
359 "app_version": panic.app_version,
360 "os_name": panic.os_name,
361 "os_version": panic.os_version,
362 "incident_id": incident_id,
363 "installation_id": panic.installation_id,
364 "description": panic.payload,
365 "backtrace": backtrace,
366 });
367 SnowflakeRow::new(
368 "Panic Report",
369 None,
370 false,
371 panic.installation_id.clone(),
372 properties,
373 )
374 .write(&Some(kinesis_client), &Some(stream))
375 .await
376 .log_err();
377 }
378 }
379
380 let backtrace = if panic.backtrace.len() > 25 {
381 let total = panic.backtrace.len();
382 format!(
383 "{}\n and {} more",
384 panic
385 .backtrace
386 .iter()
387 .take(20)
388 .cloned()
389 .collect::<Vec<_>>()
390 .join("\n"),
391 total - 20
392 )
393 } else {
394 panic.backtrace.join("\n")
395 };
396
397 if !report_to_slack(&panic) {
398 return Ok(());
399 }
400
401 let backtrace_with_summary = panic.payload + "\n" + &backtrace;
402
403 if let Some(slack_panics_webhook) = app.config.slack_panics_webhook.clone() {
404 let payload = slack::WebhookBody::new(|w| {
405 w.add_section(|s| s.text(slack::Text::markdown("Panic request".to_string())))
406 .add_section(|s| {
407 s.add_field(slack::Text::markdown(format!(
408 "*Version:*\n {} ",
409 panic.app_version
410 )))
411 .add_field({
412 let hostname = app.config.blob_store_url.clone().unwrap_or_default();
413 let hostname = hostname.strip_prefix("https://").unwrap_or_else(|| {
414 hostname.strip_prefix("http://").unwrap_or_default()
415 });
416
417 slack::Text::markdown(format!(
418 "*{} {}:*\n<https://{}.{}/{}.json|{}…>",
419 panic.os_name,
420 panic.os_version.unwrap_or_default(),
421 CRASH_REPORTS_BUCKET,
422 hostname,
423 incident_id,
424 incident_id.chars().take(8).collect::<String>(),
425 ))
426 })
427 })
428 .add_rich_text(|r| r.add_preformatted(|p| p.add_text(backtrace_with_summary)))
429 });
430 let payload_json = serde_json::to_string(&payload).map_err(|err| {
431 log::error!("Failed to serialize payload to JSON: {err}");
432 Error::Internal(anyhow!(err))
433 })?;
434
435 reqwest::Client::new()
436 .post(slack_panics_webhook)
437 .header("Content-Type", "application/json")
438 .body(payload_json)
439 .send()
440 .await
441 .map_err(|err| {
442 log::error!("Failed to send payload to Slack: {err}");
443 Error::Internal(anyhow!(err))
444 })?;
445 }
446
447 Ok(())
448}
449
450fn report_to_slack(panic: &Panic) -> bool {
451 // Panics on macOS should make their way to Slack as a crash report,
452 // so we don't need to send them a second time via this channel.
453 if panic.os_name == "macOS" {
454 return false;
455 }
456
457 if panic.payload.contains("ERROR_SURFACE_LOST_KHR") {
458 return false;
459 }
460
461 if panic.payload.contains("ERROR_INITIALIZATION_FAILED") {
462 return false;
463 }
464
465 if panic
466 .payload
467 .contains("GPU has crashed, and no debug information is available")
468 {
469 return false;
470 }
471
472 true
473}
474
475pub async fn post_events(
476 Extension(app): Extension<Arc<AppState>>,
477 TypedHeader(ZedChecksumHeader(checksum)): TypedHeader<ZedChecksumHeader>,
478 country_code_header: Option<TypedHeader<CloudflareIpCountryHeader>>,
479 body: Bytes,
480) -> Result<()> {
481 let Some(expected) = calculate_json_checksum(app.clone(), &body) else {
482 return Err(Error::http(
483 StatusCode::INTERNAL_SERVER_ERROR,
484 "events not enabled".into(),
485 ))?;
486 };
487
488 let checksum_matched = checksum == expected;
489
490 let request_body: telemetry_events::EventRequestBody =
491 serde_json::from_slice(&body).map_err(|err| {
492 log::error!("can't parse event json: {err}");
493 Error::Internal(anyhow!(err))
494 })?;
495
496 let Some(last_event) = request_body.events.last() else {
497 return Err(Error::http(StatusCode::BAD_REQUEST, "no events".into()))?;
498 };
499 let country_code = country_code_header.map(|h| h.to_string());
500
501 let first_event_at = chrono::Utc::now()
502 - chrono::Duration::milliseconds(last_event.milliseconds_since_first_event);
503
504 if let Some(kinesis_client) = app.kinesis_client.clone() {
505 if let Some(stream) = app.config.kinesis_stream.clone() {
506 let mut request = kinesis_client.put_records().stream_name(stream);
507 for row in for_snowflake(
508 request_body.clone(),
509 first_event_at,
510 country_code.clone(),
511 checksum_matched,
512 ) {
513 if let Some(data) = serde_json::to_vec(&row).log_err() {
514 request = request.records(
515 aws_sdk_kinesis::types::PutRecordsRequestEntry::builder()
516 .partition_key(request_body.system_id.clone().unwrap_or_default())
517 .data(data.into())
518 .build()
519 .unwrap(),
520 );
521 }
522 }
523 request.send().await.log_err();
524 }
525 };
526
527 Ok(())
528}
529
530pub fn calculate_json_checksum(app: Arc<AppState>, json: &impl AsRef<[u8]>) -> Option<Vec<u8>> {
531 let checksum_seed = app.config.zed_client_checksum_seed.as_ref()?;
532
533 let mut summer = Sha256::new();
534 summer.update(checksum_seed);
535 summer.update(json);
536 summer.update(checksum_seed);
537 Some(summer.finalize().into_iter().collect())
538}
539
540fn for_snowflake(
541 body: EventRequestBody,
542 first_event_at: chrono::DateTime<chrono::Utc>,
543 country_code: Option<String>,
544 checksum_matched: bool,
545) -> impl Iterator<Item = SnowflakeRow> {
546 body.events.into_iter().flat_map(move |event| {
547 let timestamp =
548 first_event_at + Duration::milliseconds(event.milliseconds_since_first_event);
549 // We will need to double check, but I believe all of the events that
550 // are being transformed here are now migrated over to use the
551 // telemetry::event! macro, as of this commit so this code can go away
552 // when we feel enough users have upgraded past this point.
553 let (event_type, mut event_properties) = match &event.event {
554 Event::Editor(e) => (
555 match e.operation.as_str() {
556 "open" => "Editor Opened".to_string(),
557 "save" => "Editor Saved".to_string(),
558 _ => format!("Unknown Editor Event: {}", e.operation),
559 },
560 serde_json::to_value(e).unwrap(),
561 ),
562 Event::InlineCompletion(e) => (
563 format!(
564 "Edit Prediction {}",
565 if e.suggestion_accepted {
566 "Accepted"
567 } else {
568 "Discarded"
569 }
570 ),
571 serde_json::to_value(e).unwrap(),
572 ),
573 Event::InlineCompletionRating(e) => (
574 "Edit Prediction Rated".to_string(),
575 serde_json::to_value(e).unwrap(),
576 ),
577 Event::Call(e) => {
578 let event_type = match e.operation.trim() {
579 "unshare project" => "Project Unshared".to_string(),
580 "open channel notes" => "Channel Notes Opened".to_string(),
581 "share project" => "Project Shared".to_string(),
582 "join channel" => "Channel Joined".to_string(),
583 "hang up" => "Call Ended".to_string(),
584 "accept incoming" => "Incoming Call Accepted".to_string(),
585 "invite" => "Participant Invited".to_string(),
586 "disable microphone" => "Microphone Disabled".to_string(),
587 "enable microphone" => "Microphone Enabled".to_string(),
588 "enable screen share" => "Screen Share Enabled".to_string(),
589 "disable screen share" => "Screen Share Disabled".to_string(),
590 "decline incoming" => "Incoming Call Declined".to_string(),
591 _ => format!("Unknown Call Event: {}", e.operation),
592 };
593
594 (event_type, serde_json::to_value(e).unwrap())
595 }
596 Event::Assistant(e) => (
597 match e.phase {
598 telemetry_events::AssistantPhase::Response => "Assistant Responded".to_string(),
599 telemetry_events::AssistantPhase::Invoked => "Assistant Invoked".to_string(),
600 telemetry_events::AssistantPhase::Accepted => {
601 "Assistant Response Accepted".to_string()
602 }
603 telemetry_events::AssistantPhase::Rejected => {
604 "Assistant Response Rejected".to_string()
605 }
606 },
607 serde_json::to_value(e).unwrap(),
608 ),
609 Event::Cpu(_) | Event::Memory(_) => return None,
610 Event::App(e) => {
611 let mut properties = json!({});
612 let event_type = match e.operation.trim() {
613 // App
614 "open" => "App Opened".to_string(),
615 "first open" => "App First Opened".to_string(),
616 "first open for release channel" => {
617 "App First Opened For Release Channel".to_string()
618 }
619 "close" => "App Closed".to_string(),
620
621 // Project
622 "open project" => "Project Opened".to_string(),
623 "open node project" => {
624 properties["project_type"] = json!("node");
625 "Project Opened".to_string()
626 }
627 "open pnpm project" => {
628 properties["project_type"] = json!("pnpm");
629 "Project Opened".to_string()
630 }
631 "open yarn project" => {
632 properties["project_type"] = json!("yarn");
633 "Project Opened".to_string()
634 }
635
636 // SSH
637 "create ssh server" => "SSH Server Created".to_string(),
638 "create ssh project" => "SSH Project Created".to_string(),
639 "open ssh project" => "SSH Project Opened".to_string(),
640
641 // Welcome Page
642 "welcome page: change keymap" => "Welcome Keymap Changed".to_string(),
643 "welcome page: change theme" => "Welcome Theme Changed".to_string(),
644 "welcome page: close" => "Welcome Page Closed".to_string(),
645 "welcome page: edit settings" => "Welcome Settings Edited".to_string(),
646 "welcome page: install cli" => "Welcome CLI Installed".to_string(),
647 "welcome page: open" => "Welcome Page Opened".to_string(),
648 "welcome page: open extensions" => "Welcome Extensions Page Opened".to_string(),
649 "welcome page: sign in to copilot" => "Welcome Copilot Signed In".to_string(),
650 "welcome page: toggle diagnostic telemetry" => {
651 "Welcome Diagnostic Telemetry Toggled".to_string()
652 }
653 "welcome page: toggle metric telemetry" => {
654 "Welcome Metric Telemetry Toggled".to_string()
655 }
656 "welcome page: toggle vim" => "Welcome Vim Mode Toggled".to_string(),
657 "welcome page: view docs" => "Welcome Documentation Viewed".to_string(),
658
659 // Extensions
660 "extensions page: open" => "Extensions Page Opened".to_string(),
661 "extensions: install extension" => "Extension Installed".to_string(),
662 "extensions: uninstall extension" => "Extension Uninstalled".to_string(),
663
664 // Misc
665 "markdown preview: open" => "Markdown Preview Opened".to_string(),
666 "project diagnostics: open" => "Project Diagnostics Opened".to_string(),
667 "project search: open" => "Project Search Opened".to_string(),
668 "repl sessions: open" => "REPL Session Started".to_string(),
669
670 // Feature Upsell
671 "feature upsell: toggle vim" => {
672 properties["source"] = json!("Feature Upsell");
673 "Vim Mode Toggled".to_string()
674 }
675 _ => e
676 .operation
677 .strip_prefix("feature upsell: viewed docs (")
678 .and_then(|s| s.strip_suffix(')'))
679 .map_or_else(
680 || format!("Unknown App Event: {}", e.operation),
681 |docs_url| {
682 properties["url"] = json!(docs_url);
683 properties["source"] = json!("Feature Upsell");
684 "Documentation Viewed".to_string()
685 },
686 ),
687 };
688 (event_type, properties)
689 }
690 Event::Setting(e) => (
691 "Settings Changed".to_string(),
692 serde_json::to_value(e).unwrap(),
693 ),
694 Event::Extension(e) => (
695 "Extension Loaded".to_string(),
696 serde_json::to_value(e).unwrap(),
697 ),
698 Event::Edit(e) => (
699 "Editor Edited".to_string(),
700 serde_json::to_value(e).unwrap(),
701 ),
702 Event::Action(e) => (
703 "Action Invoked".to_string(),
704 serde_json::to_value(e).unwrap(),
705 ),
706 Event::Repl(e) => (
707 "Kernel Status Changed".to_string(),
708 serde_json::to_value(e).unwrap(),
709 ),
710 Event::Flexible(e) => (
711 e.event_type.clone(),
712 serde_json::to_value(&e.event_properties).unwrap(),
713 ),
714 };
715
716 if let serde_json::Value::Object(ref mut map) = event_properties {
717 map.insert("app_version".to_string(), body.app_version.clone().into());
718 map.insert("os_name".to_string(), body.os_name.clone().into());
719 map.insert("os_version".to_string(), body.os_version.clone().into());
720 map.insert("architecture".to_string(), body.architecture.clone().into());
721 map.insert(
722 "release_channel".to_string(),
723 body.release_channel.clone().into(),
724 );
725 map.insert("signed_in".to_string(), event.signed_in.into());
726 map.insert("checksum_matched".to_string(), checksum_matched.into());
727 if let Some(country_code) = country_code.as_ref() {
728 map.insert("country".to_string(), country_code.clone().into());
729 }
730 }
731
732 // NOTE: most amplitude user properties are read out of our event_properties
733 // dictionary. See https://app.amplitude.com/data/zed/Zed/sources/detail/production/falcon%3A159998
734 // for how that is configured.
735 let user_properties = Some(serde_json::json!({
736 "is_staff": body.is_staff,
737 }));
738
739 Some(SnowflakeRow {
740 time: timestamp,
741 user_id: body.metrics_id.clone(),
742 device_id: body.system_id.clone(),
743 event_type,
744 event_properties,
745 user_properties,
746 insert_id: Some(Uuid::new_v4().to_string()),
747 })
748 })
749}
750
751#[derive(Serialize, Deserialize, Debug)]
752pub struct SnowflakeRow {
753 pub time: chrono::DateTime<chrono::Utc>,
754 pub user_id: Option<String>,
755 pub device_id: Option<String>,
756 pub event_type: String,
757 pub event_properties: serde_json::Value,
758 pub user_properties: Option<serde_json::Value>,
759 pub insert_id: Option<String>,
760}
761
762impl SnowflakeRow {
763 pub fn new(
764 event_type: impl Into<String>,
765 metrics_id: Option<Uuid>,
766 is_staff: bool,
767 system_id: Option<String>,
768 event_properties: serde_json::Value,
769 ) -> Self {
770 Self {
771 time: chrono::Utc::now(),
772 event_type: event_type.into(),
773 device_id: system_id,
774 user_id: metrics_id.map(|id| id.to_string()),
775 insert_id: Some(uuid::Uuid::new_v4().to_string()),
776 event_properties,
777 user_properties: Some(json!({"is_staff": is_staff})),
778 }
779 }
780
781 pub async fn write(
782 self,
783 client: &Option<aws_sdk_kinesis::Client>,
784 stream: &Option<String>,
785 ) -> anyhow::Result<()> {
786 let Some((client, stream)) = client.as_ref().zip(stream.as_ref()) else {
787 return Ok(());
788 };
789 let row = serde_json::to_vec(&self)?;
790 client
791 .put_record()
792 .stream_name(stream)
793 .partition_key(&self.user_id.unwrap_or_default())
794 .data(row.into())
795 .send()
796 .await?;
797 Ok(())
798 }
799}