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