1#[cfg(any(test, feature = "test-support"))]
   2pub mod test;
   3
   4mod proxy;
   5pub mod telemetry;
   6pub mod user;
   7pub mod zed_urls;
   8
   9use anyhow::{Context as _, Result, anyhow};
  10use async_tungstenite::tungstenite::{
  11    client::IntoClientRequest,
  12    error::Error as WebsocketError,
  13    http::{HeaderValue, Request, StatusCode},
  14};
  15use clock::SystemClock;
  16use cloud_api_client::CloudApiClient;
  17use cloud_api_client::websocket_protocol::MessageToClient;
  18use credentials_provider::CredentialsProvider;
  19use feature_flags::FeatureFlagAppExt as _;
  20use futures::{
  21    AsyncReadExt, FutureExt, SinkExt, Stream, StreamExt, TryFutureExt as _, TryStreamExt,
  22    channel::oneshot, future::BoxFuture,
  23};
  24use gpui::{App, AsyncApp, Entity, Global, Task, WeakEntity, actions};
  25use http_client::{HttpClient, HttpClientWithUrl, http};
  26use parking_lot::RwLock;
  27use postage::watch;
  28use proxy::connect_proxy_stream;
  29use rand::prelude::*;
  30use release_channel::{AppVersion, ReleaseChannel};
  31use rpc::proto::{AnyTypedEnvelope, EnvelopedMessage, PeerId, RequestMessage};
  32use schemars::JsonSchema;
  33use serde::{Deserialize, Serialize};
  34use settings::{Settings, SettingsContent, SettingsKey, SettingsUi};
  35use std::{
  36    any::TypeId,
  37    convert::TryFrom,
  38    fmt::Write as _,
  39    future::Future,
  40    marker::PhantomData,
  41    path::PathBuf,
  42    sync::{
  43        Arc, LazyLock, Weak,
  44        atomic::{AtomicU64, Ordering},
  45    },
  46    time::{Duration, Instant},
  47};
  48use std::{cmp, pin::Pin};
  49use telemetry::Telemetry;
  50use thiserror::Error;
  51use tokio::net::TcpStream;
  52use url::Url;
  53use util::{ConnectionResult, MergeFrom, ResultExt};
  54
  55pub use rpc::*;
  56pub use telemetry_events::Event;
  57pub use user::*;
  58
  59static ZED_SERVER_URL: LazyLock<Option<String>> =
  60    LazyLock::new(|| std::env::var("ZED_SERVER_URL").ok());
  61static ZED_RPC_URL: LazyLock<Option<String>> = LazyLock::new(|| std::env::var("ZED_RPC_URL").ok());
  62
  63pub static IMPERSONATE_LOGIN: LazyLock<Option<String>> = LazyLock::new(|| {
  64    std::env::var("ZED_IMPERSONATE")
  65        .ok()
  66        .and_then(|s| if s.is_empty() { None } else { Some(s) })
  67});
  68
  69pub static USE_WEB_LOGIN: LazyLock<bool> = LazyLock::new(|| std::env::var("ZED_WEB_LOGIN").is_ok());
  70
  71pub static ADMIN_API_TOKEN: LazyLock<Option<String>> = LazyLock::new(|| {
  72    std::env::var("ZED_ADMIN_API_TOKEN")
  73        .ok()
  74        .and_then(|s| if s.is_empty() { None } else { Some(s) })
  75});
  76
  77pub static ZED_APP_PATH: LazyLock<Option<PathBuf>> =
  78    LazyLock::new(|| std::env::var("ZED_APP_PATH").ok().map(PathBuf::from));
  79
  80pub static ZED_ALWAYS_ACTIVE: LazyLock<bool> =
  81    LazyLock::new(|| std::env::var("ZED_ALWAYS_ACTIVE").is_ok_and(|e| !e.is_empty()));
  82
  83pub const INITIAL_RECONNECTION_DELAY: Duration = Duration::from_millis(500);
  84pub const MAX_RECONNECTION_DELAY: Duration = Duration::from_secs(30);
  85pub const CONNECTION_TIMEOUT: Duration = Duration::from_secs(20);
  86
  87actions!(
  88    client,
  89    [
  90        /// Signs in to Zed account.
  91        SignIn,
  92        /// Signs out of Zed account.
  93        SignOut,
  94        /// Reconnects to the collaboration server.
  95        Reconnect
  96    ]
  97);
  98
  99#[derive(Deserialize)]
 100pub struct ClientSettings {
 101    pub server_url: String,
 102}
 103
 104impl Settings for ClientSettings {
 105    fn from_defaults(content: &settings::SettingsContent, _cx: &mut App) -> Self {
 106        Self {
 107            server_url: content.server_url.clone().unwrap(),
 108        }
 109    }
 110
 111    fn refine(&mut self, content: &settings::SettingsContent, _: &mut App) {
 112        if let Some(server_url) = content.server_url.clone() {
 113            self.server_url = server_url
 114        }
 115    }
 116
 117    fn import_from_vscode(_vscode: &settings::VsCodeSettings, _current: &mut SettingsContent) {}
 118}
 119
 120#[derive(Default, Clone, Serialize, Deserialize, JsonSchema, SettingsUi, SettingsKey)]
 121#[settings_key(None)]
 122pub struct ProxySettingsContent {
 123    proxy: Option<String>,
 124}
 125
 126#[derive(Deserialize, Default)]
 127pub struct ProxySettings {
 128    pub proxy: Option<String>,
 129}
 130
 131impl Settings for ProxySettings {
 132    fn from_defaults(content: &settings::SettingsContent, _cx: &mut App) -> Self {
 133        Self {
 134            proxy: content.proxy.clone(),
 135        }
 136    }
 137
 138    fn refine(&mut self, content: &settings::SettingsContent, _: &mut App) {
 139        if let Some(proxy) = content.proxy.clone() {
 140            self.proxy = Some(proxy)
 141        }
 142    }
 143
 144    fn import_from_vscode(vscode: &settings::VsCodeSettings, current: &mut SettingsContent) {
 145        vscode.string_setting("http.proxy", &mut current.proxy);
 146    }
 147}
 148
 149pub fn init_settings(cx: &mut App) {
 150    TelemetrySettings::register(cx);
 151    ClientSettings::register(cx);
 152    ProxySettings::register(cx);
 153}
 154
 155pub fn init(client: &Arc<Client>, cx: &mut App) {
 156    let client = Arc::downgrade(client);
 157    cx.on_action({
 158        let client = client.clone();
 159        move |_: &SignIn, cx| {
 160            if let Some(client) = client.upgrade() {
 161                cx.spawn(async move |cx| client.sign_in_with_optional_connect(true, cx).await)
 162                    .detach_and_log_err(cx);
 163            }
 164        }
 165    });
 166
 167    cx.on_action({
 168        let client = client.clone();
 169        move |_: &SignOut, cx| {
 170            if let Some(client) = client.upgrade() {
 171                cx.spawn(async move |cx| {
 172                    client.sign_out(cx).await;
 173                })
 174                .detach();
 175            }
 176        }
 177    });
 178
 179    cx.on_action({
 180        let client = client;
 181        move |_: &Reconnect, cx| {
 182            if let Some(client) = client.upgrade() {
 183                cx.spawn(async move |cx| {
 184                    client.reconnect(cx);
 185                })
 186                .detach();
 187            }
 188        }
 189    });
 190}
 191
 192pub type MessageToClientHandler = Box<dyn Fn(&MessageToClient, &mut App) + Send + Sync + 'static>;
 193
 194struct GlobalClient(Arc<Client>);
 195
 196impl Global for GlobalClient {}
 197
 198pub struct Client {
 199    id: AtomicU64,
 200    peer: Arc<Peer>,
 201    http: Arc<HttpClientWithUrl>,
 202    cloud_client: Arc<CloudApiClient>,
 203    telemetry: Arc<Telemetry>,
 204    credentials_provider: ClientCredentialsProvider,
 205    state: RwLock<ClientState>,
 206    handler_set: parking_lot::Mutex<ProtoMessageHandlerSet>,
 207    message_to_client_handlers: parking_lot::Mutex<Vec<MessageToClientHandler>>,
 208
 209    #[allow(clippy::type_complexity)]
 210    #[cfg(any(test, feature = "test-support"))]
 211    authenticate:
 212        RwLock<Option<Box<dyn 'static + Send + Sync + Fn(&AsyncApp) -> Task<Result<Credentials>>>>>,
 213
 214    #[allow(clippy::type_complexity)]
 215    #[cfg(any(test, feature = "test-support"))]
 216    establish_connection: RwLock<
 217        Option<
 218            Box<
 219                dyn 'static
 220                    + Send
 221                    + Sync
 222                    + Fn(
 223                        &Credentials,
 224                        &AsyncApp,
 225                    ) -> Task<Result<Connection, EstablishConnectionError>>,
 226            >,
 227        >,
 228    >,
 229
 230    #[cfg(any(test, feature = "test-support"))]
 231    rpc_url: RwLock<Option<Url>>,
 232}
 233
 234#[derive(Error, Debug)]
 235pub enum EstablishConnectionError {
 236    #[error("upgrade required")]
 237    UpgradeRequired,
 238    #[error("unauthorized")]
 239    Unauthorized,
 240    #[error("{0}")]
 241    Other(#[from] anyhow::Error),
 242    #[error("{0}")]
 243    InvalidHeaderValue(#[from] async_tungstenite::tungstenite::http::header::InvalidHeaderValue),
 244    #[error("{0}")]
 245    Io(#[from] std::io::Error),
 246    #[error("{0}")]
 247    Websocket(#[from] async_tungstenite::tungstenite::http::Error),
 248}
 249
 250impl From<WebsocketError> for EstablishConnectionError {
 251    fn from(error: WebsocketError) -> Self {
 252        if let WebsocketError::Http(response) = &error {
 253            match response.status() {
 254                StatusCode::UNAUTHORIZED => return EstablishConnectionError::Unauthorized,
 255                StatusCode::UPGRADE_REQUIRED => return EstablishConnectionError::UpgradeRequired,
 256                _ => {}
 257            }
 258        }
 259        EstablishConnectionError::Other(error.into())
 260    }
 261}
 262
 263impl EstablishConnectionError {
 264    pub fn other(error: impl Into<anyhow::Error> + Send + Sync) -> Self {
 265        Self::Other(error.into())
 266    }
 267}
 268
 269#[derive(Copy, Clone, Debug, PartialEq)]
 270pub enum Status {
 271    SignedOut,
 272    UpgradeRequired,
 273    Authenticating,
 274    Authenticated,
 275    AuthenticationError,
 276    Connecting,
 277    ConnectionError,
 278    Connected {
 279        peer_id: PeerId,
 280        connection_id: ConnectionId,
 281    },
 282    ConnectionLost,
 283    Reauthenticating,
 284    Reauthenticated,
 285    Reconnecting,
 286    ReconnectionError {
 287        next_reconnection: Instant,
 288    },
 289}
 290
 291impl Status {
 292    pub fn is_connected(&self) -> bool {
 293        matches!(self, Self::Connected { .. })
 294    }
 295
 296    pub fn was_connected(&self) -> bool {
 297        matches!(
 298            self,
 299            Self::ConnectionLost
 300                | Self::Reauthenticating
 301                | Self::Reauthenticated
 302                | Self::Reconnecting
 303        )
 304    }
 305
 306    /// Returns whether the client is currently connected or was connected at some point.
 307    pub fn is_or_was_connected(&self) -> bool {
 308        self.is_connected() || self.was_connected()
 309    }
 310
 311    pub fn is_signing_in(&self) -> bool {
 312        matches!(
 313            self,
 314            Self::Authenticating | Self::Reauthenticating | Self::Connecting | Self::Reconnecting
 315        )
 316    }
 317
 318    pub fn is_signed_out(&self) -> bool {
 319        matches!(self, Self::SignedOut | Self::UpgradeRequired)
 320    }
 321}
 322
 323struct ClientState {
 324    credentials: Option<Credentials>,
 325    status: (watch::Sender<Status>, watch::Receiver<Status>),
 326    _reconnect_task: Option<Task<()>>,
 327}
 328
 329#[derive(Clone, Debug, Eq, PartialEq)]
 330pub struct Credentials {
 331    pub user_id: u64,
 332    pub access_token: String,
 333}
 334
 335impl Credentials {
 336    pub fn authorization_header(&self) -> String {
 337        format!("{} {}", self.user_id, self.access_token)
 338    }
 339}
 340
 341pub struct ClientCredentialsProvider {
 342    provider: Arc<dyn CredentialsProvider>,
 343}
 344
 345impl ClientCredentialsProvider {
 346    pub fn new(cx: &App) -> Self {
 347        Self {
 348            provider: <dyn CredentialsProvider>::global(cx),
 349        }
 350    }
 351
 352    fn server_url(&self, cx: &AsyncApp) -> Result<String> {
 353        cx.update(|cx| ClientSettings::get_global(cx).server_url.clone())
 354    }
 355
 356    /// Reads the credentials from the provider.
 357    fn read_credentials<'a>(
 358        &'a self,
 359        cx: &'a AsyncApp,
 360    ) -> Pin<Box<dyn Future<Output = Option<Credentials>> + 'a>> {
 361        async move {
 362            if IMPERSONATE_LOGIN.is_some() {
 363                return None;
 364            }
 365
 366            let server_url = self.server_url(cx).ok()?;
 367            let (user_id, access_token) = self
 368                .provider
 369                .read_credentials(&server_url, cx)
 370                .await
 371                .log_err()
 372                .flatten()?;
 373
 374            Some(Credentials {
 375                user_id: user_id.parse().ok()?,
 376                access_token: String::from_utf8(access_token).ok()?,
 377            })
 378        }
 379        .boxed_local()
 380    }
 381
 382    /// Writes the credentials to the provider.
 383    fn write_credentials<'a>(
 384        &'a self,
 385        user_id: u64,
 386        access_token: String,
 387        cx: &'a AsyncApp,
 388    ) -> Pin<Box<dyn Future<Output = Result<()>> + 'a>> {
 389        async move {
 390            let server_url = self.server_url(cx)?;
 391            self.provider
 392                .write_credentials(
 393                    &server_url,
 394                    &user_id.to_string(),
 395                    access_token.as_bytes(),
 396                    cx,
 397                )
 398                .await
 399        }
 400        .boxed_local()
 401    }
 402
 403    /// Deletes the credentials from the provider.
 404    fn delete_credentials<'a>(
 405        &'a self,
 406        cx: &'a AsyncApp,
 407    ) -> Pin<Box<dyn Future<Output = Result<()>> + 'a>> {
 408        async move {
 409            let server_url = self.server_url(cx)?;
 410            self.provider.delete_credentials(&server_url, cx).await
 411        }
 412        .boxed_local()
 413    }
 414}
 415
 416impl Default for ClientState {
 417    fn default() -> Self {
 418        Self {
 419            credentials: None,
 420            status: watch::channel_with(Status::SignedOut),
 421            _reconnect_task: None,
 422        }
 423    }
 424}
 425
 426pub enum Subscription {
 427    Entity {
 428        client: Weak<Client>,
 429        id: (TypeId, u64),
 430    },
 431    Message {
 432        client: Weak<Client>,
 433        id: TypeId,
 434    },
 435}
 436
 437impl Drop for Subscription {
 438    fn drop(&mut self) {
 439        match self {
 440            Subscription::Entity { client, id } => {
 441                if let Some(client) = client.upgrade() {
 442                    let mut state = client.handler_set.lock();
 443                    let _ = state.entities_by_type_and_remote_id.remove(id);
 444                }
 445            }
 446            Subscription::Message { client, id } => {
 447                if let Some(client) = client.upgrade() {
 448                    let mut state = client.handler_set.lock();
 449                    let _ = state.entity_types_by_message_type.remove(id);
 450                    let _ = state.message_handlers.remove(id);
 451                }
 452            }
 453        }
 454    }
 455}
 456
 457pub struct PendingEntitySubscription<T: 'static> {
 458    client: Arc<Client>,
 459    remote_id: u64,
 460    _entity_type: PhantomData<T>,
 461    consumed: bool,
 462}
 463
 464impl<T: 'static> PendingEntitySubscription<T> {
 465    pub fn set_entity(mut self, entity: &Entity<T>, cx: &AsyncApp) -> Subscription {
 466        self.consumed = true;
 467        let mut handlers = self.client.handler_set.lock();
 468        let id = (TypeId::of::<T>(), self.remote_id);
 469        let Some(EntityMessageSubscriber::Pending(messages)) =
 470            handlers.entities_by_type_and_remote_id.remove(&id)
 471        else {
 472            unreachable!()
 473        };
 474
 475        handlers.entities_by_type_and_remote_id.insert(
 476            id,
 477            EntityMessageSubscriber::Entity {
 478                handle: entity.downgrade().into(),
 479            },
 480        );
 481        drop(handlers);
 482        for message in messages {
 483            let client_id = self.client.id();
 484            let type_name = message.payload_type_name();
 485            let sender_id = message.original_sender_id();
 486            log::debug!(
 487                "handling queued rpc message. client_id:{}, sender_id:{:?}, type:{}",
 488                client_id,
 489                sender_id,
 490                type_name
 491            );
 492            self.client.handle_message(message, cx);
 493        }
 494        Subscription::Entity {
 495            client: Arc::downgrade(&self.client),
 496            id,
 497        }
 498    }
 499}
 500
 501impl<T: 'static> Drop for PendingEntitySubscription<T> {
 502    fn drop(&mut self) {
 503        if !self.consumed {
 504            let mut state = self.client.handler_set.lock();
 505            if let Some(EntityMessageSubscriber::Pending(messages)) = state
 506                .entities_by_type_and_remote_id
 507                .remove(&(TypeId::of::<T>(), self.remote_id))
 508            {
 509                for message in messages {
 510                    log::info!("unhandled message {}", message.payload_type_name());
 511                }
 512            }
 513        }
 514    }
 515}
 516
 517#[derive(Copy, Clone, Deserialize, Debug)]
 518pub struct TelemetrySettings {
 519    pub diagnostics: bool,
 520    pub metrics: bool,
 521}
 522
 523impl settings::Settings for TelemetrySettings {
 524    fn from_defaults(content: &SettingsContent, _cx: &mut App) -> Self {
 525        Self {
 526            diagnostics: content.telemetry.as_ref().unwrap().diagnostics.unwrap(),
 527            metrics: content.telemetry.as_ref().unwrap().metrics.unwrap(),
 528        }
 529    }
 530
 531    fn refine(&mut self, content: &SettingsContent, cx: &mut App) {
 532        let Some(telemetry) = &content.telemetry else {
 533            return;
 534        };
 535        self.diagnostics.merge_from(&telemetry.diagnostics);
 536        self.metrics.merge_from(&telemetry.metrics);
 537    }
 538
 539    fn import_from_vscode(vscode: &settings::VsCodeSettings, current: &mut SettingsContent) {
 540        let mut telemetry = settings::TelemetrySettingsContent::default();
 541        vscode.enum_setting("telemetry.telemetryLevel", &mut telemetry.metrics, |s| {
 542            Some(s == "all")
 543        });
 544        vscode.enum_setting(
 545            "telemetry.telemetryLevel",
 546            &mut telemetry.diagnostics,
 547            |s| Some(matches!(s, "all" | "error" | "crash")),
 548        );
 549        // we could translate telemetry.telemetryLevel, but just because users didn't want
 550        // to send microsoft telemetry doesn't mean they don't want to send it to zed. their
 551        // all/error/crash/off correspond to combinations of our "diagnostics" and "metrics".
 552        if let Some(diagnostics) = telemetry.diagnostics {
 553            current.telemetry.get_or_insert_default().diagnostics = Some(diagnostics)
 554        }
 555        if let Some(metrics) = telemetry.metrics {
 556            current.telemetry.get_or_insert_default().metrics = Some(metrics)
 557        }
 558    }
 559}
 560
 561impl Client {
 562    pub fn new(
 563        clock: Arc<dyn SystemClock>,
 564        http: Arc<HttpClientWithUrl>,
 565        cx: &mut App,
 566    ) -> Arc<Self> {
 567        Arc::new(Self {
 568            id: AtomicU64::new(0),
 569            peer: Peer::new(0),
 570            telemetry: Telemetry::new(clock, http.clone(), cx),
 571            cloud_client: Arc::new(CloudApiClient::new(http.clone())),
 572            http,
 573            credentials_provider: ClientCredentialsProvider::new(cx),
 574            state: Default::default(),
 575            handler_set: Default::default(),
 576            message_to_client_handlers: parking_lot::Mutex::new(Vec::new()),
 577
 578            #[cfg(any(test, feature = "test-support"))]
 579            authenticate: Default::default(),
 580            #[cfg(any(test, feature = "test-support"))]
 581            establish_connection: Default::default(),
 582            #[cfg(any(test, feature = "test-support"))]
 583            rpc_url: RwLock::default(),
 584        })
 585    }
 586
 587    pub fn production(cx: &mut App) -> Arc<Self> {
 588        let clock = Arc::new(clock::RealSystemClock);
 589        let http = Arc::new(HttpClientWithUrl::new_url(
 590            cx.http_client(),
 591            &ClientSettings::get_global(cx).server_url,
 592            cx.http_client().proxy().cloned(),
 593        ));
 594        Self::new(clock, http, cx)
 595    }
 596
 597    pub fn id(&self) -> u64 {
 598        self.id.load(Ordering::SeqCst)
 599    }
 600
 601    pub fn http_client(&self) -> Arc<HttpClientWithUrl> {
 602        self.http.clone()
 603    }
 604
 605    pub fn cloud_client(&self) -> Arc<CloudApiClient> {
 606        self.cloud_client.clone()
 607    }
 608
 609    pub fn set_id(&self, id: u64) -> &Self {
 610        self.id.store(id, Ordering::SeqCst);
 611        self
 612    }
 613
 614    #[cfg(any(test, feature = "test-support"))]
 615    pub fn teardown(&self) {
 616        let mut state = self.state.write();
 617        state._reconnect_task.take();
 618        self.handler_set.lock().clear();
 619        self.peer.teardown();
 620    }
 621
 622    #[cfg(any(test, feature = "test-support"))]
 623    pub fn override_authenticate<F>(&self, authenticate: F) -> &Self
 624    where
 625        F: 'static + Send + Sync + Fn(&AsyncApp) -> Task<Result<Credentials>>,
 626    {
 627        *self.authenticate.write() = Some(Box::new(authenticate));
 628        self
 629    }
 630
 631    #[cfg(any(test, feature = "test-support"))]
 632    pub fn override_establish_connection<F>(&self, connect: F) -> &Self
 633    where
 634        F: 'static
 635            + Send
 636            + Sync
 637            + Fn(&Credentials, &AsyncApp) -> Task<Result<Connection, EstablishConnectionError>>,
 638    {
 639        *self.establish_connection.write() = Some(Box::new(connect));
 640        self
 641    }
 642
 643    #[cfg(any(test, feature = "test-support"))]
 644    pub fn override_rpc_url(&self, url: Url) -> &Self {
 645        *self.rpc_url.write() = Some(url);
 646        self
 647    }
 648
 649    pub fn global(cx: &App) -> Arc<Self> {
 650        cx.global::<GlobalClient>().0.clone()
 651    }
 652    pub fn set_global(client: Arc<Client>, cx: &mut App) {
 653        cx.set_global(GlobalClient(client))
 654    }
 655
 656    pub fn user_id(&self) -> Option<u64> {
 657        self.state
 658            .read()
 659            .credentials
 660            .as_ref()
 661            .map(|credentials| credentials.user_id)
 662    }
 663
 664    pub fn peer_id(&self) -> Option<PeerId> {
 665        if let Status::Connected { peer_id, .. } = &*self.status().borrow() {
 666            Some(*peer_id)
 667        } else {
 668            None
 669        }
 670    }
 671
 672    pub fn status(&self) -> watch::Receiver<Status> {
 673        self.state.read().status.1.clone()
 674    }
 675
 676    fn set_status(self: &Arc<Self>, status: Status, cx: &AsyncApp) {
 677        log::info!("set status on client {}: {:?}", self.id(), status);
 678        let mut state = self.state.write();
 679        *state.status.0.borrow_mut() = status;
 680
 681        match status {
 682            Status::Connected { .. } => {
 683                state._reconnect_task = None;
 684            }
 685            Status::ConnectionLost => {
 686                let client = self.clone();
 687                state._reconnect_task = Some(cx.spawn(async move |cx| {
 688                    #[cfg(any(test, feature = "test-support"))]
 689                    let mut rng = StdRng::seed_from_u64(0);
 690                    #[cfg(not(any(test, feature = "test-support")))]
 691                    let mut rng = StdRng::from_os_rng();
 692
 693                    let mut delay = INITIAL_RECONNECTION_DELAY;
 694                    loop {
 695                        match client.connect(true, cx).await {
 696                            ConnectionResult::Timeout => {
 697                                log::error!("client connect attempt timed out")
 698                            }
 699                            ConnectionResult::ConnectionReset => {
 700                                log::error!("client connect attempt reset")
 701                            }
 702                            ConnectionResult::Result(r) => {
 703                                if let Err(error) = r {
 704                                    log::error!("failed to connect: {error}");
 705                                } else {
 706                                    break;
 707                                }
 708                            }
 709                        }
 710
 711                        if matches!(
 712                            *client.status().borrow(),
 713                            Status::AuthenticationError | Status::ConnectionError
 714                        ) {
 715                            client.set_status(
 716                                Status::ReconnectionError {
 717                                    next_reconnection: Instant::now() + delay,
 718                                },
 719                                cx,
 720                            );
 721                            let jitter = Duration::from_millis(
 722                                rng.random_range(0..delay.as_millis() as u64),
 723                            );
 724                            cx.background_executor().timer(delay + jitter).await;
 725                            delay = cmp::min(delay * 2, MAX_RECONNECTION_DELAY);
 726                        } else {
 727                            break;
 728                        }
 729                    }
 730                }));
 731            }
 732            Status::SignedOut | Status::UpgradeRequired => {
 733                self.telemetry.set_authenticated_user_info(None, false);
 734                state._reconnect_task.take();
 735            }
 736            _ => {}
 737        }
 738    }
 739
 740    pub fn subscribe_to_entity<T>(
 741        self: &Arc<Self>,
 742        remote_id: u64,
 743    ) -> Result<PendingEntitySubscription<T>>
 744    where
 745        T: 'static,
 746    {
 747        let id = (TypeId::of::<T>(), remote_id);
 748
 749        let mut state = self.handler_set.lock();
 750        anyhow::ensure!(
 751            !state.entities_by_type_and_remote_id.contains_key(&id),
 752            "already subscribed to entity"
 753        );
 754
 755        state
 756            .entities_by_type_and_remote_id
 757            .insert(id, EntityMessageSubscriber::Pending(Default::default()));
 758
 759        Ok(PendingEntitySubscription {
 760            client: self.clone(),
 761            remote_id,
 762            consumed: false,
 763            _entity_type: PhantomData,
 764        })
 765    }
 766
 767    #[track_caller]
 768    pub fn add_message_handler<M, E, H, F>(
 769        self: &Arc<Self>,
 770        entity: WeakEntity<E>,
 771        handler: H,
 772    ) -> Subscription
 773    where
 774        M: EnvelopedMessage,
 775        E: 'static,
 776        H: 'static + Sync + Fn(Entity<E>, TypedEnvelope<M>, AsyncApp) -> F + Send + Sync,
 777        F: 'static + Future<Output = Result<()>>,
 778    {
 779        self.add_message_handler_impl(entity, move |entity, message, _, cx| {
 780            handler(entity, message, cx)
 781        })
 782    }
 783
 784    fn add_message_handler_impl<M, E, H, F>(
 785        self: &Arc<Self>,
 786        entity: WeakEntity<E>,
 787        handler: H,
 788    ) -> Subscription
 789    where
 790        M: EnvelopedMessage,
 791        E: 'static,
 792        H: 'static
 793            + Sync
 794            + Fn(Entity<E>, TypedEnvelope<M>, AnyProtoClient, AsyncApp) -> F
 795            + Send
 796            + Sync,
 797        F: 'static + Future<Output = Result<()>>,
 798    {
 799        let message_type_id = TypeId::of::<M>();
 800        let mut state = self.handler_set.lock();
 801        state
 802            .entities_by_message_type
 803            .insert(message_type_id, entity.into());
 804
 805        let prev_handler = state.message_handlers.insert(
 806            message_type_id,
 807            Arc::new(move |subscriber, envelope, client, cx| {
 808                let subscriber = subscriber.downcast::<E>().unwrap();
 809                let envelope = envelope.into_any().downcast::<TypedEnvelope<M>>().unwrap();
 810                handler(subscriber, *envelope, client, cx).boxed_local()
 811            }),
 812        );
 813        if prev_handler.is_some() {
 814            let location = std::panic::Location::caller();
 815            panic!(
 816                "{}:{} registered handler for the same message {} twice",
 817                location.file(),
 818                location.line(),
 819                std::any::type_name::<M>()
 820            );
 821        }
 822
 823        Subscription::Message {
 824            client: Arc::downgrade(self),
 825            id: message_type_id,
 826        }
 827    }
 828
 829    pub fn add_request_handler<M, E, H, F>(
 830        self: &Arc<Self>,
 831        entity: WeakEntity<E>,
 832        handler: H,
 833    ) -> Subscription
 834    where
 835        M: RequestMessage,
 836        E: 'static,
 837        H: 'static + Sync + Fn(Entity<E>, TypedEnvelope<M>, AsyncApp) -> F + Send + Sync,
 838        F: 'static + Future<Output = Result<M::Response>>,
 839    {
 840        self.add_message_handler_impl(entity, move |handle, envelope, this, cx| {
 841            Self::respond_to_request(envelope.receipt(), handler(handle, envelope, cx), this)
 842        })
 843    }
 844
 845    async fn respond_to_request<T: RequestMessage, F: Future<Output = Result<T::Response>>>(
 846        receipt: Receipt<T>,
 847        response: F,
 848        client: AnyProtoClient,
 849    ) -> Result<()> {
 850        match response.await {
 851            Ok(response) => {
 852                client.send_response(receipt.message_id, response)?;
 853                Ok(())
 854            }
 855            Err(error) => {
 856                client.send_response(receipt.message_id, error.to_proto())?;
 857                Err(error)
 858            }
 859        }
 860    }
 861
 862    pub async fn has_credentials(&self, cx: &AsyncApp) -> bool {
 863        self.credentials_provider
 864            .read_credentials(cx)
 865            .await
 866            .is_some()
 867    }
 868
 869    pub async fn sign_in(
 870        self: &Arc<Self>,
 871        try_provider: bool,
 872        cx: &AsyncApp,
 873    ) -> Result<Credentials> {
 874        let is_reauthenticating = if self.status().borrow().is_signed_out() {
 875            self.set_status(Status::Authenticating, cx);
 876            false
 877        } else {
 878            self.set_status(Status::Reauthenticating, cx);
 879            true
 880        };
 881
 882        let mut credentials = None;
 883
 884        let old_credentials = self.state.read().credentials.clone();
 885        if let Some(old_credentials) = old_credentials
 886            && self.validate_credentials(&old_credentials, cx).await?
 887        {
 888            credentials = Some(old_credentials);
 889        }
 890
 891        if credentials.is_none()
 892            && try_provider
 893            && let Some(stored_credentials) = self.credentials_provider.read_credentials(cx).await
 894        {
 895            if self.validate_credentials(&stored_credentials, cx).await? {
 896                credentials = Some(stored_credentials);
 897            } else {
 898                self.credentials_provider
 899                    .delete_credentials(cx)
 900                    .await
 901                    .log_err();
 902            }
 903        }
 904
 905        if credentials.is_none() {
 906            let mut status_rx = self.status();
 907            let _ = status_rx.next().await;
 908            futures::select_biased! {
 909                authenticate = self.authenticate(cx).fuse() => {
 910                    match authenticate {
 911                        Ok(creds) => {
 912                            if IMPERSONATE_LOGIN.is_none() {
 913                                self.credentials_provider
 914                                    .write_credentials(creds.user_id, creds.access_token.clone(), cx)
 915                                    .await
 916                                    .log_err();
 917                            }
 918
 919                            credentials = Some(creds);
 920                        },
 921                        Err(err) => {
 922                            self.set_status(Status::AuthenticationError, cx);
 923                            return Err(err);
 924                        }
 925                    }
 926                }
 927                _ = status_rx.next().fuse() => {
 928                    return Err(anyhow!("authentication canceled"));
 929                }
 930            }
 931        }
 932
 933        let credentials = credentials.unwrap();
 934        self.set_id(credentials.user_id);
 935        self.cloud_client
 936            .set_credentials(credentials.user_id as u32, credentials.access_token.clone());
 937        self.state.write().credentials = Some(credentials.clone());
 938        self.set_status(
 939            if is_reauthenticating {
 940                Status::Reauthenticated
 941            } else {
 942                Status::Authenticated
 943            },
 944            cx,
 945        );
 946
 947        Ok(credentials)
 948    }
 949
 950    async fn validate_credentials(
 951        self: &Arc<Self>,
 952        credentials: &Credentials,
 953        cx: &AsyncApp,
 954    ) -> Result<bool> {
 955        match self
 956            .cloud_client
 957            .validate_credentials(credentials.user_id as u32, &credentials.access_token)
 958            .await
 959        {
 960            Ok(valid) => Ok(valid),
 961            Err(err) => {
 962                self.set_status(Status::AuthenticationError, cx);
 963                Err(anyhow!("failed to validate credentials: {}", err))
 964            }
 965        }
 966    }
 967
 968    /// Establishes a WebSocket connection with Cloud for receiving updates from the server.
 969    async fn connect_to_cloud(self: &Arc<Self>, cx: &AsyncApp) -> Result<()> {
 970        let connect_task = cx.update({
 971            let cloud_client = self.cloud_client.clone();
 972            move |cx| cloud_client.connect(cx)
 973        })??;
 974        let connection = connect_task.await?;
 975
 976        let (mut messages, task) = cx.update(|cx| connection.spawn(cx))?;
 977        task.detach();
 978
 979        cx.spawn({
 980            let this = self.clone();
 981            async move |cx| {
 982                while let Some(message) = messages.next().await {
 983                    if let Some(message) = message.log_err() {
 984                        this.handle_message_to_client(message, cx);
 985                    }
 986                }
 987            }
 988        })
 989        .detach();
 990
 991        Ok(())
 992    }
 993
 994    /// Performs a sign-in and also (optionally) connects to Collab.
 995    ///
 996    /// Only Zed staff automatically connect to Collab.
 997    pub async fn sign_in_with_optional_connect(
 998        self: &Arc<Self>,
 999        try_provider: bool,
1000        cx: &AsyncApp,
1001    ) -> Result<()> {
1002        // Don't try to sign in again if we're already connected to Collab, as it will temporarily disconnect us.
1003        if self.status().borrow().is_connected() {
1004            return Ok(());
1005        }
1006
1007        let (is_staff_tx, is_staff_rx) = oneshot::channel::<bool>();
1008        let mut is_staff_tx = Some(is_staff_tx);
1009        cx.update(|cx| {
1010            cx.on_flags_ready(move |state, _cx| {
1011                if let Some(is_staff_tx) = is_staff_tx.take() {
1012                    is_staff_tx.send(state.is_staff).log_err();
1013                }
1014            })
1015            .detach();
1016        })
1017        .log_err();
1018
1019        let credentials = self.sign_in(try_provider, cx).await?;
1020
1021        self.connect_to_cloud(cx).await.log_err();
1022
1023        cx.update(move |cx| {
1024            cx.spawn({
1025                let client = self.clone();
1026                async move |cx| {
1027                    let is_staff = is_staff_rx.await?;
1028                    if is_staff {
1029                        match client.connect_with_credentials(credentials, cx).await {
1030                            ConnectionResult::Timeout => Err(anyhow!("connection timed out")),
1031                            ConnectionResult::ConnectionReset => Err(anyhow!("connection reset")),
1032                            ConnectionResult::Result(result) => {
1033                                result.context("client auth and connect")
1034                            }
1035                        }
1036                    } else {
1037                        Ok(())
1038                    }
1039                }
1040            })
1041            .detach_and_log_err(cx);
1042        })
1043        .log_err();
1044
1045        Ok(())
1046    }
1047
1048    pub async fn connect(
1049        self: &Arc<Self>,
1050        try_provider: bool,
1051        cx: &AsyncApp,
1052    ) -> ConnectionResult<()> {
1053        let was_disconnected = match *self.status().borrow() {
1054            Status::SignedOut | Status::Authenticated => true,
1055            Status::ConnectionError
1056            | Status::ConnectionLost
1057            | Status::Authenticating
1058            | Status::AuthenticationError
1059            | Status::Reauthenticating
1060            | Status::Reauthenticated
1061            | Status::ReconnectionError { .. } => false,
1062            Status::Connected { .. } | Status::Connecting | Status::Reconnecting => {
1063                return ConnectionResult::Result(Ok(()));
1064            }
1065            Status::UpgradeRequired => {
1066                return ConnectionResult::Result(
1067                    Err(EstablishConnectionError::UpgradeRequired)
1068                        .context("client auth and connect"),
1069                );
1070            }
1071        };
1072        let credentials = match self.sign_in(try_provider, cx).await {
1073            Ok(credentials) => credentials,
1074            Err(err) => return ConnectionResult::Result(Err(err)),
1075        };
1076
1077        if was_disconnected {
1078            self.set_status(Status::Connecting, cx);
1079        } else {
1080            self.set_status(Status::Reconnecting, cx);
1081        }
1082
1083        self.connect_with_credentials(credentials, cx).await
1084    }
1085
1086    async fn connect_with_credentials(
1087        self: &Arc<Self>,
1088        credentials: Credentials,
1089        cx: &AsyncApp,
1090    ) -> ConnectionResult<()> {
1091        let mut timeout =
1092            futures::FutureExt::fuse(cx.background_executor().timer(CONNECTION_TIMEOUT));
1093        futures::select_biased! {
1094            connection = self.establish_connection(&credentials, cx).fuse() => {
1095                match connection {
1096                    Ok(conn) => {
1097                        futures::select_biased! {
1098                            result = self.set_connection(conn, cx).fuse() => {
1099                                match result.context("client auth and connect") {
1100                                    Ok(()) => ConnectionResult::Result(Ok(())),
1101                                    Err(err) => {
1102                                        self.set_status(Status::ConnectionError, cx);
1103                                        ConnectionResult::Result(Err(err))
1104                                    },
1105                                }
1106                            },
1107                            _ = timeout => {
1108                                self.set_status(Status::ConnectionError, cx);
1109                                ConnectionResult::Timeout
1110                            }
1111                        }
1112                    }
1113                    Err(EstablishConnectionError::Unauthorized) => {
1114                        self.set_status(Status::ConnectionError, cx);
1115                        ConnectionResult::Result(Err(EstablishConnectionError::Unauthorized).context("client auth and connect"))
1116                    }
1117                    Err(EstablishConnectionError::UpgradeRequired) => {
1118                        self.set_status(Status::UpgradeRequired, cx);
1119                        ConnectionResult::Result(Err(EstablishConnectionError::UpgradeRequired).context("client auth and connect"))
1120                    }
1121                    Err(error) => {
1122                        self.set_status(Status::ConnectionError, cx);
1123                        ConnectionResult::Result(Err(error).context("client auth and connect"))
1124                    }
1125                }
1126            }
1127            _ = &mut timeout => {
1128                self.set_status(Status::ConnectionError, cx);
1129                ConnectionResult::Timeout
1130            }
1131        }
1132    }
1133
1134    async fn set_connection(self: &Arc<Self>, conn: Connection, cx: &AsyncApp) -> Result<()> {
1135        let executor = cx.background_executor();
1136        log::debug!("add connection to peer");
1137        let (connection_id, handle_io, mut incoming) = self.peer.add_connection(conn, {
1138            let executor = executor.clone();
1139            move |duration| executor.timer(duration)
1140        });
1141        let handle_io = executor.spawn(handle_io);
1142
1143        let peer_id = async {
1144            log::debug!("waiting for server hello");
1145            let message = incoming.next().await.context("no hello message received")?;
1146            log::debug!("got server hello");
1147            let hello_message_type_name = message.payload_type_name().to_string();
1148            let hello = message
1149                .into_any()
1150                .downcast::<TypedEnvelope<proto::Hello>>()
1151                .map_err(|_| {
1152                    anyhow!(
1153                        "invalid hello message received: {:?}",
1154                        hello_message_type_name
1155                    )
1156                })?;
1157            let peer_id = hello.payload.peer_id.context("invalid peer id")?;
1158            Ok(peer_id)
1159        };
1160
1161        let peer_id = match peer_id.await {
1162            Ok(peer_id) => peer_id,
1163            Err(error) => {
1164                self.peer.disconnect(connection_id);
1165                return Err(error);
1166            }
1167        };
1168
1169        log::debug!(
1170            "set status to connected (connection id: {:?}, peer id: {:?})",
1171            connection_id,
1172            peer_id
1173        );
1174        self.set_status(
1175            Status::Connected {
1176                peer_id,
1177                connection_id,
1178            },
1179            cx,
1180        );
1181
1182        cx.spawn({
1183            let this = self.clone();
1184            async move |cx| {
1185                while let Some(message) = incoming.next().await {
1186                    this.handle_message(message, cx);
1187                    // Don't starve the main thread when receiving lots of messages at once.
1188                    smol::future::yield_now().await;
1189                }
1190            }
1191        })
1192        .detach();
1193
1194        cx.spawn({
1195            let this = self.clone();
1196            async move |cx| match handle_io.await {
1197                Ok(()) => {
1198                    if *this.status().borrow()
1199                        == (Status::Connected {
1200                            connection_id,
1201                            peer_id,
1202                        })
1203                    {
1204                        this.set_status(Status::SignedOut, cx);
1205                    }
1206                }
1207                Err(err) => {
1208                    log::error!("connection error: {:?}", err);
1209                    this.set_status(Status::ConnectionLost, cx);
1210                }
1211            }
1212        })
1213        .detach();
1214
1215        Ok(())
1216    }
1217
1218    fn authenticate(self: &Arc<Self>, cx: &AsyncApp) -> Task<Result<Credentials>> {
1219        #[cfg(any(test, feature = "test-support"))]
1220        if let Some(callback) = self.authenticate.read().as_ref() {
1221            return callback(cx);
1222        }
1223
1224        self.authenticate_with_browser(cx)
1225    }
1226
1227    fn establish_connection(
1228        self: &Arc<Self>,
1229        credentials: &Credentials,
1230        cx: &AsyncApp,
1231    ) -> Task<Result<Connection, EstablishConnectionError>> {
1232        #[cfg(any(test, feature = "test-support"))]
1233        if let Some(callback) = self.establish_connection.read().as_ref() {
1234            return callback(credentials, cx);
1235        }
1236
1237        self.establish_websocket_connection(credentials, cx)
1238    }
1239
1240    fn rpc_url(
1241        &self,
1242        http: Arc<HttpClientWithUrl>,
1243        release_channel: Option<ReleaseChannel>,
1244    ) -> impl Future<Output = Result<url::Url>> + use<> {
1245        #[cfg(any(test, feature = "test-support"))]
1246        let url_override = self.rpc_url.read().clone();
1247
1248        async move {
1249            #[cfg(any(test, feature = "test-support"))]
1250            if let Some(url) = url_override {
1251                return Ok(url);
1252            }
1253
1254            if let Some(url) = &*ZED_RPC_URL {
1255                return Url::parse(url).context("invalid rpc url");
1256            }
1257
1258            let mut url = http.build_url("/rpc");
1259            if let Some(preview_param) =
1260                release_channel.and_then(|channel| channel.release_query_param())
1261            {
1262                url += "?";
1263                url += preview_param;
1264            }
1265
1266            let response = http.get(&url, Default::default(), false).await?;
1267            anyhow::ensure!(
1268                response.status().is_redirection(),
1269                "unexpected /rpc response status {}",
1270                response.status()
1271            );
1272            let collab_url = response
1273                .headers()
1274                .get("Location")
1275                .context("missing location header in /rpc response")?
1276                .to_str()
1277                .map_err(EstablishConnectionError::other)?
1278                .to_string();
1279            Url::parse(&collab_url).with_context(|| format!("parsing collab rpc url {collab_url}"))
1280        }
1281    }
1282
1283    fn establish_websocket_connection(
1284        self: &Arc<Self>,
1285        credentials: &Credentials,
1286        cx: &AsyncApp,
1287    ) -> Task<Result<Connection, EstablishConnectionError>> {
1288        let release_channel = cx
1289            .update(|cx| ReleaseChannel::try_global(cx))
1290            .ok()
1291            .flatten();
1292        let app_version = cx
1293            .update(|cx| AppVersion::global(cx).to_string())
1294            .ok()
1295            .unwrap_or_default();
1296
1297        let http = self.http.clone();
1298        let proxy = http.proxy().cloned();
1299        let user_agent = http.user_agent().cloned();
1300        let credentials = credentials.clone();
1301        let rpc_url = self.rpc_url(http, release_channel);
1302        let system_id = self.telemetry.system_id();
1303        let metrics_id = self.telemetry.metrics_id();
1304        cx.spawn(async move |cx| {
1305            use HttpOrHttps::*;
1306
1307            #[derive(Debug)]
1308            enum HttpOrHttps {
1309                Http,
1310                Https,
1311            }
1312
1313            let mut rpc_url = rpc_url.await?;
1314            let url_scheme = match rpc_url.scheme() {
1315                "https" => Https,
1316                "http" => Http,
1317                _ => Err(anyhow!("invalid rpc url: {}", rpc_url))?,
1318            };
1319
1320            let stream = gpui_tokio::Tokio::spawn_result(cx, {
1321                let rpc_url = rpc_url.clone();
1322                async move {
1323                    let rpc_host = rpc_url
1324                        .host_str()
1325                        .zip(rpc_url.port_or_known_default())
1326                        .context("missing host in rpc url")?;
1327                    Ok(match proxy {
1328                        Some(proxy) => connect_proxy_stream(&proxy, rpc_host).await?,
1329                        None => Box::new(TcpStream::connect(rpc_host).await?),
1330                    })
1331                }
1332            })?
1333            .await?;
1334
1335            log::info!("connected to rpc endpoint {}", rpc_url);
1336
1337            rpc_url
1338                .set_scheme(match url_scheme {
1339                    Https => "wss",
1340                    Http => "ws",
1341                })
1342                .unwrap();
1343
1344            // We call `into_client_request` to let `tungstenite` construct the WebSocket request
1345            // for us from the RPC URL.
1346            //
1347            // Among other things, it will generate and set a `Sec-WebSocket-Key` header for us.
1348            let mut request = IntoClientRequest::into_client_request(rpc_url.as_str())?;
1349
1350            // We then modify the request to add our desired headers.
1351            let request_headers = request.headers_mut();
1352            request_headers.insert(
1353                http::header::AUTHORIZATION,
1354                HeaderValue::from_str(&credentials.authorization_header())?,
1355            );
1356            request_headers.insert(
1357                "x-zed-protocol-version",
1358                HeaderValue::from_str(&rpc::PROTOCOL_VERSION.to_string())?,
1359            );
1360            request_headers.insert("x-zed-app-version", HeaderValue::from_str(&app_version)?);
1361            request_headers.insert(
1362                "x-zed-release-channel",
1363                HeaderValue::from_str(release_channel.map(|r| r.dev_name()).unwrap_or("unknown"))?,
1364            );
1365            if let Some(user_agent) = user_agent {
1366                request_headers.insert(http::header::USER_AGENT, user_agent);
1367            }
1368            if let Some(system_id) = system_id {
1369                request_headers.insert("x-zed-system-id", HeaderValue::from_str(&system_id)?);
1370            }
1371            if let Some(metrics_id) = metrics_id {
1372                request_headers.insert("x-zed-metrics-id", HeaderValue::from_str(&metrics_id)?);
1373            }
1374
1375            let (stream, _) = async_tungstenite::tokio::client_async_tls_with_connector_and_config(
1376                request,
1377                stream,
1378                Some(Arc::new(http_client_tls::tls_config()).into()),
1379                None,
1380            )
1381            .await?;
1382
1383            Ok(Connection::new(
1384                stream
1385                    .map_err(|error| anyhow!(error))
1386                    .sink_map_err(|error| anyhow!(error)),
1387            ))
1388        })
1389    }
1390
1391    pub fn authenticate_with_browser(self: &Arc<Self>, cx: &AsyncApp) -> Task<Result<Credentials>> {
1392        let http = self.http.clone();
1393        let this = self.clone();
1394        cx.spawn(async move |cx| {
1395            let background = cx.background_executor().clone();
1396
1397            let (open_url_tx, open_url_rx) = oneshot::channel::<String>();
1398            cx.update(|cx| {
1399                cx.spawn(async move |cx| {
1400                    let url = open_url_rx.await?;
1401                    cx.update(|cx| cx.open_url(&url))
1402                })
1403                .detach_and_log_err(cx);
1404            })
1405            .log_err();
1406
1407            let credentials = background
1408                .clone()
1409                .spawn(async move {
1410                    // Generate a pair of asymmetric encryption keys. The public key will be used by the
1411                    // zed server to encrypt the user's access token, so that it can'be intercepted by
1412                    // any other app running on the user's device.
1413                    let (public_key, private_key) =
1414                        rpc::auth::keypair().expect("failed to generate keypair for auth");
1415                    let public_key_string = String::try_from(public_key)
1416                        .expect("failed to serialize public key for auth");
1417
1418                    if let Some((login, token)) =
1419                        IMPERSONATE_LOGIN.as_ref().zip(ADMIN_API_TOKEN.as_ref())
1420                    {
1421                        if !*USE_WEB_LOGIN {
1422                            eprintln!("authenticate as admin {login}, {token}");
1423
1424                            return this
1425                                .authenticate_as_admin(http, login.clone(), token.clone())
1426                                .await;
1427                        }
1428                    }
1429
1430                    // Start an HTTP server to receive the redirect from Zed's sign-in page.
1431                    let server =
1432                        tiny_http::Server::http("127.0.0.1:0").expect("failed to find open port");
1433                    let port = server.server_addr().port();
1434
1435                    // Open the Zed sign-in page in the user's browser, with query parameters that indicate
1436                    // that the user is signing in from a Zed app running on the same device.
1437                    let mut url = http.build_url(&format!(
1438                        "/native_app_signin?native_app_port={}&native_app_public_key={}",
1439                        port, public_key_string
1440                    ));
1441
1442                    if let Some(impersonate_login) = IMPERSONATE_LOGIN.as_ref() {
1443                        log::info!("impersonating user @{}", impersonate_login);
1444                        write!(&mut url, "&impersonate={}", impersonate_login).unwrap();
1445                    }
1446
1447                    open_url_tx.send(url).log_err();
1448
1449                    #[derive(Deserialize)]
1450                    struct CallbackParams {
1451                        pub user_id: String,
1452                        pub access_token: String,
1453                    }
1454
1455                    // Receive the HTTP request from the user's browser. Retrieve the user id and encrypted
1456                    // access token from the query params.
1457                    //
1458                    // TODO - Avoid ever starting more than one HTTP server. Maybe switch to using a
1459                    // custom URL scheme instead of this local HTTP server.
1460                    let (user_id, access_token) = background
1461                        .spawn(async move {
1462                            for _ in 0..100 {
1463                                if let Some(req) = server.recv_timeout(Duration::from_secs(1))? {
1464                                    let path = req.url();
1465                                    let url = Url::parse(&format!("http://example.com{}", path))
1466                                        .context("failed to parse login notification url")?;
1467                                    let callback_params: CallbackParams =
1468                                        serde_urlencoded::from_str(url.query().unwrap_or_default())
1469                                            .context(
1470                                                "failed to parse sign-in callback query parameters",
1471                                            )?;
1472
1473                                    let post_auth_url =
1474                                        http.build_url("/native_app_signin_succeeded");
1475                                    req.respond(
1476                                        tiny_http::Response::empty(302).with_header(
1477                                            tiny_http::Header::from_bytes(
1478                                                &b"Location"[..],
1479                                                post_auth_url.as_bytes(),
1480                                            )
1481                                            .unwrap(),
1482                                        ),
1483                                    )
1484                                    .context("failed to respond to login http request")?;
1485                                    return Ok((
1486                                        callback_params.user_id,
1487                                        callback_params.access_token,
1488                                    ));
1489                                }
1490                            }
1491
1492                            anyhow::bail!("didn't receive login redirect");
1493                        })
1494                        .await?;
1495
1496                    let access_token = private_key
1497                        .decrypt_string(&access_token)
1498                        .context("failed to decrypt access token")?;
1499
1500                    Ok(Credentials {
1501                        user_id: user_id.parse()?,
1502                        access_token,
1503                    })
1504                })
1505                .await?;
1506
1507            cx.update(|cx| cx.activate(true))?;
1508            Ok(credentials)
1509        })
1510    }
1511
1512    async fn authenticate_as_admin(
1513        self: &Arc<Self>,
1514        http: Arc<HttpClientWithUrl>,
1515        login: String,
1516        api_token: String,
1517    ) -> Result<Credentials> {
1518        #[derive(Serialize)]
1519        struct ImpersonateUserBody {
1520            github_login: String,
1521        }
1522
1523        #[derive(Deserialize)]
1524        struct ImpersonateUserResponse {
1525            user_id: u64,
1526            access_token: String,
1527        }
1528
1529        let url = self
1530            .http
1531            .build_zed_cloud_url("/internal/users/impersonate", &[])?;
1532        let request = Request::post(url.as_str())
1533            .header("Content-Type", "application/json")
1534            .header("Authorization", format!("Bearer {api_token}"))
1535            .body(
1536                serde_json::to_string(&ImpersonateUserBody {
1537                    github_login: login,
1538                })?
1539                .into(),
1540            )?;
1541
1542        let mut response = http.send(request).await?;
1543        let mut body = String::new();
1544        response.body_mut().read_to_string(&mut body).await?;
1545        anyhow::ensure!(
1546            response.status().is_success(),
1547            "admin user request failed {} - {}",
1548            response.status().as_u16(),
1549            body,
1550        );
1551        let response: ImpersonateUserResponse = serde_json::from_str(&body)?;
1552
1553        Ok(Credentials {
1554            user_id: response.user_id,
1555            access_token: response.access_token,
1556        })
1557    }
1558
1559    pub async fn sign_out(self: &Arc<Self>, cx: &AsyncApp) {
1560        self.state.write().credentials = None;
1561        self.cloud_client.clear_credentials();
1562        self.disconnect(cx);
1563
1564        if self.has_credentials(cx).await {
1565            self.credentials_provider
1566                .delete_credentials(cx)
1567                .await
1568                .log_err();
1569        }
1570    }
1571
1572    pub fn disconnect(self: &Arc<Self>, cx: &AsyncApp) {
1573        self.peer.teardown();
1574        self.set_status(Status::SignedOut, cx);
1575    }
1576
1577    pub fn reconnect(self: &Arc<Self>, cx: &AsyncApp) {
1578        self.peer.teardown();
1579        self.set_status(Status::ConnectionLost, cx);
1580    }
1581
1582    fn connection_id(&self) -> Result<ConnectionId> {
1583        if let Status::Connected { connection_id, .. } = *self.status().borrow() {
1584            Ok(connection_id)
1585        } else {
1586            anyhow::bail!("not connected");
1587        }
1588    }
1589
1590    pub fn send<T: EnvelopedMessage>(&self, message: T) -> Result<()> {
1591        log::debug!("rpc send. client_id:{}, name:{}", self.id(), T::NAME);
1592        self.peer.send(self.connection_id()?, message)
1593    }
1594
1595    pub fn request<T: RequestMessage>(
1596        &self,
1597        request: T,
1598    ) -> impl Future<Output = Result<T::Response>> + use<T> {
1599        self.request_envelope(request)
1600            .map_ok(|envelope| envelope.payload)
1601    }
1602
1603    pub fn request_stream<T: RequestMessage>(
1604        &self,
1605        request: T,
1606    ) -> impl Future<Output = Result<impl Stream<Item = Result<T::Response>>>> {
1607        let client_id = self.id.load(Ordering::SeqCst);
1608        log::debug!(
1609            "rpc request start. client_id:{}. name:{}",
1610            client_id,
1611            T::NAME
1612        );
1613        let response = self
1614            .connection_id()
1615            .map(|conn_id| self.peer.request_stream(conn_id, request));
1616        async move {
1617            let response = response?.await;
1618            log::debug!(
1619                "rpc request finish. client_id:{}. name:{}",
1620                client_id,
1621                T::NAME
1622            );
1623            response
1624        }
1625    }
1626
1627    pub fn request_envelope<T: RequestMessage>(
1628        &self,
1629        request: T,
1630    ) -> impl Future<Output = Result<TypedEnvelope<T::Response>>> + use<T> {
1631        let client_id = self.id();
1632        log::debug!(
1633            "rpc request start. client_id:{}. name:{}",
1634            client_id,
1635            T::NAME
1636        );
1637        let response = self
1638            .connection_id()
1639            .map(|conn_id| self.peer.request_envelope(conn_id, request));
1640        async move {
1641            let response = response?.await;
1642            log::debug!(
1643                "rpc request finish. client_id:{}. name:{}",
1644                client_id,
1645                T::NAME
1646            );
1647            response
1648        }
1649    }
1650
1651    pub fn request_dynamic(
1652        &self,
1653        envelope: proto::Envelope,
1654        request_type: &'static str,
1655    ) -> impl Future<Output = Result<proto::Envelope>> + use<> {
1656        let client_id = self.id();
1657        log::debug!(
1658            "rpc request start. client_id:{}. name:{}",
1659            client_id,
1660            request_type
1661        );
1662        let response = self
1663            .connection_id()
1664            .map(|conn_id| self.peer.request_dynamic(conn_id, envelope, request_type));
1665        async move {
1666            let response = response?.await;
1667            log::debug!(
1668                "rpc request finish. client_id:{}. name:{}",
1669                client_id,
1670                request_type
1671            );
1672            Ok(response?.0)
1673        }
1674    }
1675
1676    fn handle_message(self: &Arc<Client>, message: Box<dyn AnyTypedEnvelope>, cx: &AsyncApp) {
1677        let sender_id = message.sender_id();
1678        let request_id = message.message_id();
1679        let type_name = message.payload_type_name();
1680        let original_sender_id = message.original_sender_id();
1681
1682        if let Some(future) = ProtoMessageHandlerSet::handle_message(
1683            &self.handler_set,
1684            message,
1685            self.clone().into(),
1686            cx.clone(),
1687        ) {
1688            let client_id = self.id();
1689            log::debug!(
1690                "rpc message received. client_id:{}, sender_id:{:?}, type:{}",
1691                client_id,
1692                original_sender_id,
1693                type_name
1694            );
1695            cx.spawn(async move |_| match future.await {
1696                Ok(()) => {
1697                    log::debug!("rpc message handled. client_id:{client_id}, sender_id:{original_sender_id:?}, type:{type_name}");
1698                }
1699                Err(error) => {
1700                    log::error!("error handling message. client_id:{client_id}, sender_id:{original_sender_id:?}, type:{type_name}, error:{error:#}");
1701                }
1702            })
1703            .detach();
1704        } else {
1705            log::info!("unhandled message {}", type_name);
1706            self.peer
1707                .respond_with_unhandled_message(sender_id.into(), request_id, type_name)
1708                .log_err();
1709        }
1710    }
1711
1712    pub fn add_message_to_client_handler(
1713        self: &Arc<Client>,
1714        handler: impl Fn(&MessageToClient, &mut App) + Send + Sync + 'static,
1715    ) {
1716        self.message_to_client_handlers
1717            .lock()
1718            .push(Box::new(handler));
1719    }
1720
1721    fn handle_message_to_client(self: &Arc<Client>, message: MessageToClient, cx: &AsyncApp) {
1722        cx.update(|cx| {
1723            for handler in self.message_to_client_handlers.lock().iter() {
1724                handler(&message, cx);
1725            }
1726        })
1727        .ok();
1728    }
1729
1730    pub fn telemetry(&self) -> &Arc<Telemetry> {
1731        &self.telemetry
1732    }
1733}
1734
1735impl ProtoClient for Client {
1736    fn request(
1737        &self,
1738        envelope: proto::Envelope,
1739        request_type: &'static str,
1740    ) -> BoxFuture<'static, Result<proto::Envelope>> {
1741        self.request_dynamic(envelope, request_type).boxed()
1742    }
1743
1744    fn send(&self, envelope: proto::Envelope, message_type: &'static str) -> Result<()> {
1745        log::debug!("rpc send. client_id:{}, name:{}", self.id(), message_type);
1746        let connection_id = self.connection_id()?;
1747        self.peer.send_dynamic(connection_id, envelope)
1748    }
1749
1750    fn send_response(&self, envelope: proto::Envelope, message_type: &'static str) -> Result<()> {
1751        log::debug!(
1752            "rpc respond. client_id:{}, name:{}",
1753            self.id(),
1754            message_type
1755        );
1756        let connection_id = self.connection_id()?;
1757        self.peer.send_dynamic(connection_id, envelope)
1758    }
1759
1760    fn message_handler_set(&self) -> &parking_lot::Mutex<ProtoMessageHandlerSet> {
1761        &self.handler_set
1762    }
1763
1764    fn is_via_collab(&self) -> bool {
1765        true
1766    }
1767}
1768
1769/// prefix for the zed:// url scheme
1770pub const ZED_URL_SCHEME: &str = "zed";
1771
1772/// Parses the given link into a Zed link.
1773///
1774/// Returns a [`Some`] containing the unprefixed link if the link is a Zed link.
1775/// Returns [`None`] otherwise.
1776pub fn parse_zed_link<'a>(link: &'a str, cx: &App) -> Option<&'a str> {
1777    let server_url = &ClientSettings::get_global(cx).server_url;
1778    if let Some(stripped) = link
1779        .strip_prefix(server_url)
1780        .and_then(|result| result.strip_prefix('/'))
1781    {
1782        return Some(stripped);
1783    }
1784    if let Some(stripped) = link
1785        .strip_prefix(ZED_URL_SCHEME)
1786        .and_then(|result| result.strip_prefix("://"))
1787    {
1788        return Some(stripped);
1789    }
1790
1791    None
1792}
1793
1794#[cfg(test)]
1795mod tests {
1796    use super::*;
1797    use crate::test::{FakeServer, parse_authorization_header};
1798
1799    use clock::FakeSystemClock;
1800    use gpui::{AppContext as _, BackgroundExecutor, TestAppContext};
1801    use http_client::FakeHttpClient;
1802    use parking_lot::Mutex;
1803    use proto::TypedEnvelope;
1804    use settings::SettingsStore;
1805    use std::future;
1806
1807    #[gpui::test(iterations = 10)]
1808    async fn test_reconnection(cx: &mut TestAppContext) {
1809        init_test(cx);
1810        let user_id = 5;
1811        let client = cx.update(|cx| {
1812            Client::new(
1813                Arc::new(FakeSystemClock::new()),
1814                FakeHttpClient::with_404_response(),
1815                cx,
1816            )
1817        });
1818        let server = FakeServer::for_client(user_id, &client, cx).await;
1819        let mut status = client.status();
1820        assert!(matches!(
1821            status.next().await,
1822            Some(Status::Connected { .. })
1823        ));
1824        assert_eq!(server.auth_count(), 1);
1825
1826        server.forbid_connections();
1827        server.disconnect();
1828        while !matches!(status.next().await, Some(Status::ReconnectionError { .. })) {}
1829
1830        server.allow_connections();
1831        cx.executor().advance_clock(Duration::from_secs(10));
1832        while !matches!(status.next().await, Some(Status::Connected { .. })) {}
1833        assert_eq!(server.auth_count(), 1); // Client reused the cached credentials when reconnecting
1834
1835        server.forbid_connections();
1836        server.disconnect();
1837        while !matches!(status.next().await, Some(Status::ReconnectionError { .. })) {}
1838
1839        // Clear cached credentials after authentication fails
1840        server.roll_access_token();
1841        server.allow_connections();
1842        cx.executor().run_until_parked();
1843        cx.executor().advance_clock(Duration::from_secs(10));
1844        while !matches!(status.next().await, Some(Status::Connected { .. })) {}
1845        assert_eq!(server.auth_count(), 2); // Client re-authenticated due to an invalid token
1846    }
1847
1848    #[gpui::test(iterations = 10)]
1849    async fn test_auth_failure_during_reconnection(cx: &mut TestAppContext) {
1850        init_test(cx);
1851        let http_client = FakeHttpClient::with_200_response();
1852        let client =
1853            cx.update(|cx| Client::new(Arc::new(FakeSystemClock::new()), http_client.clone(), cx));
1854        let server = FakeServer::for_client(42, &client, cx).await;
1855        let mut status = client.status();
1856        assert!(matches!(
1857            status.next().await,
1858            Some(Status::Connected { .. })
1859        ));
1860        assert_eq!(server.auth_count(), 1);
1861
1862        // Simulate an auth failure during reconnection.
1863        http_client
1864            .as_fake()
1865            .replace_handler(|_, _request| async move {
1866                Ok(http_client::Response::builder()
1867                    .status(503)
1868                    .body("".into())
1869                    .unwrap())
1870            });
1871        server.disconnect();
1872        while !matches!(status.next().await, Some(Status::ReconnectionError { .. })) {}
1873
1874        // Restore the ability to authenticate.
1875        http_client
1876            .as_fake()
1877            .replace_handler(|_, _request| async move {
1878                Ok(http_client::Response::builder()
1879                    .status(200)
1880                    .body("".into())
1881                    .unwrap())
1882            });
1883        cx.executor().advance_clock(Duration::from_secs(10));
1884        while !matches!(status.next().await, Some(Status::Connected { .. })) {}
1885        assert_eq!(server.auth_count(), 1); // Client reused the cached credentials when reconnecting
1886    }
1887
1888    #[gpui::test(iterations = 10)]
1889    async fn test_connection_timeout(executor: BackgroundExecutor, cx: &mut TestAppContext) {
1890        init_test(cx);
1891        let user_id = 5;
1892        let client = cx.update(|cx| {
1893            Client::new(
1894                Arc::new(FakeSystemClock::new()),
1895                FakeHttpClient::with_404_response(),
1896                cx,
1897            )
1898        });
1899        let mut status = client.status();
1900
1901        // Time out when client tries to connect.
1902        client.override_authenticate(move |cx| {
1903            cx.background_spawn(async move {
1904                Ok(Credentials {
1905                    user_id,
1906                    access_token: "token".into(),
1907                })
1908            })
1909        });
1910        client.override_establish_connection(|_, cx| {
1911            cx.background_spawn(async move {
1912                future::pending::<()>().await;
1913                unreachable!()
1914            })
1915        });
1916        let auth_and_connect = cx.spawn({
1917            let client = client.clone();
1918            |cx| async move { client.connect(false, &cx).await }
1919        });
1920        executor.run_until_parked();
1921        assert!(matches!(status.next().await, Some(Status::Connecting)));
1922
1923        executor.advance_clock(CONNECTION_TIMEOUT);
1924        assert!(matches!(status.next().await, Some(Status::ConnectionError)));
1925        auth_and_connect.await.into_response().unwrap_err();
1926
1927        // Allow the connection to be established.
1928        let server = FakeServer::for_client(user_id, &client, cx).await;
1929        assert!(matches!(
1930            status.next().await,
1931            Some(Status::Connected { .. })
1932        ));
1933
1934        // Disconnect client.
1935        server.forbid_connections();
1936        server.disconnect();
1937        while !matches!(status.next().await, Some(Status::ReconnectionError { .. })) {}
1938
1939        // Time out when re-establishing the connection.
1940        server.allow_connections();
1941        client.override_establish_connection(|_, cx| {
1942            cx.background_spawn(async move {
1943                future::pending::<()>().await;
1944                unreachable!()
1945            })
1946        });
1947        executor.advance_clock(2 * INITIAL_RECONNECTION_DELAY);
1948        assert!(matches!(status.next().await, Some(Status::Reconnecting)));
1949
1950        executor.advance_clock(CONNECTION_TIMEOUT);
1951        assert!(matches!(
1952            status.next().await,
1953            Some(Status::ReconnectionError { .. })
1954        ));
1955    }
1956
1957    #[gpui::test(iterations = 10)]
1958    async fn test_reauthenticate_only_if_unauthorized(cx: &mut TestAppContext) {
1959        init_test(cx);
1960        let auth_count = Arc::new(Mutex::new(0));
1961        let http_client = FakeHttpClient::create(|_request| async move {
1962            Ok(http_client::Response::builder()
1963                .status(200)
1964                .body("".into())
1965                .unwrap())
1966        });
1967        let client =
1968            cx.update(|cx| Client::new(Arc::new(FakeSystemClock::new()), http_client.clone(), cx));
1969        client.override_authenticate({
1970            let auth_count = auth_count.clone();
1971            move |cx| {
1972                let auth_count = auth_count.clone();
1973                cx.background_spawn(async move {
1974                    *auth_count.lock() += 1;
1975                    Ok(Credentials {
1976                        user_id: 1,
1977                        access_token: auth_count.lock().to_string(),
1978                    })
1979                })
1980            }
1981        });
1982
1983        let credentials = client.sign_in(false, &cx.to_async()).await.unwrap();
1984        assert_eq!(*auth_count.lock(), 1);
1985        assert_eq!(credentials.access_token, "1");
1986
1987        // If credentials are still valid, signing in doesn't trigger authentication.
1988        let credentials = client.sign_in(false, &cx.to_async()).await.unwrap();
1989        assert_eq!(*auth_count.lock(), 1);
1990        assert_eq!(credentials.access_token, "1");
1991
1992        // If the server is unavailable, signing in doesn't trigger authentication.
1993        http_client
1994            .as_fake()
1995            .replace_handler(|_, _request| async move {
1996                Ok(http_client::Response::builder()
1997                    .status(503)
1998                    .body("".into())
1999                    .unwrap())
2000            });
2001        client.sign_in(false, &cx.to_async()).await.unwrap_err();
2002        assert_eq!(*auth_count.lock(), 1);
2003
2004        // If credentials became invalid, signing in triggers authentication.
2005        http_client
2006            .as_fake()
2007            .replace_handler(|_, request| async move {
2008                let credentials = parse_authorization_header(&request).unwrap();
2009                if credentials.access_token == "2" {
2010                    Ok(http_client::Response::builder()
2011                        .status(200)
2012                        .body("".into())
2013                        .unwrap())
2014                } else {
2015                    Ok(http_client::Response::builder()
2016                        .status(401)
2017                        .body("".into())
2018                        .unwrap())
2019                }
2020            });
2021        let credentials = client.sign_in(false, &cx.to_async()).await.unwrap();
2022        assert_eq!(*auth_count.lock(), 2);
2023        assert_eq!(credentials.access_token, "2");
2024    }
2025
2026    #[gpui::test(iterations = 10)]
2027    async fn test_authenticating_more_than_once(
2028        cx: &mut TestAppContext,
2029        executor: BackgroundExecutor,
2030    ) {
2031        init_test(cx);
2032        let auth_count = Arc::new(Mutex::new(0));
2033        let dropped_auth_count = Arc::new(Mutex::new(0));
2034        let client = cx.update(|cx| {
2035            Client::new(
2036                Arc::new(FakeSystemClock::new()),
2037                FakeHttpClient::with_404_response(),
2038                cx,
2039            )
2040        });
2041        client.override_authenticate({
2042            let auth_count = auth_count.clone();
2043            let dropped_auth_count = dropped_auth_count.clone();
2044            move |cx| {
2045                let auth_count = auth_count.clone();
2046                let dropped_auth_count = dropped_auth_count.clone();
2047                cx.background_spawn(async move {
2048                    *auth_count.lock() += 1;
2049                    let _drop = util::defer(move || *dropped_auth_count.lock() += 1);
2050                    future::pending::<()>().await;
2051                    unreachable!()
2052                })
2053            }
2054        });
2055
2056        let _authenticate = cx.spawn({
2057            let client = client.clone();
2058            move |cx| async move { client.connect(false, &cx).await }
2059        });
2060        executor.run_until_parked();
2061        assert_eq!(*auth_count.lock(), 1);
2062        assert_eq!(*dropped_auth_count.lock(), 0);
2063
2064        let _authenticate = cx.spawn(|cx| async move { client.connect(false, &cx).await });
2065        executor.run_until_parked();
2066        assert_eq!(*auth_count.lock(), 2);
2067        assert_eq!(*dropped_auth_count.lock(), 1);
2068    }
2069
2070    #[gpui::test]
2071    async fn test_subscribing_to_entity(cx: &mut TestAppContext) {
2072        init_test(cx);
2073        let user_id = 5;
2074        let client = cx.update(|cx| {
2075            Client::new(
2076                Arc::new(FakeSystemClock::new()),
2077                FakeHttpClient::with_404_response(),
2078                cx,
2079            )
2080        });
2081        let server = FakeServer::for_client(user_id, &client, cx).await;
2082
2083        let (done_tx1, done_rx1) = smol::channel::unbounded();
2084        let (done_tx2, done_rx2) = smol::channel::unbounded();
2085        AnyProtoClient::from(client.clone()).add_entity_message_handler(
2086            move |entity: Entity<TestEntity>, _: TypedEnvelope<proto::JoinProject>, cx| {
2087                match entity.read_with(&cx, |entity, _| entity.id).unwrap() {
2088                    1 => done_tx1.try_send(()).unwrap(),
2089                    2 => done_tx2.try_send(()).unwrap(),
2090                    _ => unreachable!(),
2091                }
2092                async { Ok(()) }
2093            },
2094        );
2095        let entity1 = cx.new(|_| TestEntity {
2096            id: 1,
2097            subscription: None,
2098        });
2099        let entity2 = cx.new(|_| TestEntity {
2100            id: 2,
2101            subscription: None,
2102        });
2103        let entity3 = cx.new(|_| TestEntity {
2104            id: 3,
2105            subscription: None,
2106        });
2107
2108        let _subscription1 = client
2109            .subscribe_to_entity(1)
2110            .unwrap()
2111            .set_entity(&entity1, &cx.to_async());
2112        let _subscription2 = client
2113            .subscribe_to_entity(2)
2114            .unwrap()
2115            .set_entity(&entity2, &cx.to_async());
2116        // Ensure dropping a subscription for the same entity type still allows receiving of
2117        // messages for other entity IDs of the same type.
2118        let subscription3 = client
2119            .subscribe_to_entity(3)
2120            .unwrap()
2121            .set_entity(&entity3, &cx.to_async());
2122        drop(subscription3);
2123
2124        server.send(proto::JoinProject {
2125            project_id: 1,
2126            committer_name: None,
2127            committer_email: None,
2128        });
2129        server.send(proto::JoinProject {
2130            project_id: 2,
2131            committer_name: None,
2132            committer_email: None,
2133        });
2134        done_rx1.recv().await.unwrap();
2135        done_rx2.recv().await.unwrap();
2136    }
2137
2138    #[gpui::test]
2139    async fn test_subscribing_after_dropping_subscription(cx: &mut TestAppContext) {
2140        init_test(cx);
2141        let user_id = 5;
2142        let client = cx.update(|cx| {
2143            Client::new(
2144                Arc::new(FakeSystemClock::new()),
2145                FakeHttpClient::with_404_response(),
2146                cx,
2147            )
2148        });
2149        let server = FakeServer::for_client(user_id, &client, cx).await;
2150
2151        let entity = cx.new(|_| TestEntity::default());
2152        let (done_tx1, _done_rx1) = smol::channel::unbounded();
2153        let (done_tx2, done_rx2) = smol::channel::unbounded();
2154        let subscription1 = client.add_message_handler(
2155            entity.downgrade(),
2156            move |_, _: TypedEnvelope<proto::Ping>, _| {
2157                done_tx1.try_send(()).unwrap();
2158                async { Ok(()) }
2159            },
2160        );
2161        drop(subscription1);
2162        let _subscription2 = client.add_message_handler(
2163            entity.downgrade(),
2164            move |_, _: TypedEnvelope<proto::Ping>, _| {
2165                done_tx2.try_send(()).unwrap();
2166                async { Ok(()) }
2167            },
2168        );
2169        server.send(proto::Ping {});
2170        done_rx2.recv().await.unwrap();
2171    }
2172
2173    #[gpui::test]
2174    async fn test_dropping_subscription_in_handler(cx: &mut TestAppContext) {
2175        init_test(cx);
2176        let user_id = 5;
2177        let client = cx.update(|cx| {
2178            Client::new(
2179                Arc::new(FakeSystemClock::new()),
2180                FakeHttpClient::with_404_response(),
2181                cx,
2182            )
2183        });
2184        let server = FakeServer::for_client(user_id, &client, cx).await;
2185
2186        let entity = cx.new(|_| TestEntity::default());
2187        let (done_tx, done_rx) = smol::channel::unbounded();
2188        let subscription = client.add_message_handler(
2189            entity.clone().downgrade(),
2190            move |entity: Entity<TestEntity>, _: TypedEnvelope<proto::Ping>, mut cx| {
2191                entity
2192                    .update(&mut cx, |entity, _| entity.subscription.take())
2193                    .unwrap();
2194                done_tx.try_send(()).unwrap();
2195                async { Ok(()) }
2196            },
2197        );
2198        entity.update(cx, |entity, _| {
2199            entity.subscription = Some(subscription);
2200        });
2201        server.send(proto::Ping {});
2202        done_rx.recv().await.unwrap();
2203    }
2204
2205    #[derive(Default)]
2206    struct TestEntity {
2207        id: usize,
2208        subscription: Option<Subscription>,
2209    }
2210
2211    fn init_test(cx: &mut TestAppContext) {
2212        cx.update(|cx| {
2213            let settings_store = SettingsStore::test(cx);
2214            cx.set_global(settings_store);
2215            init_settings(cx);
2216        });
2217    }
2218}