use super::ips_file::IpsFile;use crate::api::CloudflareIpCountryHeader;use crate::clickhouse::write_to_table;use crate::{api::slack, AppState, Error, Result};use anyhow::{anyhow, Context};use aws_sdk_s3::primitives::ByteStream;use axum::{    body::Bytes,    headers::Header,    http::{HeaderMap, HeaderName, StatusCode},    routing::post,    Extension, Router, TypedHeader,};use rpc::ExtensionMetadata;use semantic_version::SemanticVersion;use serde::{Serialize, Serializer};use sha2::{Digest, Sha256};use std::sync::{Arc, OnceLock};use telemetry_events::{    ActionEvent, AppEvent, AssistantEvent, CallEvent, CpuEvent, EditEvent, EditorEvent, Event,    EventRequestBody, EventWrapper, ExtensionEvent, InlineCompletionEvent, MemoryEvent, ReplEvent,    SettingEvent,};use uuid::Uuid; static CRASH_REPORTS_BUCKET: &str = "zed-crash-reports"; pub fn router() -> Router {    Router::new()        .route("/telemetry/events", post(post_events))        .route("/telemetry/crashes", post(post_crash))        .route("/telemetry/panics", post(post_panic))        .route("/telemetry/hangs", post(post_hang))} pub struct ZedChecksumHeader(Vec<u8>); impl Header for ZedChecksumHeader {    fn name() -> &'static HeaderName {        static ZED_CHECKSUM_HEADER: OnceLock<HeaderName> = OnceLock::new();        ZED_CHECKSUM_HEADER.get_or_init(|| HeaderName::from_static("x-zed-checksum"))    }     fn decode<'i, I>(values: &mut I) -> Result<Self, axum::headers::Error>    where        Self: Sized,        I: Iterator<Item = &'i axum::http::HeaderValue>,    {        let checksum = values            .next()            .ok_or_else(axum::headers::Error::invalid)?            .to_str()            .map_err(|_| axum::headers::Error::invalid())?;         let bytes = hex::decode(checksum).map_err(|_| axum::headers::Error::invalid())?;        Ok(Self(bytes))    }     fn encode<E: Extend<axum::http::HeaderValue>>(&self, _values: &mut E) {        unimplemented!()    }} pub async fn post_crash(    Extension(app): Extension<Arc<AppState>>,    headers: HeaderMap,    body: Bytes,) -> Result<()> {    let report = IpsFile::parse(&body)?;    let version_threshold = SemanticVersion::new(0, 123, 0);     let bundle_id = &report.header.bundle_id;    let app_version = &report.app_version();     if bundle_id == "dev.zed.Zed-Dev" {        log::error!("Crash uploads from {} are ignored.", bundle_id);        return Ok(());    }     if app_version.is_none() || app_version.unwrap() < version_threshold {        log::error!(            "Crash uploads from {} are ignored.",            report.header.app_version        );        return Ok(());    }    let app_version = app_version.unwrap();     if let Some(blob_store_client) = app.blob_store_client.as_ref() {        let response = blob_store_client            .head_object()            .bucket(CRASH_REPORTS_BUCKET)            .key(report.header.incident_id.clone() + ".ips")            .send()            .await;         if response.is_ok() {            log::info!("We've already uploaded this crash");            return Ok(());        }         blob_store_client            .put_object()            .bucket(CRASH_REPORTS_BUCKET)            .key(report.header.incident_id.clone() + ".ips")            .acl(aws_sdk_s3::types::ObjectCannedAcl::PublicRead)            .body(ByteStream::from(body.to_vec()))            .send()            .await            .map_err(|e| log::error!("Failed to upload crash: {}", e))            .ok();    }     let recent_panic_on: Option<i64> = headers        .get("x-zed-panicked-on")        .and_then(|h| h.to_str().ok())        .and_then(|s| s.parse().ok());     let installation_id = headers        .get("x-zed-installation-id")        .and_then(|h| h.to_str().ok())        .map(|s| s.to_string())        .unwrap_or_default();     let mut recent_panic = None;     if let Some(recent_panic_on) = recent_panic_on {        let crashed_at = match report.timestamp() {            Ok(t) => Some(t),            Err(e) => {                log::error!("Can't parse {}: {}", report.header.timestamp, e);                None            }        };        if crashed_at.is_some_and(|t| (t.timestamp_millis() - recent_panic_on).abs() <= 30000) {            recent_panic = headers.get("x-zed-panic").and_then(|h| h.to_str().ok());        }    }     let description = report.description(recent_panic);    let summary = report.backtrace_summary();     tracing::error!(        service = "client",        version = %report.header.app_version,        os_version = %report.header.os_version,        bundle_id = %report.header.bundle_id,        incident_id = %report.header.incident_id,        installation_id = %installation_id,        description = %description,        backtrace = %summary,        "crash report"    );     if let Some(slack_panics_webhook) = app.config.slack_panics_webhook.clone() {        let payload = slack::WebhookBody::new(|w| {            w.add_section(|s| s.text(slack::Text::markdown(description)))                .add_section(|s| {                    s.add_field(slack::Text::markdown(format!(                        "*Version:*\n{} ({})",                        bundle_id, app_version                    )))                    .add_field({                        let hostname = app.config.blob_store_url.clone().unwrap_or_default();                        let hostname = hostname.strip_prefix("https://").unwrap_or_else(|| {                            hostname.strip_prefix("http://").unwrap_or_default()                        });                         slack::Text::markdown(format!(                            "*Incident:*\n<https://{}.{}/{}.ips|{}…>",                            CRASH_REPORTS_BUCKET,                            hostname,                            report.header.incident_id,                            report                                .header                                .incident_id                                .chars()                                .take(8)                                .collect::<String>(),                        ))                    })                })                .add_rich_text(|r| r.add_preformatted(|p| p.add_text(summary)))        });        let payload_json = serde_json::to_string(&payload).map_err(|err| {            log::error!("Failed to serialize payload to JSON: {err}");            Error::Internal(anyhow!(err))        })?;         reqwest::Client::new()            .post(slack_panics_webhook)            .header("Content-Type", "application/json")            .body(payload_json)            .send()            .await            .map_err(|err| {                log::error!("Failed to send payload to Slack: {err}");                Error::Internal(anyhow!(err))            })?;    }     Ok(())} pub async fn post_hang(    Extension(app): Extension<Arc<AppState>>,    TypedHeader(ZedChecksumHeader(checksum)): TypedHeader<ZedChecksumHeader>,    body: Bytes,) -> Result<()> {    let Some(expected) = calculate_json_checksum(app.clone(), &body) else {        return Err(Error::http(            StatusCode::INTERNAL_SERVER_ERROR,            "events not enabled".into(),        ))?;    };     if checksum != expected {        return Err(Error::http(            StatusCode::BAD_REQUEST,            "invalid checksum".into(),        ))?;    }     let incident_id = Uuid::new_v4().to_string();     // dump JSON into S3 so we can get frame offsets if we need to.    if let Some(blob_store_client) = app.blob_store_client.as_ref() {        blob_store_client            .put_object()            .bucket(CRASH_REPORTS_BUCKET)            .key(incident_id.clone() + ".hang.json")            .acl(aws_sdk_s3::types::ObjectCannedAcl::PublicRead)            .body(ByteStream::from(body.to_vec()))            .send()            .await            .map_err(|e| log::error!("Failed to upload crash: {}", e))            .ok();    }     let report: telemetry_events::HangReport = serde_json::from_slice(&body).map_err(|err| {        log::error!("can't parse report json: {err}");        Error::Internal(anyhow!(err))    })?;     let mut backtrace = "Possible hang detected on main thread:".to_string();    let unknown = "<unknown>".to_string();    for frame in report.backtrace.iter() {        backtrace.push_str(&format!("\n{}", frame.symbols.first().unwrap_or(&unknown)));    }     tracing::error!(        service = "client",        version = %report.app_version.unwrap_or_default().to_string(),        os_name = %report.os_name,        os_version = report.os_version.unwrap_or_default().to_string(),        incident_id = %incident_id,        installation_id = %report.installation_id.unwrap_or_default(),        backtrace = %backtrace,        "hang report");     Ok(())} pub async fn post_panic(    Extension(app): Extension<Arc<AppState>>,    TypedHeader(ZedChecksumHeader(checksum)): TypedHeader<ZedChecksumHeader>,    body: Bytes,) -> Result<()> {    let Some(expected) = calculate_json_checksum(app.clone(), &body) else {        return Err(Error::http(            StatusCode::INTERNAL_SERVER_ERROR,            "events not enabled".into(),        ))?;    };     if checksum != expected {        return Err(Error::http(            StatusCode::BAD_REQUEST,            "invalid checksum".into(),        ))?;    }     let report: telemetry_events::PanicRequest = serde_json::from_slice(&body)        .map_err(|_| Error::http(StatusCode::BAD_REQUEST, "invalid json".into()))?;    let panic = report.panic;     if panic.os_name == "Linux" && panic.os_version == Some("1.0.0".to_string()) {        return Err(Error::http(            StatusCode::BAD_REQUEST,            "invalid os version".into(),        ))?;    }     tracing::error!(        service = "client",        version = %panic.app_version,        os_name = %panic.os_name,        os_version = %panic.os_version.clone().unwrap_or_default(),        installation_id = %panic.installation_id.unwrap_or_default(),        description = %panic.payload,        backtrace = %panic.backtrace.join("\n"),        "panic report");     let backtrace = if panic.backtrace.len() > 25 {        let total = panic.backtrace.len();        format!(            "{}\n   and {} more",            panic                .backtrace                .iter()                .take(20)                .cloned()                .collect::<Vec<_>>()                .join("\n"),            total - 20        )    } else {        panic.backtrace.join("\n")    };    let backtrace_with_summary = panic.payload + "\n" + &backtrace;     if let Some(slack_panics_webhook) = app.config.slack_panics_webhook.clone() {        let payload = slack::WebhookBody::new(|w| {            w.add_section(|s| s.text(slack::Text::markdown("Panic request".to_string())))                .add_section(|s| {                    s.add_field(slack::Text::markdown(format!(                        "*Version:*\n {} ",                        panic.app_version                    )))                    .add_field({                        slack::Text::markdown(format!(                            "*OS:*\n{} {}",                            panic.os_name,                            panic.os_version.unwrap_or_default()                        ))                    })                })                .add_rich_text(|r| r.add_preformatted(|p| p.add_text(backtrace_with_summary)))        });        let payload_json = serde_json::to_string(&payload).map_err(|err| {            log::error!("Failed to serialize payload to JSON: {err}");            Error::Internal(anyhow!(err))        })?;         reqwest::Client::new()            .post(slack_panics_webhook)            .header("Content-Type", "application/json")            .body(payload_json)            .send()            .await            .map_err(|err| {                log::error!("Failed to send payload to Slack: {err}");                Error::Internal(anyhow!(err))            })?;    }     Ok(())} pub async fn post_events(    Extension(app): Extension<Arc<AppState>>,    TypedHeader(ZedChecksumHeader(checksum)): TypedHeader<ZedChecksumHeader>,    country_code_header: Option<TypedHeader<CloudflareIpCountryHeader>>,    body: Bytes,) -> Result<()> {    let Some(clickhouse_client) = app.clickhouse_client.clone() else {        Err(Error::http(            StatusCode::NOT_IMPLEMENTED,            "not supported".into(),        ))?    };     let Some(expected) = calculate_json_checksum(app.clone(), &body) else {        return Err(Error::http(            StatusCode::INTERNAL_SERVER_ERROR,            "events not enabled".into(),        ))?;    };     let checksum_matched = checksum == expected;     let request_body: telemetry_events::EventRequestBody =        serde_json::from_slice(&body).map_err(|err| {            log::error!("can't parse event json: {err}");            Error::Internal(anyhow!(err))        })?;     let mut to_upload = ToUpload::default();    let Some(last_event) = request_body.events.last() else {        return Err(Error::http(StatusCode::BAD_REQUEST, "no events".into()))?;    };    let country_code = country_code_header.map(|h| h.to_string());     let first_event_at = chrono::Utc::now()        - chrono::Duration::milliseconds(last_event.milliseconds_since_first_event);     for wrapper in &request_body.events {        match &wrapper.event {            Event::Editor(event) => to_upload.editor_events.push(EditorEventRow::from_event(                event.clone(),                wrapper,                &request_body,                first_event_at,                country_code.clone(),                checksum_matched,            )),            // Needed for clients sending old copilot_event types            Event::Copilot(_) => {}            Event::InlineCompletion(event) => {                to_upload                    .inline_completion_events                    .push(InlineCompletionEventRow::from_event(                        event.clone(),                        wrapper,                        &request_body,                        first_event_at,                        country_code.clone(),                        checksum_matched,                    ))            }            Event::Call(event) => to_upload.call_events.push(CallEventRow::from_event(                event.clone(),                wrapper,                &request_body,                first_event_at,                checksum_matched,            )),            Event::Assistant(event) => {                to_upload                    .assistant_events                    .push(AssistantEventRow::from_event(                        event.clone(),                        wrapper,                        &request_body,                        first_event_at,                        checksum_matched,                    ))            }            Event::Cpu(event) => to_upload.cpu_events.push(CpuEventRow::from_event(                event.clone(),                wrapper,                &request_body,                first_event_at,                checksum_matched,            )),            Event::Memory(event) => to_upload.memory_events.push(MemoryEventRow::from_event(                event.clone(),                wrapper,                &request_body,                first_event_at,                checksum_matched,            )),            Event::App(event) => to_upload.app_events.push(AppEventRow::from_event(                event.clone(),                wrapper,                &request_body,                first_event_at,                checksum_matched,            )),            Event::Setting(event) => to_upload.setting_events.push(SettingEventRow::from_event(                event.clone(),                wrapper,                &request_body,                first_event_at,                checksum_matched,            )),            Event::Edit(event) => to_upload.edit_events.push(EditEventRow::from_event(                event.clone(),                wrapper,                &request_body,                first_event_at,                checksum_matched,            )),            Event::Action(event) => to_upload.action_events.push(ActionEventRow::from_event(                event.clone(),                wrapper,                &request_body,                first_event_at,                checksum_matched,            )),            Event::Extension(event) => {                let metadata = app                    .db                    .get_extension_version(&event.extension_id, &event.version)                    .await?;                to_upload                    .extension_events                    .push(ExtensionEventRow::from_event(                        event.clone(),                        wrapper,                        &request_body,                        metadata,                        first_event_at,                        checksum_matched,                    ))            }            Event::Repl(event) => to_upload.repl_events.push(ReplEventRow::from_event(                event.clone(),                wrapper,                &request_body,

Lines 1–500 of 1353. Load more