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