events.rs

   1use super::ips_file::IpsFile;
   2use crate::{api::slack, AppState, Error, Result};
   3use anyhow::{anyhow, Context};
   4use aws_sdk_s3::primitives::ByteStream;
   5use axum::{
   6    body::Bytes,
   7    headers::Header,
   8    http::{HeaderMap, HeaderName, StatusCode},
   9    routing::post,
  10    Extension, Router, TypedHeader,
  11};
  12use rpc::ExtensionMetadata;
  13use semantic_version::SemanticVersion;
  14use serde::{Serialize, Serializer};
  15use sha2::{Digest, Sha256};
  16use std::sync::{Arc, OnceLock};
  17use telemetry_events::{
  18    ActionEvent, AppEvent, AssistantEvent, CallEvent, CpuEvent, EditEvent, EditorEvent, Event,
  19    EventRequestBody, EventWrapper, ExtensionEvent, InlineCompletionEvent, MemoryEvent,
  20    SettingEvent,
  21};
  22use uuid::Uuid;
  23
  24static CRASH_REPORTS_BUCKET: &str = "zed-crash-reports";
  25
  26pub fn router() -> Router {
  27    Router::new()
  28        .route("/telemetry/events", post(post_events))
  29        .route("/telemetry/crashes", post(post_crash))
  30        .route("/telemetry/panics", post(post_panic))
  31        .route("/telemetry/hangs", post(post_hang))
  32}
  33
  34pub struct ZedChecksumHeader(Vec<u8>);
  35
  36impl Header for ZedChecksumHeader {
  37    fn name() -> &'static HeaderName {
  38        static ZED_CHECKSUM_HEADER: OnceLock<HeaderName> = OnceLock::new();
  39        ZED_CHECKSUM_HEADER.get_or_init(|| HeaderName::from_static("x-zed-checksum"))
  40    }
  41
  42    fn decode<'i, I>(values: &mut I) -> Result<Self, axum::headers::Error>
  43    where
  44        Self: Sized,
  45        I: Iterator<Item = &'i axum::http::HeaderValue>,
  46    {
  47        let checksum = values
  48            .next()
  49            .ok_or_else(axum::headers::Error::invalid)?
  50            .to_str()
  51            .map_err(|_| axum::headers::Error::invalid())?;
  52
  53        let bytes = hex::decode(checksum).map_err(|_| axum::headers::Error::invalid())?;
  54        Ok(Self(bytes))
  55    }
  56
  57    fn encode<E: Extend<axum::http::HeaderValue>>(&self, _values: &mut E) {
  58        unimplemented!()
  59    }
  60}
  61
  62pub struct CloudflareIpCountryHeader(String);
  63
  64impl Header for CloudflareIpCountryHeader {
  65    fn name() -> &'static HeaderName {
  66        static CLOUDFLARE_IP_COUNTRY_HEADER: OnceLock<HeaderName> = OnceLock::new();
  67        CLOUDFLARE_IP_COUNTRY_HEADER.get_or_init(|| HeaderName::from_static("cf-ipcountry"))
  68    }
  69
  70    fn decode<'i, I>(values: &mut I) -> Result<Self, axum::headers::Error>
  71    where
  72        Self: Sized,
  73        I: Iterator<Item = &'i axum::http::HeaderValue>,
  74    {
  75        let country_code = values
  76            .next()
  77            .ok_or_else(axum::headers::Error::invalid)?
  78            .to_str()
  79            .map_err(|_| axum::headers::Error::invalid())?;
  80
  81        Ok(Self(country_code.to_string()))
  82    }
  83
  84    fn encode<E: Extend<axum::http::HeaderValue>>(&self, _values: &mut E) {
  85        unimplemented!()
  86    }
  87}
  88
  89pub async fn post_crash(
  90    Extension(app): Extension<Arc<AppState>>,
  91    headers: HeaderMap,
  92    body: Bytes,
  93) -> Result<()> {
  94    let report = IpsFile::parse(&body)?;
  95    let version_threshold = SemanticVersion::new(0, 123, 0);
  96
  97    let bundle_id = &report.header.bundle_id;
  98    let app_version = &report.app_version();
  99
 100    if bundle_id == "dev.zed.Zed-Dev" {
 101        log::error!("Crash uploads from {} are ignored.", bundle_id);
 102        return Ok(());
 103    }
 104
 105    if app_version.is_none() || app_version.unwrap() < version_threshold {
 106        log::error!(
 107            "Crash uploads from {} are ignored.",
 108            report.header.app_version
 109        );
 110        return Ok(());
 111    }
 112    let app_version = app_version.unwrap();
 113
 114    if let Some(blob_store_client) = app.blob_store_client.as_ref() {
 115        let response = blob_store_client
 116            .head_object()
 117            .bucket(CRASH_REPORTS_BUCKET)
 118            .key(report.header.incident_id.clone() + ".ips")
 119            .send()
 120            .await;
 121
 122        if response.is_ok() {
 123            log::info!("We've already uploaded this crash");
 124            return Ok(());
 125        }
 126
 127        blob_store_client
 128            .put_object()
 129            .bucket(CRASH_REPORTS_BUCKET)
 130            .key(report.header.incident_id.clone() + ".ips")
 131            .acl(aws_sdk_s3::types::ObjectCannedAcl::PublicRead)
 132            .body(ByteStream::from(body.to_vec()))
 133            .send()
 134            .await
 135            .map_err(|e| log::error!("Failed to upload crash: {}", e))
 136            .ok();
 137    }
 138
 139    let recent_panic_on: Option<i64> = headers
 140        .get("x-zed-panicked-on")
 141        .and_then(|h| h.to_str().ok())
 142        .and_then(|s| s.parse().ok());
 143
 144    let installation_id = headers
 145        .get("x-zed-installation-id")
 146        .and_then(|h| h.to_str().ok())
 147        .map(|s| s.to_string())
 148        .unwrap_or_default();
 149
 150    let mut recent_panic = None;
 151
 152    if let Some(recent_panic_on) = recent_panic_on {
 153        let crashed_at = match report.timestamp() {
 154            Ok(t) => Some(t),
 155            Err(e) => {
 156                log::error!("Can't parse {}: {}", report.header.timestamp, e);
 157                None
 158            }
 159        };
 160        if crashed_at.is_some_and(|t| (t.timestamp_millis() - recent_panic_on).abs() <= 30000) {
 161            recent_panic = headers.get("x-zed-panic").and_then(|h| h.to_str().ok());
 162        }
 163    }
 164
 165    let description = report.description(recent_panic);
 166    let summary = report.backtrace_summary();
 167
 168    tracing::error!(
 169        service = "client",
 170        version = %report.header.app_version,
 171        os_version = %report.header.os_version,
 172        bundle_id = %report.header.bundle_id,
 173        incident_id = %report.header.incident_id,
 174        installation_id = %installation_id,
 175        description = %description,
 176        backtrace = %summary,
 177        "crash report");
 178
 179    if let Some(slack_panics_webhook) = app.config.slack_panics_webhook.clone() {
 180        let payload = slack::WebhookBody::new(|w| {
 181            w.add_section(|s| s.text(slack::Text::markdown(description)))
 182                .add_section(|s| {
 183                    s.add_field(slack::Text::markdown(format!(
 184                        "*Version:*\n{} ({})",
 185                        bundle_id, app_version
 186                    )))
 187                    .add_field({
 188                        let hostname = app.config.blob_store_url.clone().unwrap_or_default();
 189                        let hostname = hostname.strip_prefix("https://").unwrap_or_else(|| {
 190                            hostname.strip_prefix("http://").unwrap_or_default()
 191                        });
 192
 193                        slack::Text::markdown(format!(
 194                            "*Incident:*\n<https://{}.{}/{}.ips|{}…>",
 195                            CRASH_REPORTS_BUCKET,
 196                            hostname,
 197                            report.header.incident_id,
 198                            report
 199                                .header
 200                                .incident_id
 201                                .chars()
 202                                .take(8)
 203                                .collect::<String>(),
 204                        ))
 205                    })
 206                })
 207                .add_rich_text(|r| r.add_preformatted(|p| p.add_text(summary)))
 208        });
 209        let payload_json = serde_json::to_string(&payload).map_err(|err| {
 210            log::error!("Failed to serialize payload to JSON: {err}");
 211            Error::Internal(anyhow!(err))
 212        })?;
 213
 214        reqwest::Client::new()
 215            .post(slack_panics_webhook)
 216            .header("Content-Type", "application/json")
 217            .body(payload_json)
 218            .send()
 219            .await
 220            .map_err(|err| {
 221                log::error!("Failed to send payload to Slack: {err}");
 222                Error::Internal(anyhow!(err))
 223            })?;
 224    }
 225
 226    Ok(())
 227}
 228
 229pub async fn post_hang(
 230    Extension(app): Extension<Arc<AppState>>,
 231    TypedHeader(ZedChecksumHeader(checksum)): TypedHeader<ZedChecksumHeader>,
 232    body: Bytes,
 233) -> Result<()> {
 234    let Some(expected) = calculate_json_checksum(app.clone(), &body) else {
 235        return Err(Error::Http(
 236            StatusCode::INTERNAL_SERVER_ERROR,
 237            "events not enabled".into(),
 238        ))?;
 239    };
 240
 241    if checksum != expected {
 242        return Err(Error::Http(
 243            StatusCode::BAD_REQUEST,
 244            "invalid checksum".into(),
 245        ))?;
 246    }
 247
 248    let incident_id = Uuid::new_v4().to_string();
 249
 250    // dump JSON into S3 so we can get frame offsets if we need to.
 251    if let Some(blob_store_client) = app.blob_store_client.as_ref() {
 252        blob_store_client
 253            .put_object()
 254            .bucket(CRASH_REPORTS_BUCKET)
 255            .key(incident_id.clone() + ".hang.json")
 256            .acl(aws_sdk_s3::types::ObjectCannedAcl::PublicRead)
 257            .body(ByteStream::from(body.to_vec()))
 258            .send()
 259            .await
 260            .map_err(|e| log::error!("Failed to upload crash: {}", e))
 261            .ok();
 262    }
 263
 264    let report: telemetry_events::HangReport = serde_json::from_slice(&body).map_err(|err| {
 265        log::error!("can't parse report json: {err}");
 266        Error::Internal(anyhow!(err))
 267    })?;
 268
 269    let mut backtrace = "Possible hang detected on main thread:".to_string();
 270    let unknown = "<unknown>".to_string();
 271    for frame in report.backtrace.iter() {
 272        backtrace.push_str(&format!("\n{}", frame.symbols.first().unwrap_or(&unknown)));
 273    }
 274
 275    tracing::error!(
 276        service = "client",
 277        version = %report.app_version.unwrap_or_default().to_string(),
 278        os_name = %report.os_name,
 279        os_version = report.os_version.unwrap_or_default().to_string(),
 280        incident_id = %incident_id,
 281        installation_id = %report.installation_id.unwrap_or_default(),
 282        backtrace = %backtrace,
 283        "hang report");
 284
 285    Ok(())
 286}
 287
 288pub async fn post_panic(
 289    Extension(app): Extension<Arc<AppState>>,
 290    TypedHeader(ZedChecksumHeader(checksum)): TypedHeader<ZedChecksumHeader>,
 291    body: Bytes,
 292) -> Result<()> {
 293    let Some(expected) = calculate_json_checksum(app.clone(), &body) else {
 294        return Err(Error::Http(
 295            StatusCode::INTERNAL_SERVER_ERROR,
 296            "events not enabled".into(),
 297        ))?;
 298    };
 299
 300    if checksum != expected {
 301        return Err(Error::Http(
 302            StatusCode::BAD_REQUEST,
 303            "invalid checksum".into(),
 304        ))?;
 305    }
 306
 307    let report: telemetry_events::PanicRequest = serde_json::from_slice(&body)
 308        .map_err(|_| Error::Http(StatusCode::BAD_REQUEST, "invalid json".into()))?;
 309    let panic = report.panic;
 310
 311    if panic.os_name == "Linux" && panic.os_version == Some("1.0.0".to_string()) {
 312        return Err(Error::Http(
 313            StatusCode::BAD_REQUEST,
 314            "invalid os version".into(),
 315        ))?;
 316    }
 317
 318    tracing::error!(
 319        service = "client",
 320        version = %panic.app_version,
 321        os_name = %panic.os_name,
 322        os_version = %panic.os_version.clone().unwrap_or_default(),
 323        installation_id = %panic.installation_id.unwrap_or_default(),
 324        description = %panic.payload,
 325        backtrace = %panic.backtrace.join("\n"),
 326        "panic report");
 327
 328    let backtrace = if panic.backtrace.len() > 25 {
 329        let total = panic.backtrace.len();
 330        format!(
 331            "{}\n   and {} more",
 332            panic
 333                .backtrace
 334                .iter()
 335                .take(20)
 336                .cloned()
 337                .collect::<Vec<_>>()
 338                .join("\n"),
 339            total - 20
 340        )
 341    } else {
 342        panic.backtrace.join("\n")
 343    };
 344    let backtrace_with_summary = panic.payload + "\n" + &backtrace;
 345
 346    if let Some(slack_panics_webhook) = app.config.slack_panics_webhook.clone() {
 347        let payload = slack::WebhookBody::new(|w| {
 348            w.add_section(|s| s.text(slack::Text::markdown("Panic request".to_string())))
 349                .add_section(|s| {
 350                    s.add_field(slack::Text::markdown(format!(
 351                        "*Version:*\n {} ",
 352                        panic.app_version
 353                    )))
 354                    .add_field({
 355                        slack::Text::markdown(format!(
 356                            "*OS:*\n{} {}",
 357                            panic.os_name,
 358                            panic.os_version.unwrap_or_default()
 359                        ))
 360                    })
 361                })
 362                .add_rich_text(|r| r.add_preformatted(|p| p.add_text(backtrace_with_summary)))
 363        });
 364        let payload_json = serde_json::to_string(&payload).map_err(|err| {
 365            log::error!("Failed to serialize payload to JSON: {err}");
 366            Error::Internal(anyhow!(err))
 367        })?;
 368
 369        reqwest::Client::new()
 370            .post(slack_panics_webhook)
 371            .header("Content-Type", "application/json")
 372            .body(payload_json)
 373            .send()
 374            .await
 375            .map_err(|err| {
 376                log::error!("Failed to send payload to Slack: {err}");
 377                Error::Internal(anyhow!(err))
 378            })?;
 379    }
 380
 381    Ok(())
 382}
 383
 384pub async fn post_events(
 385    Extension(app): Extension<Arc<AppState>>,
 386    TypedHeader(ZedChecksumHeader(checksum)): TypedHeader<ZedChecksumHeader>,
 387    country_code_header: Option<TypedHeader<CloudflareIpCountryHeader>>,
 388    body: Bytes,
 389) -> Result<()> {
 390    let Some(clickhouse_client) = app.clickhouse_client.clone() else {
 391        Err(Error::Http(
 392            StatusCode::NOT_IMPLEMENTED,
 393            "not supported".into(),
 394        ))?
 395    };
 396
 397    let Some(expected) = calculate_json_checksum(app.clone(), &body) else {
 398        return Err(Error::Http(
 399            StatusCode::INTERNAL_SERVER_ERROR,
 400            "events not enabled".into(),
 401        ))?;
 402    };
 403
 404    let checksum_matched = checksum == expected;
 405
 406    let request_body: telemetry_events::EventRequestBody =
 407        serde_json::from_slice(&body).map_err(|err| {
 408            log::error!("can't parse event json: {err}");
 409            Error::Internal(anyhow!(err))
 410        })?;
 411
 412    let mut to_upload = ToUpload::default();
 413    let Some(last_event) = request_body.events.last() else {
 414        return Err(Error::Http(StatusCode::BAD_REQUEST, "no events".into()))?;
 415    };
 416    let country_code = country_code_header.map(|h| h.0 .0);
 417
 418    let first_event_at = chrono::Utc::now()
 419        - chrono::Duration::milliseconds(last_event.milliseconds_since_first_event);
 420
 421    for wrapper in &request_body.events {
 422        match &wrapper.event {
 423            Event::Editor(event) => to_upload.editor_events.push(EditorEventRow::from_event(
 424                event.clone(),
 425                &wrapper,
 426                &request_body,
 427                first_event_at,
 428                country_code.clone(),
 429                checksum_matched,
 430            )),
 431            // Needed for clients sending old copilot_event types
 432            Event::Copilot(_) => {}
 433            Event::InlineCompletion(event) => {
 434                to_upload
 435                    .inline_completion_events
 436                    .push(InlineCompletionEventRow::from_event(
 437                        event.clone(),
 438                        &wrapper,
 439                        &request_body,
 440                        first_event_at,
 441                        country_code.clone(),
 442                        checksum_matched,
 443                    ))
 444            }
 445            Event::Call(event) => to_upload.call_events.push(CallEventRow::from_event(
 446                event.clone(),
 447                &wrapper,
 448                &request_body,
 449                first_event_at,
 450                checksum_matched,
 451            )),
 452            Event::Assistant(event) => {
 453                to_upload
 454                    .assistant_events
 455                    .push(AssistantEventRow::from_event(
 456                        event.clone(),
 457                        &wrapper,
 458                        &request_body,
 459                        first_event_at,
 460                        checksum_matched,
 461                    ))
 462            }
 463            Event::Cpu(event) => to_upload.cpu_events.push(CpuEventRow::from_event(
 464                event.clone(),
 465                &wrapper,
 466                &request_body,
 467                first_event_at,
 468                checksum_matched,
 469            )),
 470            Event::Memory(event) => to_upload.memory_events.push(MemoryEventRow::from_event(
 471                event.clone(),
 472                &wrapper,
 473                &request_body,
 474                first_event_at,
 475                checksum_matched,
 476            )),
 477            Event::App(event) => to_upload.app_events.push(AppEventRow::from_event(
 478                event.clone(),
 479                &wrapper,
 480                &request_body,
 481                first_event_at,
 482                checksum_matched,
 483            )),
 484            Event::Setting(event) => to_upload.setting_events.push(SettingEventRow::from_event(
 485                event.clone(),
 486                &wrapper,
 487                &request_body,
 488                first_event_at,
 489                checksum_matched,
 490            )),
 491            Event::Edit(event) => to_upload.edit_events.push(EditEventRow::from_event(
 492                event.clone(),
 493                &wrapper,
 494                &request_body,
 495                first_event_at,
 496                checksum_matched,
 497            )),
 498            Event::Action(event) => to_upload.action_events.push(ActionEventRow::from_event(
 499                event.clone(),
 500                &wrapper,
 501                &request_body,
 502                first_event_at,
 503                checksum_matched,
 504            )),
 505            Event::Extension(event) => {
 506                let metadata = app
 507                    .db
 508                    .get_extension_version(&event.extension_id, &event.version)
 509                    .await?;
 510                to_upload
 511                    .extension_events
 512                    .push(ExtensionEventRow::from_event(
 513                        event.clone(),
 514                        &wrapper,
 515                        &request_body,
 516                        metadata,
 517                        first_event_at,
 518                        checksum_matched,
 519                    ))
 520            }
 521        }
 522    }
 523
 524    to_upload
 525        .upload(&clickhouse_client)
 526        .await
 527        .map_err(|err| Error::Internal(anyhow!(err)))?;
 528
 529    Ok(())
 530}
 531
 532#[derive(Default)]
 533struct ToUpload {
 534    editor_events: Vec<EditorEventRow>,
 535    inline_completion_events: Vec<InlineCompletionEventRow>,
 536    assistant_events: Vec<AssistantEventRow>,
 537    call_events: Vec<CallEventRow>,
 538    cpu_events: Vec<CpuEventRow>,
 539    memory_events: Vec<MemoryEventRow>,
 540    app_events: Vec<AppEventRow>,
 541    setting_events: Vec<SettingEventRow>,
 542    extension_events: Vec<ExtensionEventRow>,
 543    edit_events: Vec<EditEventRow>,
 544    action_events: Vec<ActionEventRow>,
 545}
 546
 547impl ToUpload {
 548    pub async fn upload(&self, clickhouse_client: &clickhouse::Client) -> anyhow::Result<()> {
 549        const EDITOR_EVENTS_TABLE: &str = "editor_events";
 550        Self::upload_to_table(EDITOR_EVENTS_TABLE, &self.editor_events, clickhouse_client)
 551            .await
 552            .with_context(|| format!("failed to upload to table '{EDITOR_EVENTS_TABLE}'"))?;
 553
 554        const INLINE_COMPLETION_EVENTS_TABLE: &str = "inline_completion_events";
 555        Self::upload_to_table(
 556            INLINE_COMPLETION_EVENTS_TABLE,
 557            &self.inline_completion_events,
 558            clickhouse_client,
 559        )
 560        .await
 561        .with_context(|| format!("failed to upload to table '{INLINE_COMPLETION_EVENTS_TABLE}'"))?;
 562
 563        const ASSISTANT_EVENTS_TABLE: &str = "assistant_events";
 564        Self::upload_to_table(
 565            ASSISTANT_EVENTS_TABLE,
 566            &self.assistant_events,
 567            clickhouse_client,
 568        )
 569        .await
 570        .with_context(|| format!("failed to upload to table '{ASSISTANT_EVENTS_TABLE}'"))?;
 571
 572        const CALL_EVENTS_TABLE: &str = "call_events";
 573        Self::upload_to_table(CALL_EVENTS_TABLE, &self.call_events, clickhouse_client)
 574            .await
 575            .with_context(|| format!("failed to upload to table '{CALL_EVENTS_TABLE}'"))?;
 576
 577        const CPU_EVENTS_TABLE: &str = "cpu_events";
 578        Self::upload_to_table(CPU_EVENTS_TABLE, &self.cpu_events, clickhouse_client)
 579            .await
 580            .with_context(|| format!("failed to upload to table '{CPU_EVENTS_TABLE}'"))?;
 581
 582        const MEMORY_EVENTS_TABLE: &str = "memory_events";
 583        Self::upload_to_table(MEMORY_EVENTS_TABLE, &self.memory_events, clickhouse_client)
 584            .await
 585            .with_context(|| format!("failed to upload to table '{MEMORY_EVENTS_TABLE}'"))?;
 586
 587        const APP_EVENTS_TABLE: &str = "app_events";
 588        Self::upload_to_table(APP_EVENTS_TABLE, &self.app_events, clickhouse_client)
 589            .await
 590            .with_context(|| format!("failed to upload to table '{APP_EVENTS_TABLE}'"))?;
 591
 592        const SETTING_EVENTS_TABLE: &str = "setting_events";
 593        Self::upload_to_table(
 594            SETTING_EVENTS_TABLE,
 595            &self.setting_events,
 596            clickhouse_client,
 597        )
 598        .await
 599        .with_context(|| format!("failed to upload to table '{SETTING_EVENTS_TABLE}'"))?;
 600
 601        const EXTENSION_EVENTS_TABLE: &str = "extension_events";
 602        Self::upload_to_table(
 603            EXTENSION_EVENTS_TABLE,
 604            &self.extension_events,
 605            clickhouse_client,
 606        )
 607        .await
 608        .with_context(|| format!("failed to upload to table '{EXTENSION_EVENTS_TABLE}'"))?;
 609
 610        const EDIT_EVENTS_TABLE: &str = "edit_events";
 611        Self::upload_to_table(EDIT_EVENTS_TABLE, &self.edit_events, clickhouse_client)
 612            .await
 613            .with_context(|| format!("failed to upload to table '{EDIT_EVENTS_TABLE}'"))?;
 614
 615        const ACTION_EVENTS_TABLE: &str = "action_events";
 616        Self::upload_to_table(ACTION_EVENTS_TABLE, &self.action_events, clickhouse_client)
 617            .await
 618            .with_context(|| format!("failed to upload to table '{ACTION_EVENTS_TABLE}'"))?;
 619
 620        Ok(())
 621    }
 622
 623    async fn upload_to_table<T: clickhouse::Row + Serialize + std::fmt::Debug>(
 624        table: &str,
 625        rows: &[T],
 626        clickhouse_client: &clickhouse::Client,
 627    ) -> anyhow::Result<()> {
 628        if !rows.is_empty() {
 629            let mut insert = clickhouse_client.insert(table)?;
 630
 631            for event in rows {
 632                insert.write(event).await?;
 633            }
 634
 635            insert.end().await?;
 636
 637            let event_count = rows.len();
 638            log::info!(
 639                "wrote {event_count} {event_specifier} to '{table}'",
 640                event_specifier = if event_count == 1 { "event" } else { "events" }
 641            );
 642        }
 643
 644        Ok(())
 645    }
 646}
 647
 648pub fn serialize_country_code<S>(country_code: &str, serializer: S) -> Result<S::Ok, S::Error>
 649where
 650    S: Serializer,
 651{
 652    if country_code.len() != 2 {
 653        use serde::ser::Error;
 654        return Err(S::Error::custom(
 655            "country_code must be exactly 2 characters",
 656        ));
 657    }
 658
 659    let country_code = country_code.as_bytes();
 660
 661    serializer.serialize_u16(((country_code[1] as u16) << 8) + country_code[0] as u16)
 662}
 663
 664#[derive(Serialize, Debug, clickhouse::Row)]
 665pub struct EditorEventRow {
 666    installation_id: String,
 667    metrics_id: String,
 668    operation: String,
 669    app_version: String,
 670    file_extension: String,
 671    os_name: String,
 672    os_version: String,
 673    release_channel: String,
 674    signed_in: bool,
 675    vim_mode: bool,
 676    #[serde(serialize_with = "serialize_country_code")]
 677    country_code: String,
 678    region_code: String,
 679    city: String,
 680    time: i64,
 681    copilot_enabled: bool,
 682    copilot_enabled_for_language: bool,
 683    historical_event: bool,
 684    architecture: String,
 685    is_staff: Option<bool>,
 686    session_id: Option<String>,
 687    major: Option<i32>,
 688    minor: Option<i32>,
 689    patch: Option<i32>,
 690    checksum_matched: bool,
 691}
 692
 693impl EditorEventRow {
 694    fn from_event(
 695        event: EditorEvent,
 696        wrapper: &EventWrapper,
 697        body: &EventRequestBody,
 698        first_event_at: chrono::DateTime<chrono::Utc>,
 699        country_code: Option<String>,
 700        checksum_matched: bool,
 701    ) -> Self {
 702        let semver = body.semver();
 703        let time =
 704            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
 705
 706        Self {
 707            app_version: body.app_version.clone(),
 708            major: semver.map(|v| v.major() as i32),
 709            minor: semver.map(|v| v.minor() as i32),
 710            patch: semver.map(|v| v.patch() as i32),
 711            checksum_matched,
 712            release_channel: body.release_channel.clone().unwrap_or_default(),
 713            os_name: body.os_name.clone(),
 714            os_version: body.os_version.clone().unwrap_or_default(),
 715            architecture: body.architecture.clone(),
 716            installation_id: body.installation_id.clone().unwrap_or_default(),
 717            metrics_id: body.metrics_id.clone().unwrap_or_default(),
 718            session_id: body.session_id.clone(),
 719            is_staff: body.is_staff,
 720            time: time.timestamp_millis(),
 721            operation: event.operation,
 722            file_extension: event.file_extension.unwrap_or_default(),
 723            signed_in: wrapper.signed_in,
 724            vim_mode: event.vim_mode,
 725            copilot_enabled: event.copilot_enabled,
 726            copilot_enabled_for_language: event.copilot_enabled_for_language,
 727            country_code: country_code.unwrap_or("XX".to_string()),
 728            region_code: "".to_string(),
 729            city: "".to_string(),
 730            historical_event: false,
 731        }
 732    }
 733}
 734
 735#[derive(Serialize, Debug, clickhouse::Row)]
 736pub struct InlineCompletionEventRow {
 737    installation_id: String,
 738    provider: String,
 739    suggestion_accepted: bool,
 740    app_version: String,
 741    file_extension: String,
 742    os_name: String,
 743    os_version: String,
 744    release_channel: String,
 745    signed_in: bool,
 746    #[serde(serialize_with = "serialize_country_code")]
 747    country_code: String,
 748    region_code: String,
 749    city: String,
 750    time: i64,
 751    is_staff: Option<bool>,
 752    session_id: Option<String>,
 753    major: Option<i32>,
 754    minor: Option<i32>,
 755    patch: Option<i32>,
 756    checksum_matched: bool,
 757}
 758
 759impl InlineCompletionEventRow {
 760    fn from_event(
 761        event: InlineCompletionEvent,
 762        wrapper: &EventWrapper,
 763        body: &EventRequestBody,
 764        first_event_at: chrono::DateTime<chrono::Utc>,
 765        country_code: Option<String>,
 766        checksum_matched: bool,
 767    ) -> Self {
 768        let semver = body.semver();
 769        let time =
 770            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
 771
 772        Self {
 773            app_version: body.app_version.clone(),
 774            major: semver.map(|v| v.major() as i32),
 775            minor: semver.map(|v| v.minor() as i32),
 776            patch: semver.map(|v| v.patch() as i32),
 777            checksum_matched,
 778            release_channel: body.release_channel.clone().unwrap_or_default(),
 779            os_name: body.os_name.clone(),
 780            os_version: body.os_version.clone().unwrap_or_default(),
 781            installation_id: body.installation_id.clone().unwrap_or_default(),
 782            session_id: body.session_id.clone(),
 783            is_staff: body.is_staff,
 784            time: time.timestamp_millis(),
 785            file_extension: event.file_extension.unwrap_or_default(),
 786            signed_in: wrapper.signed_in,
 787            country_code: country_code.unwrap_or("XX".to_string()),
 788            region_code: "".to_string(),
 789            city: "".to_string(),
 790            provider: event.provider,
 791            suggestion_accepted: event.suggestion_accepted,
 792        }
 793    }
 794}
 795
 796#[derive(Serialize, Debug, clickhouse::Row)]
 797pub struct CallEventRow {
 798    // AppInfoBase
 799    app_version: String,
 800    major: Option<i32>,
 801    minor: Option<i32>,
 802    patch: Option<i32>,
 803    release_channel: String,
 804    os_name: String,
 805    os_version: String,
 806    checksum_matched: bool,
 807
 808    // ClientEventBase
 809    installation_id: String,
 810    session_id: Option<String>,
 811    is_staff: Option<bool>,
 812    time: i64,
 813
 814    // CallEventRow
 815    operation: String,
 816    room_id: Option<u64>,
 817    channel_id: Option<u64>,
 818}
 819
 820impl CallEventRow {
 821    fn from_event(
 822        event: CallEvent,
 823        wrapper: &EventWrapper,
 824        body: &EventRequestBody,
 825        first_event_at: chrono::DateTime<chrono::Utc>,
 826        checksum_matched: bool,
 827    ) -> Self {
 828        let semver = body.semver();
 829        let time =
 830            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
 831
 832        Self {
 833            app_version: body.app_version.clone(),
 834            major: semver.map(|v| v.major() as i32),
 835            minor: semver.map(|v| v.minor() as i32),
 836            patch: semver.map(|v| v.patch() as i32),
 837            checksum_matched,
 838            release_channel: body.release_channel.clone().unwrap_or_default(),
 839            os_name: body.os_name.clone(),
 840            os_version: body.os_version.clone().unwrap_or_default(),
 841            installation_id: body.installation_id.clone().unwrap_or_default(),
 842            session_id: body.session_id.clone(),
 843            is_staff: body.is_staff,
 844            time: time.timestamp_millis(),
 845            operation: event.operation,
 846            room_id: event.room_id,
 847            channel_id: event.channel_id,
 848        }
 849    }
 850}
 851
 852#[derive(Serialize, Debug, clickhouse::Row)]
 853pub struct AssistantEventRow {
 854    // AppInfoBase
 855    app_version: String,
 856    major: Option<i32>,
 857    minor: Option<i32>,
 858    patch: Option<i32>,
 859    checksum_matched: bool,
 860    release_channel: String,
 861    os_name: String,
 862    os_version: String,
 863
 864    // ClientEventBase
 865    installation_id: Option<String>,
 866    session_id: Option<String>,
 867    is_staff: Option<bool>,
 868    time: i64,
 869
 870    // AssistantEventRow
 871    conversation_id: String,
 872    kind: String,
 873    model: String,
 874    response_latency_in_ms: Option<i64>,
 875    error_message: Option<String>,
 876}
 877
 878impl AssistantEventRow {
 879    fn from_event(
 880        event: AssistantEvent,
 881        wrapper: &EventWrapper,
 882        body: &EventRequestBody,
 883        first_event_at: chrono::DateTime<chrono::Utc>,
 884        checksum_matched: bool,
 885    ) -> Self {
 886        let semver = body.semver();
 887        let time =
 888            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
 889
 890        Self {
 891            app_version: body.app_version.clone(),
 892            major: semver.map(|v| v.major() as i32),
 893            minor: semver.map(|v| v.minor() as i32),
 894            patch: semver.map(|v| v.patch() as i32),
 895            checksum_matched,
 896            release_channel: body.release_channel.clone().unwrap_or_default(),
 897            os_name: body.os_name.clone(),
 898            os_version: body.os_version.clone().unwrap_or_default(),
 899            installation_id: body.installation_id.clone(),
 900            session_id: body.session_id.clone(),
 901            is_staff: body.is_staff,
 902            time: time.timestamp_millis(),
 903            conversation_id: event.conversation_id.unwrap_or_default(),
 904            kind: event.kind.to_string(),
 905            model: event.model,
 906            response_latency_in_ms: event
 907                .response_latency
 908                .map(|latency| latency.as_millis() as i64),
 909            error_message: event.error_message,
 910        }
 911    }
 912}
 913
 914#[derive(Debug, clickhouse::Row, Serialize)]
 915pub struct CpuEventRow {
 916    installation_id: Option<String>,
 917    is_staff: Option<bool>,
 918    usage_as_percentage: f32,
 919    core_count: u32,
 920    app_version: String,
 921    release_channel: String,
 922    os_name: String,
 923    os_version: String,
 924    time: i64,
 925    session_id: Option<String>,
 926    // pub normalized_cpu_usage: f64, MATERIALIZED
 927    major: Option<i32>,
 928    minor: Option<i32>,
 929    patch: Option<i32>,
 930    checksum_matched: bool,
 931}
 932
 933impl CpuEventRow {
 934    fn from_event(
 935        event: CpuEvent,
 936        wrapper: &EventWrapper,
 937        body: &EventRequestBody,
 938        first_event_at: chrono::DateTime<chrono::Utc>,
 939        checksum_matched: bool,
 940    ) -> Self {
 941        let semver = body.semver();
 942        let time =
 943            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
 944
 945        Self {
 946            app_version: body.app_version.clone(),
 947            major: semver.map(|v| v.major() as i32),
 948            minor: semver.map(|v| v.minor() as i32),
 949            patch: semver.map(|v| v.patch() as i32),
 950            checksum_matched,
 951            release_channel: body.release_channel.clone().unwrap_or_default(),
 952            os_name: body.os_name.clone(),
 953            os_version: body.os_version.clone().unwrap_or_default(),
 954            installation_id: body.installation_id.clone(),
 955            session_id: body.session_id.clone(),
 956            is_staff: body.is_staff,
 957            time: time.timestamp_millis(),
 958            usage_as_percentage: event.usage_as_percentage,
 959            core_count: event.core_count,
 960        }
 961    }
 962}
 963
 964#[derive(Serialize, Debug, clickhouse::Row)]
 965pub struct MemoryEventRow {
 966    // AppInfoBase
 967    app_version: String,
 968    major: Option<i32>,
 969    minor: Option<i32>,
 970    patch: Option<i32>,
 971    checksum_matched: bool,
 972    release_channel: String,
 973    os_name: String,
 974    os_version: String,
 975
 976    // ClientEventBase
 977    installation_id: Option<String>,
 978    session_id: Option<String>,
 979    is_staff: Option<bool>,
 980    time: i64,
 981
 982    // MemoryEventRow
 983    memory_in_bytes: u64,
 984    virtual_memory_in_bytes: u64,
 985}
 986
 987impl MemoryEventRow {
 988    fn from_event(
 989        event: MemoryEvent,
 990        wrapper: &EventWrapper,
 991        body: &EventRequestBody,
 992        first_event_at: chrono::DateTime<chrono::Utc>,
 993        checksum_matched: bool,
 994    ) -> Self {
 995        let semver = body.semver();
 996        let time =
 997            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
 998
 999        Self {
1000            app_version: body.app_version.clone(),
1001            major: semver.map(|v| v.major() as i32),
1002            minor: semver.map(|v| v.minor() as i32),
1003            patch: semver.map(|v| v.patch() as i32),
1004            checksum_matched,
1005            release_channel: body.release_channel.clone().unwrap_or_default(),
1006            os_name: body.os_name.clone(),
1007            os_version: body.os_version.clone().unwrap_or_default(),
1008            installation_id: body.installation_id.clone(),
1009            session_id: body.session_id.clone(),
1010            is_staff: body.is_staff,
1011            time: time.timestamp_millis(),
1012            memory_in_bytes: event.memory_in_bytes,
1013            virtual_memory_in_bytes: event.virtual_memory_in_bytes,
1014        }
1015    }
1016}
1017
1018#[derive(Serialize, Debug, clickhouse::Row)]
1019pub struct AppEventRow {
1020    // AppInfoBase
1021    app_version: String,
1022    major: Option<i32>,
1023    minor: Option<i32>,
1024    patch: Option<i32>,
1025    checksum_matched: bool,
1026    release_channel: String,
1027    os_name: String,
1028    os_version: String,
1029
1030    // ClientEventBase
1031    installation_id: Option<String>,
1032    session_id: Option<String>,
1033    is_staff: Option<bool>,
1034    time: i64,
1035
1036    // AppEventRow
1037    operation: String,
1038}
1039
1040impl AppEventRow {
1041    fn from_event(
1042        event: AppEvent,
1043        wrapper: &EventWrapper,
1044        body: &EventRequestBody,
1045        first_event_at: chrono::DateTime<chrono::Utc>,
1046        checksum_matched: bool,
1047    ) -> Self {
1048        let semver = body.semver();
1049        let time =
1050            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
1051
1052        Self {
1053            app_version: body.app_version.clone(),
1054            major: semver.map(|v| v.major() as i32),
1055            minor: semver.map(|v| v.minor() as i32),
1056            patch: semver.map(|v| v.patch() as i32),
1057            checksum_matched,
1058            release_channel: body.release_channel.clone().unwrap_or_default(),
1059            os_name: body.os_name.clone(),
1060            os_version: body.os_version.clone().unwrap_or_default(),
1061            installation_id: body.installation_id.clone(),
1062            session_id: body.session_id.clone(),
1063            is_staff: body.is_staff,
1064            time: time.timestamp_millis(),
1065            operation: event.operation,
1066        }
1067    }
1068}
1069
1070#[derive(Serialize, Debug, clickhouse::Row)]
1071pub struct SettingEventRow {
1072    // AppInfoBase
1073    app_version: String,
1074    major: Option<i32>,
1075    minor: Option<i32>,
1076    patch: Option<i32>,
1077    checksum_matched: bool,
1078    release_channel: String,
1079    os_name: String,
1080    os_version: String,
1081
1082    // ClientEventBase
1083    installation_id: Option<String>,
1084    session_id: Option<String>,
1085    is_staff: Option<bool>,
1086    time: i64,
1087    // SettingEventRow
1088    setting: String,
1089    value: String,
1090}
1091
1092impl SettingEventRow {
1093    fn from_event(
1094        event: SettingEvent,
1095        wrapper: &EventWrapper,
1096        body: &EventRequestBody,
1097        first_event_at: chrono::DateTime<chrono::Utc>,
1098        checksum_matched: bool,
1099    ) -> Self {
1100        let semver = body.semver();
1101        let time =
1102            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
1103
1104        Self {
1105            app_version: body.app_version.clone(),
1106            major: semver.map(|v| v.major() as i32),
1107            minor: semver.map(|v| v.minor() as i32),
1108            checksum_matched,
1109            patch: semver.map(|v| v.patch() as i32),
1110            release_channel: body.release_channel.clone().unwrap_or_default(),
1111            os_name: body.os_name.clone(),
1112            os_version: body.os_version.clone().unwrap_or_default(),
1113            installation_id: body.installation_id.clone(),
1114            session_id: body.session_id.clone(),
1115            is_staff: body.is_staff,
1116            time: time.timestamp_millis(),
1117            setting: event.setting,
1118            value: event.value,
1119        }
1120    }
1121}
1122
1123#[derive(Serialize, Debug, clickhouse::Row)]
1124pub struct ExtensionEventRow {
1125    // AppInfoBase
1126    app_version: String,
1127    major: Option<i32>,
1128    minor: Option<i32>,
1129    patch: Option<i32>,
1130    checksum_matched: bool,
1131    release_channel: String,
1132    os_name: String,
1133    os_version: String,
1134
1135    // ClientEventBase
1136    installation_id: Option<String>,
1137    session_id: Option<String>,
1138    is_staff: Option<bool>,
1139    time: i64,
1140
1141    // ExtensionEventRow
1142    extension_id: Arc<str>,
1143    extension_version: Arc<str>,
1144    dev: bool,
1145    schema_version: Option<i32>,
1146    wasm_api_version: Option<String>,
1147}
1148
1149impl ExtensionEventRow {
1150    fn from_event(
1151        event: ExtensionEvent,
1152        wrapper: &EventWrapper,
1153        body: &EventRequestBody,
1154        extension_metadata: Option<ExtensionMetadata>,
1155        first_event_at: chrono::DateTime<chrono::Utc>,
1156        checksum_matched: bool,
1157    ) -> Self {
1158        let semver = body.semver();
1159        let time =
1160            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
1161
1162        Self {
1163            app_version: body.app_version.clone(),
1164            major: semver.map(|v| v.major() as i32),
1165            minor: semver.map(|v| v.minor() as i32),
1166            patch: semver.map(|v| v.patch() as i32),
1167            checksum_matched,
1168            release_channel: body.release_channel.clone().unwrap_or_default(),
1169            os_name: body.os_name.clone(),
1170            os_version: body.os_version.clone().unwrap_or_default(),
1171            installation_id: body.installation_id.clone(),
1172            session_id: body.session_id.clone(),
1173            is_staff: body.is_staff,
1174            time: time.timestamp_millis(),
1175            extension_id: event.extension_id,
1176            extension_version: event.version,
1177            dev: extension_metadata.is_none(),
1178            schema_version: extension_metadata
1179                .as_ref()
1180                .and_then(|metadata| metadata.manifest.schema_version),
1181            wasm_api_version: extension_metadata.as_ref().and_then(|metadata| {
1182                metadata
1183                    .manifest
1184                    .wasm_api_version
1185                    .as_ref()
1186                    .map(|version| version.to_string())
1187            }),
1188        }
1189    }
1190}
1191
1192#[derive(Serialize, Debug, clickhouse::Row)]
1193pub struct EditEventRow {
1194    // AppInfoBase
1195    app_version: String,
1196    major: Option<i32>,
1197    minor: Option<i32>,
1198    patch: Option<i32>,
1199    checksum_matched: bool,
1200    release_channel: String,
1201    os_name: String,
1202    os_version: String,
1203
1204    // ClientEventBase
1205    installation_id: Option<String>,
1206    // Note: This column name has a typo in the ClickHouse table.
1207    #[serde(rename = "sesssion_id")]
1208    session_id: Option<String>,
1209    is_staff: Option<bool>,
1210    time: i64,
1211
1212    // EditEventRow
1213    period_start: i64,
1214    period_end: i64,
1215    environment: String,
1216}
1217
1218impl EditEventRow {
1219    fn from_event(
1220        event: EditEvent,
1221        wrapper: &EventWrapper,
1222        body: &EventRequestBody,
1223        first_event_at: chrono::DateTime<chrono::Utc>,
1224        checksum_matched: bool,
1225    ) -> Self {
1226        let semver = body.semver();
1227        let time =
1228            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
1229
1230        let period_start = time - chrono::Duration::milliseconds(event.duration);
1231        let period_end = time;
1232
1233        Self {
1234            app_version: body.app_version.clone(),
1235            major: semver.map(|v| v.major() as i32),
1236            minor: semver.map(|v| v.minor() as i32),
1237            patch: semver.map(|v| v.patch() as i32),
1238            checksum_matched,
1239            release_channel: body.release_channel.clone().unwrap_or_default(),
1240            os_name: body.os_name.clone(),
1241            os_version: body.os_version.clone().unwrap_or_default(),
1242            installation_id: body.installation_id.clone(),
1243            session_id: body.session_id.clone(),
1244            is_staff: body.is_staff,
1245            time: time.timestamp_millis(),
1246            period_start: period_start.timestamp_millis(),
1247            period_end: period_end.timestamp_millis(),
1248            environment: event.environment,
1249        }
1250    }
1251}
1252
1253#[derive(Serialize, Debug, clickhouse::Row)]
1254pub struct ActionEventRow {
1255    // AppInfoBase
1256    app_version: String,
1257    major: Option<i32>,
1258    minor: Option<i32>,
1259    patch: Option<i32>,
1260    checksum_matched: bool,
1261    release_channel: String,
1262    os_name: String,
1263    os_version: String,
1264
1265    // ClientEventBase
1266    installation_id: Option<String>,
1267    // Note: This column name has a typo in the ClickHouse table.
1268    #[serde(rename = "sesssion_id")]
1269    session_id: Option<String>,
1270    is_staff: Option<bool>,
1271    time: i64,
1272    // ActionEventRow
1273    source: String,
1274    action: String,
1275}
1276
1277impl ActionEventRow {
1278    fn from_event(
1279        event: ActionEvent,
1280        wrapper: &EventWrapper,
1281        body: &EventRequestBody,
1282        first_event_at: chrono::DateTime<chrono::Utc>,
1283        checksum_matched: bool,
1284    ) -> Self {
1285        let semver = body.semver();
1286        let time =
1287            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
1288
1289        Self {
1290            app_version: body.app_version.clone(),
1291            major: semver.map(|v| v.major() as i32),
1292            minor: semver.map(|v| v.minor() as i32),
1293            patch: semver.map(|v| v.patch() as i32),
1294            checksum_matched,
1295            release_channel: body.release_channel.clone().unwrap_or_default(),
1296            os_name: body.os_name.clone(),
1297            os_version: body.os_version.clone().unwrap_or_default(),
1298            installation_id: body.installation_id.clone(),
1299            session_id: body.session_id.clone(),
1300            is_staff: body.is_staff,
1301            time: time.timestamp_millis(),
1302            source: event.source,
1303            action: event.action,
1304        }
1305    }
1306}
1307
1308pub fn calculate_json_checksum(app: Arc<AppState>, json: &impl AsRef<[u8]>) -> Option<Vec<u8>> {
1309    let Some(checksum_seed) = app.config.zed_client_checksum_seed.as_ref() else {
1310        return None;
1311    };
1312
1313    let mut summer = Sha256::new();
1314    summer.update(checksum_seed);
1315    summer.update(&json);
1316    summer.update(checksum_seed);
1317    Some(summer.finalize().into_iter().collect())
1318}