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