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    // better OS reporting for linux (because linux is hard):
 312    // - Remove os_version/app_version/os_name from the gpui platform trait
 313    // - Move platform processing data into client/telemetry
 314    // - Duplicate some small code in macOS platform for a version check
 315    // - Add GPUI API for reporting the selected platform integration
 316    //  - macos-blade, macos-metal, linux-X11, linux-headless
 317    // if cfg(macos( { "Macos" } else { "Linux-{cx.compositor_name()"} ))
 318
 319    tracing::error!(
 320        service = "client",
 321        version = %panic.app_version,
 322        os_name = %panic.os_name,
 323        os_version = %panic.os_version.clone().unwrap_or_default(),
 324        installation_id = %panic.installation_id.unwrap_or_default(),
 325        description = %panic.payload,
 326        backtrace = %panic.backtrace.join("\n"),
 327        "panic report");
 328
 329    let backtrace = if panic.backtrace.len() > 25 {
 330        let total = panic.backtrace.len();
 331        format!(
 332            "{}\n   and {} more",
 333            panic
 334                .backtrace
 335                .iter()
 336                .take(20)
 337                .cloned()
 338                .collect::<Vec<_>>()
 339                .join("\n"),
 340            total - 20
 341        )
 342    } else {
 343        panic.backtrace.join("\n")
 344    };
 345    let backtrace_with_summary = panic.payload + "\n" + &backtrace;
 346
 347    if let Some(slack_panics_webhook) = app.config.slack_panics_webhook.clone() {
 348        let payload = slack::WebhookBody::new(|w| {
 349            w.add_section(|s| s.text(slack::Text::markdown("Panic request".to_string())))
 350                .add_section(|s| {
 351                    s.add_field(slack::Text::markdown(format!(
 352                        "*Version:*\n {} ",
 353                        panic.app_version
 354                    )))
 355                    .add_field({
 356                        slack::Text::markdown(format!(
 357                            "*OS:*\n{} {}",
 358                            panic.os_name,
 359                            panic.os_version.unwrap_or_default()
 360                        ))
 361                    })
 362                })
 363                .add_rich_text(|r| r.add_preformatted(|p| p.add_text(backtrace_with_summary)))
 364        });
 365        let payload_json = serde_json::to_string(&payload).map_err(|err| {
 366            log::error!("Failed to serialize payload to JSON: {err}");
 367            Error::Internal(anyhow!(err))
 368        })?;
 369
 370        reqwest::Client::new()
 371            .post(slack_panics_webhook)
 372            .header("Content-Type", "application/json")
 373            .body(payload_json)
 374            .send()
 375            .await
 376            .map_err(|err| {
 377                log::error!("Failed to send payload to Slack: {err}");
 378                Error::Internal(anyhow!(err))
 379            })?;
 380    }
 381
 382    Ok(())
 383}
 384
 385pub async fn post_events(
 386    Extension(app): Extension<Arc<AppState>>,
 387    TypedHeader(ZedChecksumHeader(checksum)): TypedHeader<ZedChecksumHeader>,
 388    country_code_header: Option<TypedHeader<CloudflareIpCountryHeader>>,
 389    body: Bytes,
 390) -> Result<()> {
 391    let Some(clickhouse_client) = app.clickhouse_client.clone() else {
 392        Err(Error::Http(
 393            StatusCode::NOT_IMPLEMENTED,
 394            "not supported".into(),
 395        ))?
 396    };
 397
 398    let Some(expected) = calculate_json_checksum(app.clone(), &body) else {
 399        return Err(Error::Http(
 400            StatusCode::INTERNAL_SERVER_ERROR,
 401            "events not enabled".into(),
 402        ))?;
 403    };
 404
 405    let checksum_matched = checksum == expected;
 406
 407    let request_body: telemetry_events::EventRequestBody =
 408        serde_json::from_slice(&body).map_err(|err| {
 409            log::error!("can't parse event json: {err}");
 410            Error::Internal(anyhow!(err))
 411        })?;
 412
 413    let mut to_upload = ToUpload::default();
 414    let Some(last_event) = request_body.events.last() else {
 415        return Err(Error::Http(StatusCode::BAD_REQUEST, "no events".into()))?;
 416    };
 417    let country_code = country_code_header.map(|h| h.0 .0);
 418
 419    let first_event_at = chrono::Utc::now()
 420        - chrono::Duration::milliseconds(last_event.milliseconds_since_first_event);
 421
 422    for wrapper in &request_body.events {
 423        match &wrapper.event {
 424            Event::Editor(event) => to_upload.editor_events.push(EditorEventRow::from_event(
 425                event.clone(),
 426                &wrapper,
 427                &request_body,
 428                first_event_at,
 429                country_code.clone(),
 430                checksum_matched,
 431            )),
 432            // Needed for clients sending old copilot_event types
 433            Event::Copilot(_) => {}
 434            Event::InlineCompletion(event) => {
 435                to_upload
 436                    .inline_completion_events
 437                    .push(InlineCompletionEventRow::from_event(
 438                        event.clone(),
 439                        &wrapper,
 440                        &request_body,
 441                        first_event_at,
 442                        country_code.clone(),
 443                        checksum_matched,
 444                    ))
 445            }
 446            Event::Call(event) => to_upload.call_events.push(CallEventRow::from_event(
 447                event.clone(),
 448                &wrapper,
 449                &request_body,
 450                first_event_at,
 451                checksum_matched,
 452            )),
 453            Event::Assistant(event) => {
 454                to_upload
 455                    .assistant_events
 456                    .push(AssistantEventRow::from_event(
 457                        event.clone(),
 458                        &wrapper,
 459                        &request_body,
 460                        first_event_at,
 461                        checksum_matched,
 462                    ))
 463            }
 464            Event::Cpu(event) => to_upload.cpu_events.push(CpuEventRow::from_event(
 465                event.clone(),
 466                &wrapper,
 467                &request_body,
 468                first_event_at,
 469                checksum_matched,
 470            )),
 471            Event::Memory(event) => to_upload.memory_events.push(MemoryEventRow::from_event(
 472                event.clone(),
 473                &wrapper,
 474                &request_body,
 475                first_event_at,
 476                checksum_matched,
 477            )),
 478            Event::App(event) => to_upload.app_events.push(AppEventRow::from_event(
 479                event.clone(),
 480                &wrapper,
 481                &request_body,
 482                first_event_at,
 483                checksum_matched,
 484            )),
 485            Event::Setting(event) => to_upload.setting_events.push(SettingEventRow::from_event(
 486                event.clone(),
 487                &wrapper,
 488                &request_body,
 489                first_event_at,
 490                checksum_matched,
 491            )),
 492            Event::Edit(event) => to_upload.edit_events.push(EditEventRow::from_event(
 493                event.clone(),
 494                &wrapper,
 495                &request_body,
 496                first_event_at,
 497                checksum_matched,
 498            )),
 499            Event::Action(event) => to_upload.action_events.push(ActionEventRow::from_event(
 500                event.clone(),
 501                &wrapper,
 502                &request_body,
 503                first_event_at,
 504                checksum_matched,
 505            )),
 506            Event::Extension(event) => {
 507                let metadata = app
 508                    .db
 509                    .get_extension_version(&event.extension_id, &event.version)
 510                    .await?;
 511                to_upload
 512                    .extension_events
 513                    .push(ExtensionEventRow::from_event(
 514                        event.clone(),
 515                        &wrapper,
 516                        &request_body,
 517                        metadata,
 518                        first_event_at,
 519                        checksum_matched,
 520                    ))
 521            }
 522        }
 523    }
 524
 525    to_upload
 526        .upload(&clickhouse_client)
 527        .await
 528        .map_err(|err| Error::Internal(anyhow!(err)))?;
 529
 530    Ok(())
 531}
 532
 533#[derive(Default)]
 534struct ToUpload {
 535    editor_events: Vec<EditorEventRow>,
 536    inline_completion_events: Vec<InlineCompletionEventRow>,
 537    assistant_events: Vec<AssistantEventRow>,
 538    call_events: Vec<CallEventRow>,
 539    cpu_events: Vec<CpuEventRow>,
 540    memory_events: Vec<MemoryEventRow>,
 541    app_events: Vec<AppEventRow>,
 542    setting_events: Vec<SettingEventRow>,
 543    extension_events: Vec<ExtensionEventRow>,
 544    edit_events: Vec<EditEventRow>,
 545    action_events: Vec<ActionEventRow>,
 546}
 547
 548impl ToUpload {
 549    pub async fn upload(&self, clickhouse_client: &clickhouse::Client) -> anyhow::Result<()> {
 550        const EDITOR_EVENTS_TABLE: &str = "editor_events";
 551        Self::upload_to_table(EDITOR_EVENTS_TABLE, &self.editor_events, clickhouse_client)
 552            .await
 553            .with_context(|| format!("failed to upload to table '{EDITOR_EVENTS_TABLE}'"))?;
 554
 555        const INLINE_COMPLETION_EVENTS_TABLE: &str = "inline_completion_events";
 556        Self::upload_to_table(
 557            INLINE_COMPLETION_EVENTS_TABLE,
 558            &self.inline_completion_events,
 559            clickhouse_client,
 560        )
 561        .await
 562        .with_context(|| format!("failed to upload to table '{INLINE_COMPLETION_EVENTS_TABLE}'"))?;
 563
 564        const ASSISTANT_EVENTS_TABLE: &str = "assistant_events";
 565        Self::upload_to_table(
 566            ASSISTANT_EVENTS_TABLE,
 567            &self.assistant_events,
 568            clickhouse_client,
 569        )
 570        .await
 571        .with_context(|| format!("failed to upload to table '{ASSISTANT_EVENTS_TABLE}'"))?;
 572
 573        const CALL_EVENTS_TABLE: &str = "call_events";
 574        Self::upload_to_table(CALL_EVENTS_TABLE, &self.call_events, clickhouse_client)
 575            .await
 576            .with_context(|| format!("failed to upload to table '{CALL_EVENTS_TABLE}'"))?;
 577
 578        const CPU_EVENTS_TABLE: &str = "cpu_events";
 579        Self::upload_to_table(CPU_EVENTS_TABLE, &self.cpu_events, clickhouse_client)
 580            .await
 581            .with_context(|| format!("failed to upload to table '{CPU_EVENTS_TABLE}'"))?;
 582
 583        const MEMORY_EVENTS_TABLE: &str = "memory_events";
 584        Self::upload_to_table(MEMORY_EVENTS_TABLE, &self.memory_events, clickhouse_client)
 585            .await
 586            .with_context(|| format!("failed to upload to table '{MEMORY_EVENTS_TABLE}'"))?;
 587
 588        const APP_EVENTS_TABLE: &str = "app_events";
 589        Self::upload_to_table(APP_EVENTS_TABLE, &self.app_events, clickhouse_client)
 590            .await
 591            .with_context(|| format!("failed to upload to table '{APP_EVENTS_TABLE}'"))?;
 592
 593        const SETTING_EVENTS_TABLE: &str = "setting_events";
 594        Self::upload_to_table(
 595            SETTING_EVENTS_TABLE,
 596            &self.setting_events,
 597            clickhouse_client,
 598        )
 599        .await
 600        .with_context(|| format!("failed to upload to table '{SETTING_EVENTS_TABLE}'"))?;
 601
 602        const EXTENSION_EVENTS_TABLE: &str = "extension_events";
 603        Self::upload_to_table(
 604            EXTENSION_EVENTS_TABLE,
 605            &self.extension_events,
 606            clickhouse_client,
 607        )
 608        .await
 609        .with_context(|| format!("failed to upload to table '{EXTENSION_EVENTS_TABLE}'"))?;
 610
 611        const EDIT_EVENTS_TABLE: &str = "edit_events";
 612        Self::upload_to_table(EDIT_EVENTS_TABLE, &self.edit_events, clickhouse_client)
 613            .await
 614            .with_context(|| format!("failed to upload to table '{EDIT_EVENTS_TABLE}'"))?;
 615
 616        const ACTION_EVENTS_TABLE: &str = "action_events";
 617        Self::upload_to_table(ACTION_EVENTS_TABLE, &self.action_events, clickhouse_client)
 618            .await
 619            .with_context(|| format!("failed to upload to table '{ACTION_EVENTS_TABLE}'"))?;
 620
 621        Ok(())
 622    }
 623
 624    async fn upload_to_table<T: clickhouse::Row + Serialize + std::fmt::Debug>(
 625        table: &str,
 626        rows: &[T],
 627        clickhouse_client: &clickhouse::Client,
 628    ) -> anyhow::Result<()> {
 629        if !rows.is_empty() {
 630            let mut insert = clickhouse_client.insert(table)?;
 631
 632            for event in rows {
 633                insert.write(event).await?;
 634            }
 635
 636            insert.end().await?;
 637
 638            let event_count = rows.len();
 639            log::info!(
 640                "wrote {event_count} {event_specifier} to '{table}'",
 641                event_specifier = if event_count == 1 { "event" } else { "events" }
 642            );
 643        }
 644
 645        Ok(())
 646    }
 647}
 648
 649pub fn serialize_country_code<S>(country_code: &str, serializer: S) -> Result<S::Ok, S::Error>
 650where
 651    S: Serializer,
 652{
 653    if country_code.len() != 2 {
 654        use serde::ser::Error;
 655        return Err(S::Error::custom(
 656            "country_code must be exactly 2 characters",
 657        ));
 658    }
 659
 660    let country_code = country_code.as_bytes();
 661
 662    serializer.serialize_u16(((country_code[1] as u16) << 8) + country_code[0] as u16)
 663}
 664
 665#[derive(Serialize, Debug, clickhouse::Row)]
 666pub struct EditorEventRow {
 667    installation_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            session_id: body.session_id.clone(),
 718            is_staff: body.is_staff,
 719            time: time.timestamp_millis(),
 720            operation: event.operation,
 721            file_extension: event.file_extension.unwrap_or_default(),
 722            signed_in: wrapper.signed_in,
 723            vim_mode: event.vim_mode,
 724            copilot_enabled: event.copilot_enabled,
 725            copilot_enabled_for_language: event.copilot_enabled_for_language,
 726            country_code: country_code.unwrap_or("XX".to_string()),
 727            region_code: "".to_string(),
 728            city: "".to_string(),
 729            historical_event: false,
 730        }
 731    }
 732}
 733
 734#[derive(Serialize, Debug, clickhouse::Row)]
 735pub struct InlineCompletionEventRow {
 736    installation_id: String,
 737    provider: String,
 738    suggestion_accepted: bool,
 739    app_version: String,
 740    file_extension: String,
 741    os_name: String,
 742    os_version: String,
 743    release_channel: String,
 744    signed_in: bool,
 745    #[serde(serialize_with = "serialize_country_code")]
 746    country_code: String,
 747    region_code: String,
 748    city: String,
 749    time: i64,
 750    is_staff: Option<bool>,
 751    session_id: Option<String>,
 752    major: Option<i32>,
 753    minor: Option<i32>,
 754    patch: Option<i32>,
 755    checksum_matched: bool,
 756}
 757
 758impl InlineCompletionEventRow {
 759    fn from_event(
 760        event: InlineCompletionEvent,
 761        wrapper: &EventWrapper,
 762        body: &EventRequestBody,
 763        first_event_at: chrono::DateTime<chrono::Utc>,
 764        country_code: Option<String>,
 765        checksum_matched: bool,
 766    ) -> Self {
 767        let semver = body.semver();
 768        let time =
 769            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
 770
 771        Self {
 772            app_version: body.app_version.clone(),
 773            major: semver.map(|v| v.major() as i32),
 774            minor: semver.map(|v| v.minor() as i32),
 775            patch: semver.map(|v| v.patch() as i32),
 776            checksum_matched,
 777            release_channel: body.release_channel.clone().unwrap_or_default(),
 778            os_name: body.os_name.clone(),
 779            os_version: body.os_version.clone().unwrap_or_default(),
 780            installation_id: body.installation_id.clone().unwrap_or_default(),
 781            session_id: body.session_id.clone(),
 782            is_staff: body.is_staff,
 783            time: time.timestamp_millis(),
 784            file_extension: event.file_extension.unwrap_or_default(),
 785            signed_in: wrapper.signed_in,
 786            country_code: country_code.unwrap_or("XX".to_string()),
 787            region_code: "".to_string(),
 788            city: "".to_string(),
 789            provider: event.provider,
 790            suggestion_accepted: event.suggestion_accepted,
 791        }
 792    }
 793}
 794
 795#[derive(Serialize, Debug, clickhouse::Row)]
 796pub struct CallEventRow {
 797    // AppInfoBase
 798    app_version: String,
 799    major: Option<i32>,
 800    minor: Option<i32>,
 801    patch: Option<i32>,
 802    release_channel: String,
 803    os_name: String,
 804    os_version: String,
 805    checksum_matched: bool,
 806
 807    // ClientEventBase
 808    installation_id: String,
 809    session_id: Option<String>,
 810    is_staff: Option<bool>,
 811    time: i64,
 812
 813    // CallEventRow
 814    operation: String,
 815    room_id: Option<u64>,
 816    channel_id: Option<u64>,
 817}
 818
 819impl CallEventRow {
 820    fn from_event(
 821        event: CallEvent,
 822        wrapper: &EventWrapper,
 823        body: &EventRequestBody,
 824        first_event_at: chrono::DateTime<chrono::Utc>,
 825        checksum_matched: bool,
 826    ) -> Self {
 827        let semver = body.semver();
 828        let time =
 829            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
 830
 831        Self {
 832            app_version: body.app_version.clone(),
 833            major: semver.map(|v| v.major() as i32),
 834            minor: semver.map(|v| v.minor() as i32),
 835            patch: semver.map(|v| v.patch() as i32),
 836            checksum_matched,
 837            release_channel: body.release_channel.clone().unwrap_or_default(),
 838            os_name: body.os_name.clone(),
 839            os_version: body.os_version.clone().unwrap_or_default(),
 840            installation_id: body.installation_id.clone().unwrap_or_default(),
 841            session_id: body.session_id.clone(),
 842            is_staff: body.is_staff,
 843            time: time.timestamp_millis(),
 844            operation: event.operation,
 845            room_id: event.room_id,
 846            channel_id: event.channel_id,
 847        }
 848    }
 849}
 850
 851#[derive(Serialize, Debug, clickhouse::Row)]
 852pub struct AssistantEventRow {
 853    // AppInfoBase
 854    app_version: String,
 855    major: Option<i32>,
 856    minor: Option<i32>,
 857    patch: Option<i32>,
 858    checksum_matched: bool,
 859    release_channel: String,
 860    os_name: String,
 861    os_version: String,
 862
 863    // ClientEventBase
 864    installation_id: Option<String>,
 865    session_id: Option<String>,
 866    is_staff: Option<bool>,
 867    time: i64,
 868
 869    // AssistantEventRow
 870    conversation_id: String,
 871    kind: String,
 872    model: String,
 873    response_latency_in_ms: Option<i64>,
 874    error_message: Option<String>,
 875}
 876
 877impl AssistantEventRow {
 878    fn from_event(
 879        event: AssistantEvent,
 880        wrapper: &EventWrapper,
 881        body: &EventRequestBody,
 882        first_event_at: chrono::DateTime<chrono::Utc>,
 883        checksum_matched: bool,
 884    ) -> Self {
 885        let semver = body.semver();
 886        let time =
 887            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
 888
 889        Self {
 890            app_version: body.app_version.clone(),
 891            major: semver.map(|v| v.major() as i32),
 892            minor: semver.map(|v| v.minor() as i32),
 893            patch: semver.map(|v| v.patch() as i32),
 894            checksum_matched,
 895            release_channel: body.release_channel.clone().unwrap_or_default(),
 896            os_name: body.os_name.clone(),
 897            os_version: body.os_version.clone().unwrap_or_default(),
 898            installation_id: body.installation_id.clone(),
 899            session_id: body.session_id.clone(),
 900            is_staff: body.is_staff,
 901            time: time.timestamp_millis(),
 902            conversation_id: event.conversation_id.unwrap_or_default(),
 903            kind: event.kind.to_string(),
 904            model: event.model,
 905            response_latency_in_ms: event
 906                .response_latency
 907                .map(|latency| latency.as_millis() as i64),
 908            error_message: event.error_message,
 909        }
 910    }
 911}
 912
 913#[derive(Debug, clickhouse::Row, Serialize)]
 914pub struct CpuEventRow {
 915    installation_id: Option<String>,
 916    is_staff: Option<bool>,
 917    usage_as_percentage: f32,
 918    core_count: u32,
 919    app_version: String,
 920    release_channel: String,
 921    os_name: String,
 922    os_version: String,
 923    time: i64,
 924    session_id: Option<String>,
 925    // pub normalized_cpu_usage: f64, MATERIALIZED
 926    major: Option<i32>,
 927    minor: Option<i32>,
 928    patch: Option<i32>,
 929    checksum_matched: bool,
 930}
 931
 932impl CpuEventRow {
 933    fn from_event(
 934        event: CpuEvent,
 935        wrapper: &EventWrapper,
 936        body: &EventRequestBody,
 937        first_event_at: chrono::DateTime<chrono::Utc>,
 938        checksum_matched: bool,
 939    ) -> Self {
 940        let semver = body.semver();
 941        let time =
 942            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
 943
 944        Self {
 945            app_version: body.app_version.clone(),
 946            major: semver.map(|v| v.major() as i32),
 947            minor: semver.map(|v| v.minor() as i32),
 948            patch: semver.map(|v| v.patch() as i32),
 949            checksum_matched,
 950            release_channel: body.release_channel.clone().unwrap_or_default(),
 951            os_name: body.os_name.clone(),
 952            os_version: body.os_version.clone().unwrap_or_default(),
 953            installation_id: body.installation_id.clone(),
 954            session_id: body.session_id.clone(),
 955            is_staff: body.is_staff,
 956            time: time.timestamp_millis(),
 957            usage_as_percentage: event.usage_as_percentage,
 958            core_count: event.core_count,
 959        }
 960    }
 961}
 962
 963#[derive(Serialize, Debug, clickhouse::Row)]
 964pub struct MemoryEventRow {
 965    // AppInfoBase
 966    app_version: String,
 967    major: Option<i32>,
 968    minor: Option<i32>,
 969    patch: Option<i32>,
 970    checksum_matched: bool,
 971    release_channel: String,
 972    os_name: String,
 973    os_version: String,
 974
 975    // ClientEventBase
 976    installation_id: Option<String>,
 977    session_id: Option<String>,
 978    is_staff: Option<bool>,
 979    time: i64,
 980
 981    // MemoryEventRow
 982    memory_in_bytes: u64,
 983    virtual_memory_in_bytes: u64,
 984}
 985
 986impl MemoryEventRow {
 987    fn from_event(
 988        event: MemoryEvent,
 989        wrapper: &EventWrapper,
 990        body: &EventRequestBody,
 991        first_event_at: chrono::DateTime<chrono::Utc>,
 992        checksum_matched: bool,
 993    ) -> Self {
 994        let semver = body.semver();
 995        let time =
 996            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
 997
 998        Self {
 999            app_version: body.app_version.clone(),
1000            major: semver.map(|v| v.major() as i32),
1001            minor: semver.map(|v| v.minor() as i32),
1002            patch: semver.map(|v| v.patch() as i32),
1003            checksum_matched,
1004            release_channel: body.release_channel.clone().unwrap_or_default(),
1005            os_name: body.os_name.clone(),
1006            os_version: body.os_version.clone().unwrap_or_default(),
1007            installation_id: body.installation_id.clone(),
1008            session_id: body.session_id.clone(),
1009            is_staff: body.is_staff,
1010            time: time.timestamp_millis(),
1011            memory_in_bytes: event.memory_in_bytes,
1012            virtual_memory_in_bytes: event.virtual_memory_in_bytes,
1013        }
1014    }
1015}
1016
1017#[derive(Serialize, Debug, clickhouse::Row)]
1018pub struct AppEventRow {
1019    // AppInfoBase
1020    app_version: String,
1021    major: Option<i32>,
1022    minor: Option<i32>,
1023    patch: Option<i32>,
1024    checksum_matched: bool,
1025    release_channel: String,
1026    os_name: String,
1027    os_version: String,
1028
1029    // ClientEventBase
1030    installation_id: Option<String>,
1031    session_id: Option<String>,
1032    is_staff: Option<bool>,
1033    time: i64,
1034
1035    // AppEventRow
1036    operation: String,
1037}
1038
1039impl AppEventRow {
1040    fn from_event(
1041        event: AppEvent,
1042        wrapper: &EventWrapper,
1043        body: &EventRequestBody,
1044        first_event_at: chrono::DateTime<chrono::Utc>,
1045        checksum_matched: bool,
1046    ) -> Self {
1047        let semver = body.semver();
1048        let time =
1049            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
1050
1051        Self {
1052            app_version: body.app_version.clone(),
1053            major: semver.map(|v| v.major() as i32),
1054            minor: semver.map(|v| v.minor() as i32),
1055            patch: semver.map(|v| v.patch() as i32),
1056            checksum_matched,
1057            release_channel: body.release_channel.clone().unwrap_or_default(),
1058            os_name: body.os_name.clone(),
1059            os_version: body.os_version.clone().unwrap_or_default(),
1060            installation_id: body.installation_id.clone(),
1061            session_id: body.session_id.clone(),
1062            is_staff: body.is_staff,
1063            time: time.timestamp_millis(),
1064            operation: event.operation,
1065        }
1066    }
1067}
1068
1069#[derive(Serialize, Debug, clickhouse::Row)]
1070pub struct SettingEventRow {
1071    // AppInfoBase
1072    app_version: String,
1073    major: Option<i32>,
1074    minor: Option<i32>,
1075    patch: Option<i32>,
1076    checksum_matched: bool,
1077    release_channel: String,
1078    os_name: String,
1079    os_version: String,
1080
1081    // ClientEventBase
1082    installation_id: Option<String>,
1083    session_id: Option<String>,
1084    is_staff: Option<bool>,
1085    time: i64,
1086    // SettingEventRow
1087    setting: String,
1088    value: String,
1089}
1090
1091impl SettingEventRow {
1092    fn from_event(
1093        event: SettingEvent,
1094        wrapper: &EventWrapper,
1095        body: &EventRequestBody,
1096        first_event_at: chrono::DateTime<chrono::Utc>,
1097        checksum_matched: bool,
1098    ) -> Self {
1099        let semver = body.semver();
1100        let time =
1101            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
1102
1103        Self {
1104            app_version: body.app_version.clone(),
1105            major: semver.map(|v| v.major() as i32),
1106            minor: semver.map(|v| v.minor() as i32),
1107            checksum_matched,
1108            patch: semver.map(|v| v.patch() as i32),
1109            release_channel: body.release_channel.clone().unwrap_or_default(),
1110            os_name: body.os_name.clone(),
1111            os_version: body.os_version.clone().unwrap_or_default(),
1112            installation_id: body.installation_id.clone(),
1113            session_id: body.session_id.clone(),
1114            is_staff: body.is_staff,
1115            time: time.timestamp_millis(),
1116            setting: event.setting,
1117            value: event.value,
1118        }
1119    }
1120}
1121
1122#[derive(Serialize, Debug, clickhouse::Row)]
1123pub struct ExtensionEventRow {
1124    // AppInfoBase
1125    app_version: String,
1126    major: Option<i32>,
1127    minor: Option<i32>,
1128    patch: Option<i32>,
1129    checksum_matched: bool,
1130    release_channel: String,
1131    os_name: String,
1132    os_version: String,
1133
1134    // ClientEventBase
1135    installation_id: Option<String>,
1136    session_id: Option<String>,
1137    is_staff: Option<bool>,
1138    time: i64,
1139
1140    // ExtensionEventRow
1141    extension_id: Arc<str>,
1142    extension_version: Arc<str>,
1143    dev: bool,
1144    schema_version: Option<i32>,
1145    wasm_api_version: Option<String>,
1146}
1147
1148impl ExtensionEventRow {
1149    fn from_event(
1150        event: ExtensionEvent,
1151        wrapper: &EventWrapper,
1152        body: &EventRequestBody,
1153        extension_metadata: Option<ExtensionMetadata>,
1154        first_event_at: chrono::DateTime<chrono::Utc>,
1155        checksum_matched: bool,
1156    ) -> Self {
1157        let semver = body.semver();
1158        let time =
1159            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
1160
1161        Self {
1162            app_version: body.app_version.clone(),
1163            major: semver.map(|v| v.major() as i32),
1164            minor: semver.map(|v| v.minor() as i32),
1165            patch: semver.map(|v| v.patch() as i32),
1166            checksum_matched,
1167            release_channel: body.release_channel.clone().unwrap_or_default(),
1168            os_name: body.os_name.clone(),
1169            os_version: body.os_version.clone().unwrap_or_default(),
1170            installation_id: body.installation_id.clone(),
1171            session_id: body.session_id.clone(),
1172            is_staff: body.is_staff,
1173            time: time.timestamp_millis(),
1174            extension_id: event.extension_id,
1175            extension_version: event.version,
1176            dev: extension_metadata.is_none(),
1177            schema_version: extension_metadata
1178                .as_ref()
1179                .and_then(|metadata| metadata.manifest.schema_version),
1180            wasm_api_version: extension_metadata.as_ref().and_then(|metadata| {
1181                metadata
1182                    .manifest
1183                    .wasm_api_version
1184                    .as_ref()
1185                    .map(|version| version.to_string())
1186            }),
1187        }
1188    }
1189}
1190
1191#[derive(Serialize, Debug, clickhouse::Row)]
1192pub struct EditEventRow {
1193    // AppInfoBase
1194    app_version: String,
1195    major: Option<i32>,
1196    minor: Option<i32>,
1197    patch: Option<i32>,
1198    checksum_matched: bool,
1199    release_channel: String,
1200    os_name: String,
1201    os_version: String,
1202
1203    // ClientEventBase
1204    installation_id: Option<String>,
1205    // Note: This column name has a typo in the ClickHouse table.
1206    #[serde(rename = "sesssion_id")]
1207    session_id: Option<String>,
1208    is_staff: Option<bool>,
1209    time: i64,
1210
1211    // EditEventRow
1212    period_start: i64,
1213    period_end: i64,
1214    environment: String,
1215}
1216
1217impl EditEventRow {
1218    fn from_event(
1219        event: EditEvent,
1220        wrapper: &EventWrapper,
1221        body: &EventRequestBody,
1222        first_event_at: chrono::DateTime<chrono::Utc>,
1223        checksum_matched: bool,
1224    ) -> Self {
1225        let semver = body.semver();
1226        let time =
1227            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
1228
1229        let period_start = time - chrono::Duration::milliseconds(event.duration);
1230        let period_end = time;
1231
1232        Self {
1233            app_version: body.app_version.clone(),
1234            major: semver.map(|v| v.major() as i32),
1235            minor: semver.map(|v| v.minor() as i32),
1236            patch: semver.map(|v| v.patch() as i32),
1237            checksum_matched,
1238            release_channel: body.release_channel.clone().unwrap_or_default(),
1239            os_name: body.os_name.clone(),
1240            os_version: body.os_version.clone().unwrap_or_default(),
1241            installation_id: body.installation_id.clone(),
1242            session_id: body.session_id.clone(),
1243            is_staff: body.is_staff,
1244            time: time.timestamp_millis(),
1245            period_start: period_start.timestamp_millis(),
1246            period_end: period_end.timestamp_millis(),
1247            environment: event.environment,
1248        }
1249    }
1250}
1251
1252#[derive(Serialize, Debug, clickhouse::Row)]
1253pub struct ActionEventRow {
1254    // AppInfoBase
1255    app_version: String,
1256    major: Option<i32>,
1257    minor: Option<i32>,
1258    patch: Option<i32>,
1259    checksum_matched: bool,
1260    release_channel: String,
1261    os_name: String,
1262    os_version: String,
1263
1264    // ClientEventBase
1265    installation_id: Option<String>,
1266    // Note: This column name has a typo in the ClickHouse table.
1267    #[serde(rename = "sesssion_id")]
1268    session_id: Option<String>,
1269    is_staff: Option<bool>,
1270    time: i64,
1271    // ActionEventRow
1272    source: String,
1273    action: String,
1274}
1275
1276impl ActionEventRow {
1277    fn from_event(
1278        event: ActionEvent,
1279        wrapper: &EventWrapper,
1280        body: &EventRequestBody,
1281        first_event_at: chrono::DateTime<chrono::Utc>,
1282        checksum_matched: bool,
1283    ) -> Self {
1284        let semver = body.semver();
1285        let time =
1286            first_event_at + chrono::Duration::milliseconds(wrapper.milliseconds_since_first_event);
1287
1288        Self {
1289            app_version: body.app_version.clone(),
1290            major: semver.map(|v| v.major() as i32),
1291            minor: semver.map(|v| v.minor() as i32),
1292            patch: semver.map(|v| v.patch() as i32),
1293            checksum_matched,
1294            release_channel: body.release_channel.clone().unwrap_or_default(),
1295            os_name: body.os_name.clone(),
1296            os_version: body.os_version.clone().unwrap_or_default(),
1297            installation_id: body.installation_id.clone(),
1298            session_id: body.session_id.clone(),
1299            is_staff: body.is_staff,
1300            time: time.timestamp_millis(),
1301            source: event.source,
1302            action: event.action,
1303        }
1304    }
1305}
1306
1307pub fn calculate_json_checksum(app: Arc<AppState>, json: &impl AsRef<[u8]>) -> Option<Vec<u8>> {
1308    let Some(checksum_seed) = app.config.zed_client_checksum_seed.as_ref() else {
1309        return None;
1310    };
1311
1312    let mut summer = Sha256::new();
1313    summer.update(checksum_seed);
1314    summer.update(&json);
1315    summer.update(checksum_seed);
1316    Some(summer.finalize().into_iter().collect())
1317}