Merge remote-tracking branch 'origin/main' into room

crates/client/src/channel.rs

@@ -601,7 +601,7 @@ mod tests {
 
         let user_id = 5;
         let http_client = FakeHttpClient::with_404_response();
-        let client = Client::new(http_client.clone());
+        let client = cx.update(|cx| Client::new(http_client.clone(), cx));
         let server = FakeServer::for_client(user_id, &client, cx).await;
 
         Channel::init(&client);

crates/client/src/client.rs

@@ -3,6 +3,7 @@ pub mod test;
 
 pub mod channel;
 pub mod http;
+pub mod telemetry;
 pub mod user;
 
 use anyhow::{anyhow, Context, Result};
@@ -11,10 +12,12 @@ use async_tungstenite::tungstenite::{
     error::Error as WebsocketError,
     http::{Request, StatusCode},
 };
+use db::Db;
 use futures::{future::LocalBoxFuture, FutureExt, SinkExt, StreamExt, TryStreamExt};
 use gpui::{
-    actions, AnyModelHandle, AnyViewHandle, AnyWeakModelHandle, AnyWeakViewHandle, AsyncAppContext,
-    Entity, ModelContext, ModelHandle, MutableAppContext, Task, View, ViewContext, ViewHandle,
+    actions, serde_json::Value, AnyModelHandle, AnyViewHandle, AnyWeakModelHandle,
+    AnyWeakViewHandle, AppContext, AsyncAppContext, Entity, ModelContext, ModelHandle,
+    MutableAppContext, Task, View, ViewContext, ViewHandle,
 };
 use http::HttpClient;
 use lazy_static::lazy_static;
@@ -28,9 +31,11 @@ use std::{
     convert::TryFrom,
     fmt::Write as _,
     future::Future,
+    path::PathBuf,
     sync::{Arc, Weak},
     time::{Duration, Instant},
 };
+use telemetry::Telemetry;
 use thiserror::Error;
 use url::Url;
 use util::{ResultExt, TryFutureExt};
@@ -51,11 +56,16 @@ pub const ZED_SECRET_CLIENT_TOKEN: &str = "618033988749894";
 
 actions!(client, [Authenticate]);
 
-pub fn init(rpc: Arc<Client>, cx: &mut MutableAppContext) {
-    cx.add_global_action(move |_: &Authenticate, cx| {
-        let rpc = rpc.clone();
-        cx.spawn(|cx| async move { rpc.authenticate_and_connect(true, &cx).log_err().await })
+pub fn init(client: Arc<Client>, cx: &mut MutableAppContext) {
+    cx.add_global_action({
+        let client = client.clone();
+        move |_: &Authenticate, cx| {
+            let client = client.clone();
+            cx.spawn(
+                |cx| async move { client.authenticate_and_connect(true, &cx).log_err().await },
+            )
             .detach();
+        }
     });
 }
 
@@ -63,6 +73,7 @@ pub struct Client {
     id: usize,
     peer: Arc<Peer>,
     http: Arc<dyn HttpClient>,
+    telemetry: Arc<Telemetry>,
     state: RwLock<ClientState>,
 
     #[allow(clippy::type_complexity)]
@@ -232,10 +243,11 @@ impl Drop for Subscription {
 }
 
 impl Client {
-    pub fn new(http: Arc<dyn HttpClient>) -> Arc<Self> {
+    pub fn new(http: Arc<dyn HttpClient>, cx: &AppContext) -> Arc<Self> {
         Arc::new(Self {
             id: 0,
             peer: Peer::new(),
+            telemetry: Telemetry::new(http.clone(), cx),
             http,
             state: Default::default(),
 
@@ -339,6 +351,7 @@ impl Client {
                 }));
             }
             Status::SignedOut | Status::UpgradeRequired => {
+                self.telemetry.set_authenticated_user_info(None, false);
                 state._reconnect_task.take();
             }
             _ => {}
@@ -618,6 +631,9 @@ impl Client {
         if credentials.is_none() && try_keychain {
             credentials = read_credentials_from_keychain(cx);
             read_from_keychain = credentials.is_some();
+            if read_from_keychain {
+                self.report_event("read credentials from keychain", Default::default());
+            }
         }
         if credentials.is_none() {
             let mut status_rx = self.status();
@@ -901,6 +917,7 @@ impl Client {
     ) -> Task<Result<Credentials>> {
         let platform = cx.platform();
         let executor = cx.background();
+        let telemetry = self.telemetry.clone();
         executor.clone().spawn(async move {
             // Generate a pair of asymmetric encryption keys. The public key will be used by the
             // zed server to encrypt the user's access token, so that it can'be intercepted by
@@ -979,6 +996,8 @@ impl Client {
                 .context("failed to decrypt access token")?;
             platform.activate(true);
 
+            telemetry.report_event("authenticate with browser", Default::default());
+
             Ok(Credentials {
                 user_id: user_id.parse()?,
                 access_token,
@@ -1043,6 +1062,18 @@ impl Client {
         log::debug!("rpc respond. client_id:{}. name:{}", self.id, T::NAME);
         self.peer.respond_with_error(receipt, error)
     }
+
+    pub fn start_telemetry(&self, db: Arc<Db>) {
+        self.telemetry.start(db);
+    }
+
+    pub fn report_event(&self, kind: &str, properties: Value) {
+        self.telemetry.report_event(kind, properties)
+    }
+
+    pub fn telemetry_log_file_path(&self) -> Option<PathBuf> {
+        self.telemetry.log_file_path()
+    }
 }
 
 impl AnyWeakEntityHandle {
@@ -1108,7 +1139,7 @@ mod tests {
         cx.foreground().forbid_parking();
 
         let user_id = 5;
-        let client = Client::new(FakeHttpClient::with_404_response());
+        let client = cx.update(|cx| Client::new(FakeHttpClient::with_404_response(), cx));
         let server = FakeServer::for_client(user_id, &client, cx).await;
         let mut status = client.status();
         assert!(matches!(
@@ -1147,7 +1178,7 @@ mod tests {
 
         let auth_count = Arc::new(Mutex::new(0));
         let dropped_auth_count = Arc::new(Mutex::new(0));
-        let client = Client::new(FakeHttpClient::with_404_response());
+        let client = cx.update(|cx| Client::new(FakeHttpClient::with_404_response(), cx));
         client.override_authenticate({
             let auth_count = auth_count.clone();
             let dropped_auth_count = dropped_auth_count.clone();
@@ -1196,7 +1227,7 @@ mod tests {
         cx.foreground().forbid_parking();
 
         let user_id = 5;
-        let client = Client::new(FakeHttpClient::with_404_response());
+        let client = cx.update(|cx| Client::new(FakeHttpClient::with_404_response(), cx));
         let server = FakeServer::for_client(user_id, &client, cx).await;
 
         let (done_tx1, mut done_rx1) = smol::channel::unbounded();
@@ -1242,7 +1273,7 @@ mod tests {
         cx.foreground().forbid_parking();
 
         let user_id = 5;
-        let client = Client::new(FakeHttpClient::with_404_response());
+        let client = cx.update(|cx| Client::new(FakeHttpClient::with_404_response(), cx));
         let server = FakeServer::for_client(user_id, &client, cx).await;
 
         let model = cx.add_model(|_| Model::default());
@@ -1270,7 +1301,7 @@ mod tests {
         cx.foreground().forbid_parking();
 
         let user_id = 5;
-        let client = Client::new(FakeHttpClient::with_404_response());
+        let client = cx.update(|cx| Client::new(FakeHttpClient::with_404_response(), cx));
         let server = FakeServer::for_client(user_id, &client, cx).await;
 
         let model = cx.add_model(|_| Model::default());

crates/client/src/telemetry.rs

@@ -0,0 +1,283 @@
+use crate::http::HttpClient;
+use db::Db;
+use gpui::{
+    executor::Background,
+    serde_json::{self, value::Map, Value},
+    AppContext, Task,
+};
+use isahc::Request;
+use lazy_static::lazy_static;
+use parking_lot::Mutex;
+use serde::Serialize;
+use serde_json::json;
+use std::{
+    io::Write,
+    mem,
+    path::PathBuf,
+    sync::Arc,
+    time::{Duration, SystemTime, UNIX_EPOCH},
+};
+use tempfile::NamedTempFile;
+use util::{post_inc, ResultExt, TryFutureExt};
+use uuid::Uuid;
+
+pub struct Telemetry {
+    http_client: Arc<dyn HttpClient>,
+    executor: Arc<Background>,
+    session_id: u128,
+    state: Mutex<TelemetryState>,
+}
+
+#[derive(Default)]
+struct TelemetryState {
+    metrics_id: Option<Arc<str>>,
+    device_id: Option<Arc<str>>,
+    app_version: Option<Arc<str>>,
+    os_version: Option<Arc<str>>,
+    os_name: &'static str,
+    queue: Vec<AmplitudeEvent>,
+    next_event_id: usize,
+    flush_task: Option<Task<()>>,
+    log_file: Option<NamedTempFile>,
+}
+
+const AMPLITUDE_EVENTS_URL: &'static str = "https://api2.amplitude.com/batch";
+
+lazy_static! {
+    static ref AMPLITUDE_API_KEY: Option<String> = std::env::var("ZED_AMPLITUDE_API_KEY")
+        .ok()
+        .or_else(|| option_env!("ZED_AMPLITUDE_API_KEY").map(|key| key.to_string()));
+}
+
+#[derive(Serialize)]
+struct AmplitudeEventBatch {
+    api_key: &'static str,
+    events: Vec<AmplitudeEvent>,
+}
+
+#[derive(Serialize)]
+struct AmplitudeEvent {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    user_id: Option<Arc<str>>,
+    device_id: Option<Arc<str>>,
+    event_type: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    event_properties: Option<Map<String, Value>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    user_properties: Option<Map<String, Value>>,
+    os_name: &'static str,
+    os_version: Option<Arc<str>>,
+    app_version: Option<Arc<str>>,
+    platform: &'static str,
+    event_id: usize,
+    session_id: u128,
+    time: u128,
+}
+
+#[cfg(debug_assertions)]
+const MAX_QUEUE_LEN: usize = 1;
+
+#[cfg(not(debug_assertions))]
+const MAX_QUEUE_LEN: usize = 10;
+
+#[cfg(debug_assertions)]
+const DEBOUNCE_INTERVAL: Duration = Duration::from_secs(1);
+
+#[cfg(not(debug_assertions))]
+const DEBOUNCE_INTERVAL: Duration = Duration::from_secs(30);
+
+impl Telemetry {
+    pub fn new(client: Arc<dyn HttpClient>, cx: &AppContext) -> Arc<Self> {
+        let platform = cx.platform();
+        let this = Arc::new(Self {
+            http_client: client,
+            executor: cx.background().clone(),
+            session_id: SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_millis(),
+            state: Mutex::new(TelemetryState {
+                os_version: platform
+                    .os_version()
+                    .log_err()
+                    .map(|v| v.to_string().into()),
+                os_name: platform.os_name().into(),
+                app_version: platform
+                    .app_version()
+                    .log_err()
+                    .map(|v| v.to_string().into()),
+                device_id: None,
+                queue: Default::default(),
+                flush_task: Default::default(),
+                next_event_id: 0,
+                log_file: None,
+                metrics_id: None,
+            }),
+        });
+
+        if AMPLITUDE_API_KEY.is_some() {
+            this.executor
+                .spawn({
+                    let this = this.clone();
+                    async move {
+                        if let Some(tempfile) = NamedTempFile::new().log_err() {
+                            this.state.lock().log_file = Some(tempfile);
+                        }
+                    }
+                })
+                .detach();
+        }
+
+        this
+    }
+
+    pub fn log_file_path(&self) -> Option<PathBuf> {
+        Some(self.state.lock().log_file.as_ref()?.path().to_path_buf())
+    }
+
+    pub fn start(self: &Arc<Self>, db: Arc<Db>) {
+        let this = self.clone();
+        self.executor
+            .spawn(
+                async move {
+                    let device_id = if let Some(device_id) = db
+                        .read(["device_id"])?
+                        .into_iter()
+                        .flatten()
+                        .next()
+                        .and_then(|bytes| String::from_utf8(bytes).ok())
+                    {
+                        device_id
+                    } else {
+                        let device_id = Uuid::new_v4().to_string();
+                        db.write([("device_id", device_id.as_bytes())])?;
+                        device_id
+                    };
+
+                    let device_id = Some(Arc::from(device_id));
+                    let mut state = this.state.lock();
+                    state.device_id = device_id.clone();
+                    for event in &mut state.queue {
+                        event.device_id = device_id.clone();
+                    }
+                    if !state.queue.is_empty() {
+                        drop(state);
+                        this.flush();
+                    }
+
+                    anyhow::Ok(())
+                }
+                .log_err(),
+            )
+            .detach();
+    }
+
+    pub fn set_authenticated_user_info(
+        self: &Arc<Self>,
+        metrics_id: Option<String>,
+        is_staff: bool,
+    ) {
+        let is_signed_in = metrics_id.is_some();
+        self.state.lock().metrics_id = metrics_id.map(|s| s.into());
+        if is_signed_in {
+            self.report_event_with_user_properties(
+                "$identify",
+                Default::default(),
+                json!({ "$set": { "staff": is_staff } }),
+            )
+        }
+    }
+
+    pub fn report_event(self: &Arc<Self>, kind: &str, properties: Value) {
+        self.report_event_with_user_properties(kind, properties, Default::default());
+    }
+
+    fn report_event_with_user_properties(
+        self: &Arc<Self>,
+        kind: &str,
+        properties: Value,
+        user_properties: Value,
+    ) {
+        if AMPLITUDE_API_KEY.is_none() {
+            return;
+        }
+
+        let mut state = self.state.lock();
+        let event = AmplitudeEvent {
+            event_type: kind.to_string(),
+            time: SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_millis(),
+            session_id: self.session_id,
+            event_properties: if let Value::Object(properties) = properties {
+                Some(properties)
+            } else {
+                None
+            },
+            user_properties: if let Value::Object(user_properties) = user_properties {
+                Some(user_properties)
+            } else {
+                None
+            },
+            user_id: state.metrics_id.clone(),
+            device_id: state.device_id.clone(),
+            os_name: state.os_name,
+            platform: "Zed",
+            os_version: state.os_version.clone(),
+            app_version: state.app_version.clone(),
+            event_id: post_inc(&mut state.next_event_id),
+        };
+        state.queue.push(event);
+        if state.device_id.is_some() {
+            if state.queue.len() >= MAX_QUEUE_LEN {
+                drop(state);
+                self.flush();
+            } else {
+                let this = self.clone();
+                let executor = self.executor.clone();
+                state.flush_task = Some(self.executor.spawn(async move {
+                    executor.timer(DEBOUNCE_INTERVAL).await;
+                    this.flush();
+                }));
+            }
+        }
+    }
+
+    fn flush(self: &Arc<Self>) {
+        let mut state = self.state.lock();
+        let events = mem::take(&mut state.queue);
+        state.flush_task.take();
+        drop(state);
+
+        if let Some(api_key) = AMPLITUDE_API_KEY.as_ref() {
+            let this = self.clone();
+            self.executor
+                .spawn(
+                    async move {
+                        let mut json_bytes = Vec::new();
+
+                        if let Some(file) = &mut this.state.lock().log_file {
+                            let file = file.as_file_mut();
+                            for event in &events {
+                                json_bytes.clear();
+                                serde_json::to_writer(&mut json_bytes, event)?;
+                                file.write_all(&json_bytes)?;
+                                file.write(b"\n")?;
+                            }
+                        }
+
+                        let batch = AmplitudeEventBatch { api_key, events };
+                        json_bytes.clear();
+                        serde_json::to_writer(&mut json_bytes, &batch)?;
+                        let request =
+                            Request::post(AMPLITUDE_EVENTS_URL).body(json_bytes.into())?;
+                        this.http_client.send(request).await?;
+                        Ok(())
+                    }
+                    .log_err(),
+                )
+                .detach();
+        }
+    }
+}

crates/client/src/test.rs

@@ -6,7 +6,10 @@ use anyhow::{anyhow, Result};
 use futures::{future::BoxFuture, stream::BoxStream, Future, StreamExt};
 use gpui::{executor, ModelHandle, TestAppContext};
 use parking_lot::Mutex;
-use rpc::{proto, ConnectionId, Peer, Receipt, TypedEnvelope};
+use rpc::{
+    proto::{self, GetPrivateUserInfo, GetPrivateUserInfoResponse},
+    ConnectionId, Peer, Receipt, TypedEnvelope,
+};
 use std::{fmt, rc::Rc, sync::Arc};
 
 pub struct FakeServer {
@@ -93,6 +96,7 @@ impl FakeServer {
             .authenticate_and_connect(false, &cx.to_async())
             .await
             .unwrap();
+
         server
     }
 
@@ -126,26 +130,45 @@ impl FakeServer {
     #[allow(clippy::await_holding_lock)]
     pub async fn receive<M: proto::EnvelopedMessage>(&self) -> Result<TypedEnvelope<M>> {
         self.executor.start_waiting();
-        let message = self
-            .state
-            .lock()
-            .incoming
-            .as_mut()
-            .expect("not connected")
-            .next()
-            .await
-            .ok_or_else(|| anyhow!("other half hung up"))?;
-        self.executor.finish_waiting();
-        let type_name = message.payload_type_name();
-        Ok(*message
-            .into_any()
-            .downcast::<TypedEnvelope<M>>()
-            .unwrap_or_else(|_| {
-                panic!(
-                    "fake server received unexpected message type: {:?}",
-                    type_name
-                );
-            }))
+
+        loop {
+            let message = self
+                .state
+                .lock()
+                .incoming
+                .as_mut()
+                .expect("not connected")
+                .next()
+                .await
+                .ok_or_else(|| anyhow!("other half hung up"))?;
+            self.executor.finish_waiting();
+            let type_name = message.payload_type_name();
+            let message = message.into_any();
+
+            if message.is::<TypedEnvelope<M>>() {
+                return Ok(*message.downcast().unwrap());
+            }
+
+            if message.is::<TypedEnvelope<GetPrivateUserInfo>>() {
+                self.respond(
+                    message
+                        .downcast::<TypedEnvelope<GetPrivateUserInfo>>()
+                        .unwrap()
+                        .receipt(),
+                    GetPrivateUserInfoResponse {
+                        metrics_id: "the-metrics-id".into(),
+                        staff: false,
+                    },
+                )
+                .await;
+                continue;
+            }
+
+            panic!(
+                "fake server received unexpected message type: {:?}",
+                type_name
+            );
+        }
     }
 
     pub async fn respond<T: proto::RequestMessage>(

crates/client/src/user.rs

@@ -135,10 +135,21 @@ impl UserStore {
                     match status {
                         Status::Connected { .. } => {
                             if let Some((this, user_id)) = this.upgrade(&cx).zip(client.user_id()) {
-                                let user = this
+                                let fetch_user = this
                                     .update(&mut cx, |this, cx| this.get_user(user_id, cx))
-                                    .log_err()
-                                    .await;
+                                    .log_err();
+                                let fetch_metrics_id =
+                                    client.request(proto::GetPrivateUserInfo {}).log_err();
+                                let (user, info) = futures::join!(fetch_user, fetch_metrics_id);
+                                if let Some(info) = info {
+                                    client.telemetry.set_authenticated_user_info(
+                                        Some(info.metrics_id),
+                                        info.staff,
+                                    );
+                                } else {
+                                    client.telemetry.set_authenticated_user_info(None, false);
+                                }
+                                client.telemetry.report_event("sign in", Default::default());
                                 current_user_tx.send(user).await.ok();
                             }
                         }