1mod store;
   2
   3use super::{
   4    auth::process_auth_header,
   5    db::{ChannelId, MessageId, UserId},
   6    AppState,
   7};
   8use anyhow::anyhow;
   9use async_io::Timer;
  10use async_std::task;
  11use async_tungstenite::{tungstenite::protocol::Role, WebSocketStream};
  12use collections::{HashMap, HashSet};
  13use futures::{channel::mpsc, future::BoxFuture, FutureExt, SinkExt, StreamExt};
  14use log::{as_debug, as_display};
  15use parking_lot::{RwLock, RwLockReadGuard, RwLockWriteGuard};
  16use rpc::{
  17    proto::{self, AnyTypedEnvelope, EntityMessage, EnvelopedMessage, RequestMessage},
  18    Connection, ConnectionId, Peer, TypedEnvelope,
  19};
  20use sha1::{Digest as _, Sha1};
  21use std::{
  22    any::TypeId,
  23    future::Future,
  24    sync::Arc,
  25    time::{Duration, Instant},
  26};
  27use store::{Store, Worktree};
  28use surf::StatusCode;
  29use tide::{
  30    http::headers::{HeaderName, CONNECTION, UPGRADE},
  31    Request, Response,
  32};
  33use time::OffsetDateTime;
  34
  35type MessageHandler = Box<
  36    dyn Send
  37        + Sync
  38        + Fn(Arc<Server>, Box<dyn AnyTypedEnvelope>) -> BoxFuture<'static, tide::Result<()>>,
  39>;
  40
  41pub struct Server {
  42    peer: Arc<Peer>,
  43    store: RwLock<Store>,
  44    app_state: Arc<AppState>,
  45    handlers: HashMap<TypeId, MessageHandler>,
  46    notifications: Option<mpsc::UnboundedSender<()>>,
  47}
  48
  49pub trait Executor: Send + Clone {
  50    type Timer: Send + Future;
  51    fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F);
  52    fn timer(&self, duration: Duration) -> Self::Timer;
  53}
  54
  55#[derive(Clone)]
  56pub struct RealExecutor;
  57
  58const MESSAGE_COUNT_PER_PAGE: usize = 100;
  59const MAX_MESSAGE_LEN: usize = 1024;
  60
  61impl Server {
  62    pub fn new(
  63        app_state: Arc<AppState>,
  64        peer: Arc<Peer>,
  65        notifications: Option<mpsc::UnboundedSender<()>>,
  66    ) -> Arc<Self> {
  67        let mut server = Self {
  68            peer,
  69            app_state,
  70            store: Default::default(),
  71            handlers: Default::default(),
  72            notifications,
  73        };
  74
  75        server
  76            .add_request_handler(Server::ping)
  77            .add_request_handler(Server::register_project)
  78            .add_message_handler(Server::unregister_project)
  79            .add_request_handler(Server::share_project)
  80            .add_message_handler(Server::unshare_project)
  81            .add_request_handler(Server::join_project)
  82            .add_message_handler(Server::leave_project)
  83            .add_request_handler(Server::register_worktree)
  84            .add_message_handler(Server::unregister_worktree)
  85            .add_request_handler(Server::update_worktree)
  86            .add_message_handler(Server::start_language_server)
  87            .add_message_handler(Server::update_language_server)
  88            .add_message_handler(Server::update_diagnostic_summary)
  89            .add_request_handler(Server::forward_project_request::<proto::GetDefinition>)
  90            .add_request_handler(Server::forward_project_request::<proto::GetReferences>)
  91            .add_request_handler(Server::forward_project_request::<proto::SearchProject>)
  92            .add_request_handler(Server::forward_project_request::<proto::GetDocumentHighlights>)
  93            .add_request_handler(Server::forward_project_request::<proto::GetProjectSymbols>)
  94            .add_request_handler(Server::forward_project_request::<proto::OpenBufferForSymbol>)
  95            .add_request_handler(Server::forward_project_request::<proto::OpenBufferById>)
  96            .add_request_handler(Server::forward_project_request::<proto::OpenBufferByPath>)
  97            .add_request_handler(Server::forward_project_request::<proto::GetCompletions>)
  98            .add_request_handler(
  99                Server::forward_project_request::<proto::ApplyCompletionAdditionalEdits>,
 100            )
 101            .add_request_handler(Server::forward_project_request::<proto::GetCodeActions>)
 102            .add_request_handler(Server::forward_project_request::<proto::ApplyCodeAction>)
 103            .add_request_handler(Server::forward_project_request::<proto::PrepareRename>)
 104            .add_request_handler(Server::forward_project_request::<proto::PerformRename>)
 105            .add_request_handler(Server::forward_project_request::<proto::ReloadBuffers>)
 106            .add_request_handler(Server::forward_project_request::<proto::FormatBuffers>)
 107            .add_request_handler(Server::update_buffer)
 108            .add_message_handler(Server::update_buffer_file)
 109            .add_message_handler(Server::buffer_reloaded)
 110            .add_message_handler(Server::buffer_saved)
 111            .add_request_handler(Server::save_buffer)
 112            .add_request_handler(Server::get_channels)
 113            .add_request_handler(Server::get_users)
 114            .add_request_handler(Server::join_channel)
 115            .add_message_handler(Server::leave_channel)
 116            .add_request_handler(Server::send_channel_message)
 117            .add_request_handler(Server::follow)
 118            .add_message_handler(Server::unfollow)
 119            .add_message_handler(Server::update_followers)
 120            .add_request_handler(Server::get_channel_messages);
 121
 122        Arc::new(server)
 123    }
 124
 125    fn add_message_handler<F, Fut, M>(&mut self, handler: F) -> &mut Self
 126    where
 127        F: 'static + Send + Sync + Fn(Arc<Self>, TypedEnvelope<M>) -> Fut,
 128        Fut: 'static + Send + Future<Output = tide::Result<()>>,
 129        M: EnvelopedMessage,
 130    {
 131        let prev_handler = self.handlers.insert(
 132            TypeId::of::<M>(),
 133            Box::new(move |server, envelope| {
 134                let envelope = envelope.into_any().downcast::<TypedEnvelope<M>>().unwrap();
 135                (handler)(server, *envelope).boxed()
 136            }),
 137        );
 138        if prev_handler.is_some() {
 139            panic!("registered a handler for the same message twice");
 140        }
 141        self
 142    }
 143
 144    fn add_request_handler<F, Fut, M>(&mut self, handler: F) -> &mut Self
 145    where
 146        F: 'static + Send + Sync + Fn(Arc<Self>, TypedEnvelope<M>) -> Fut,
 147        Fut: 'static + Send + Future<Output = tide::Result<M::Response>>,
 148        M: RequestMessage,
 149    {
 150        self.add_message_handler(move |server, envelope| {
 151            let receipt = envelope.receipt();
 152            let response = (handler)(server.clone(), envelope);
 153            async move {
 154                match response.await {
 155                    Ok(response) => {
 156                        server.peer.respond(receipt, response)?;
 157                        Ok(())
 158                    }
 159                    Err(error) => {
 160                        server.peer.respond_with_error(
 161                            receipt,
 162                            proto::Error {
 163                                message: error.to_string(),
 164                            },
 165                        )?;
 166                        Err(error)
 167                    }
 168                }
 169            }
 170        })
 171    }
 172
 173    pub fn handle_connection<E: Executor>(
 174        self: &Arc<Self>,
 175        connection: Connection,
 176        addr: String,
 177        user_id: UserId,
 178        mut send_connection_id: Option<mpsc::Sender<ConnectionId>>,
 179        executor: E,
 180    ) -> impl Future<Output = ()> {
 181        let mut this = self.clone();
 182        async move {
 183            let (connection_id, handle_io, mut incoming_rx) = this
 184                .peer
 185                .add_connection(connection, {
 186                    let executor = executor.clone();
 187                    move |duration| {
 188                        let timer = executor.timer(duration);
 189                        async move {
 190                            timer.await;
 191                        }
 192                    }
 193                })
 194                .await;
 195
 196            if let Some(send_connection_id) = send_connection_id.as_mut() {
 197                let _ = send_connection_id.send(connection_id).await;
 198            }
 199
 200            this.state_mut().add_connection(connection_id, user_id);
 201            if let Err(err) = this.update_contacts_for_users(&[user_id]) {
 202                log::error!("error updating contacts for {:?}: {}", user_id, err);
 203            }
 204
 205            let handle_io = handle_io.fuse();
 206            futures::pin_mut!(handle_io);
 207            loop {
 208                let next_message = incoming_rx.next().fuse();
 209                futures::pin_mut!(next_message);
 210                futures::select_biased! {
 211                    result = handle_io => {
 212                        if let Err(err) = result {
 213                            log::error!("error handling rpc connection {:?} - {:?}", addr, err);
 214                        }
 215                        break;
 216                    }
 217                    message = next_message => {
 218                        if let Some(message) = message {
 219                            let start_time = Instant::now();
 220                            let type_name = message.payload_type_name();
 221                            log::info!(connection_id = connection_id.0, type_name = type_name; "rpc message received");
 222                            if let Some(handler) = this.handlers.get(&message.payload_type_id()) {
 223                                let notifications = this.notifications.clone();
 224                                let is_background = message.is_background();
 225                                let handle_message = (handler)(this.clone(), message);
 226                                let handle_message = async move {
 227                                    if let Err(err) = handle_message.await {
 228                                        log::error!(connection_id = connection_id.0, type = type_name, error = as_display!(err); "rpc message error");
 229                                    } else {
 230                                        log::info!(connection_id = connection_id.0, type = type_name, duration = as_debug!(start_time.elapsed()); "rpc message handled");
 231                                    }
 232                                    if let Some(mut notifications) = notifications {
 233                                        let _ = notifications.send(()).await;
 234                                    }
 235                                };
 236                                if is_background {
 237                                    executor.spawn_detached(handle_message);
 238                                } else {
 239                                    handle_message.await;
 240                                }
 241                            } else {
 242                                log::warn!("unhandled message: {}", type_name);
 243                            }
 244                        } else {
 245                            log::info!(address = as_debug!(addr); "rpc connection closed");
 246                            break;
 247                        }
 248                    }
 249                }
 250            }
 251
 252            if let Err(err) = this.sign_out(connection_id).await {
 253                log::error!("error signing out connection {:?} - {:?}", addr, err);
 254            }
 255        }
 256    }
 257
 258    async fn sign_out(self: &mut Arc<Self>, connection_id: ConnectionId) -> tide::Result<()> {
 259        self.peer.disconnect(connection_id);
 260        let removed_connection = self.state_mut().remove_connection(connection_id)?;
 261
 262        for (project_id, project) in removed_connection.hosted_projects {
 263            if let Some(share) = project.share {
 264                broadcast(
 265                    connection_id,
 266                    share.guests.keys().copied().collect(),
 267                    |conn_id| {
 268                        self.peer
 269                            .send(conn_id, proto::UnshareProject { project_id })
 270                    },
 271                )?;
 272            }
 273        }
 274
 275        for (project_id, peer_ids) in removed_connection.guest_project_ids {
 276            broadcast(connection_id, peer_ids, |conn_id| {
 277                self.peer.send(
 278                    conn_id,
 279                    proto::RemoveProjectCollaborator {
 280                        project_id,
 281                        peer_id: connection_id.0,
 282                    },
 283                )
 284            })?;
 285        }
 286
 287        self.update_contacts_for_users(removed_connection.contact_ids.iter())?;
 288        Ok(())
 289    }
 290
 291    async fn ping(self: Arc<Server>, _: TypedEnvelope<proto::Ping>) -> tide::Result<proto::Ack> {
 292        Ok(proto::Ack {})
 293    }
 294
 295    async fn register_project(
 296        mut self: Arc<Server>,
 297        request: TypedEnvelope<proto::RegisterProject>,
 298    ) -> tide::Result<proto::RegisterProjectResponse> {
 299        let project_id = {
 300            let mut state = self.state_mut();
 301            let user_id = state.user_id_for_connection(request.sender_id)?;
 302            state.register_project(request.sender_id, user_id)
 303        };
 304        Ok(proto::RegisterProjectResponse { project_id })
 305    }
 306
 307    async fn unregister_project(
 308        mut self: Arc<Server>,
 309        request: TypedEnvelope<proto::UnregisterProject>,
 310    ) -> tide::Result<()> {
 311        let project = self
 312            .state_mut()
 313            .unregister_project(request.payload.project_id, request.sender_id)?;
 314        self.update_contacts_for_users(project.authorized_user_ids().iter())?;
 315        Ok(())
 316    }
 317
 318    async fn share_project(
 319        mut self: Arc<Server>,
 320        request: TypedEnvelope<proto::ShareProject>,
 321    ) -> tide::Result<proto::Ack> {
 322        self.state_mut()
 323            .share_project(request.payload.project_id, request.sender_id);
 324        Ok(proto::Ack {})
 325    }
 326
 327    async fn unshare_project(
 328        mut self: Arc<Server>,
 329        request: TypedEnvelope<proto::UnshareProject>,
 330    ) -> tide::Result<()> {
 331        let project_id = request.payload.project_id;
 332        let project = self
 333            .state_mut()
 334            .unshare_project(project_id, request.sender_id)?;
 335
 336        broadcast(request.sender_id, project.connection_ids, |conn_id| {
 337            self.peer
 338                .send(conn_id, proto::UnshareProject { project_id })
 339        })?;
 340        self.update_contacts_for_users(&project.authorized_user_ids)?;
 341        Ok(())
 342    }
 343
 344    async fn join_project(
 345        mut self: Arc<Server>,
 346        request: TypedEnvelope<proto::JoinProject>,
 347    ) -> tide::Result<proto::JoinProjectResponse> {
 348        let project_id = request.payload.project_id;
 349
 350        let user_id = self.state().user_id_for_connection(request.sender_id)?;
 351        let (response, connection_ids, contact_user_ids) = self
 352            .state_mut()
 353            .join_project(request.sender_id, user_id, project_id)
 354            .and_then(|joined| {
 355                let share = joined.project.share()?;
 356                let peer_count = share.guests.len();
 357                let mut collaborators = Vec::with_capacity(peer_count);
 358                collaborators.push(proto::Collaborator {
 359                    peer_id: joined.project.host_connection_id.0,
 360                    replica_id: 0,
 361                    user_id: joined.project.host_user_id.to_proto(),
 362                });
 363                let worktrees = share
 364                    .worktrees
 365                    .iter()
 366                    .filter_map(|(id, shared_worktree)| {
 367                        let worktree = joined.project.worktrees.get(&id)?;
 368                        Some(proto::Worktree {
 369                            id: *id,
 370                            root_name: worktree.root_name.clone(),
 371                            entries: shared_worktree.entries.values().cloned().collect(),
 372                            diagnostic_summaries: shared_worktree
 373                                .diagnostic_summaries
 374                                .values()
 375                                .cloned()
 376                                .collect(),
 377                            visible: worktree.visible,
 378                        })
 379                    })
 380                    .collect();
 381                for (peer_conn_id, (peer_replica_id, peer_user_id)) in &share.guests {
 382                    if *peer_conn_id != request.sender_id {
 383                        collaborators.push(proto::Collaborator {
 384                            peer_id: peer_conn_id.0,
 385                            replica_id: *peer_replica_id as u32,
 386                            user_id: peer_user_id.to_proto(),
 387                        });
 388                    }
 389                }
 390                let response = proto::JoinProjectResponse {
 391                    worktrees,
 392                    replica_id: joined.replica_id as u32,
 393                    collaborators,
 394                    language_servers: joined.project.language_servers.clone(),
 395                };
 396                let connection_ids = joined.project.connection_ids();
 397                let contact_user_ids = joined.project.authorized_user_ids();
 398                Ok((response, connection_ids, contact_user_ids))
 399            })?;
 400
 401        broadcast(request.sender_id, connection_ids, |conn_id| {
 402            self.peer.send(
 403                conn_id,
 404                proto::AddProjectCollaborator {
 405                    project_id,
 406                    collaborator: Some(proto::Collaborator {
 407                        peer_id: request.sender_id.0,
 408                        replica_id: response.replica_id,
 409                        user_id: user_id.to_proto(),
 410                    }),
 411                },
 412            )
 413        })?;
 414        self.update_contacts_for_users(&contact_user_ids)?;
 415        Ok(response)
 416    }
 417
 418    async fn leave_project(
 419        mut self: Arc<Server>,
 420        request: TypedEnvelope<proto::LeaveProject>,
 421    ) -> tide::Result<()> {
 422        let sender_id = request.sender_id;
 423        let project_id = request.payload.project_id;
 424        let worktree = self.state_mut().leave_project(sender_id, project_id)?;
 425
 426        broadcast(sender_id, worktree.connection_ids, |conn_id| {
 427            self.peer.send(
 428                conn_id,
 429                proto::RemoveProjectCollaborator {
 430                    project_id,
 431                    peer_id: sender_id.0,
 432                },
 433            )
 434        })?;
 435        self.update_contacts_for_users(&worktree.authorized_user_ids)?;
 436
 437        Ok(())
 438    }
 439
 440    async fn register_worktree(
 441        mut self: Arc<Server>,
 442        request: TypedEnvelope<proto::RegisterWorktree>,
 443    ) -> tide::Result<proto::Ack> {
 444        let host_user_id = self.state().user_id_for_connection(request.sender_id)?;
 445
 446        let mut contact_user_ids = HashSet::default();
 447        contact_user_ids.insert(host_user_id);
 448        for github_login in &request.payload.authorized_logins {
 449            let contact_user_id = self.app_state.db.create_user(github_login, false).await?;
 450            contact_user_ids.insert(contact_user_id);
 451        }
 452
 453        let contact_user_ids = contact_user_ids.into_iter().collect::<Vec<_>>();
 454        let guest_connection_ids;
 455        {
 456            let mut state = self.state_mut();
 457            guest_connection_ids = state
 458                .read_project(request.payload.project_id, request.sender_id)?
 459                .guest_connection_ids();
 460            state.register_worktree(
 461                request.payload.project_id,
 462                request.payload.worktree_id,
 463                request.sender_id,
 464                Worktree {
 465                    authorized_user_ids: contact_user_ids.clone(),
 466                    root_name: request.payload.root_name.clone(),
 467                    visible: request.payload.visible,
 468                },
 469            )?;
 470        }
 471        broadcast(request.sender_id, guest_connection_ids, |connection_id| {
 472            self.peer
 473                .forward_send(request.sender_id, connection_id, request.payload.clone())
 474        })?;
 475        self.update_contacts_for_users(&contact_user_ids)?;
 476        Ok(proto::Ack {})
 477    }
 478
 479    async fn unregister_worktree(
 480        mut self: Arc<Server>,
 481        request: TypedEnvelope<proto::UnregisterWorktree>,
 482    ) -> tide::Result<()> {
 483        let project_id = request.payload.project_id;
 484        let worktree_id = request.payload.worktree_id;
 485        let (worktree, guest_connection_ids) =
 486            self.state_mut()
 487                .unregister_worktree(project_id, worktree_id, request.sender_id)?;
 488        broadcast(request.sender_id, guest_connection_ids, |conn_id| {
 489            self.peer.send(
 490                conn_id,
 491                proto::UnregisterWorktree {
 492                    project_id,
 493                    worktree_id,
 494                },
 495            )
 496        })?;
 497        self.update_contacts_for_users(&worktree.authorized_user_ids)?;
 498        Ok(())
 499    }
 500
 501    async fn update_worktree(
 502        mut self: Arc<Server>,
 503        request: TypedEnvelope<proto::UpdateWorktree>,
 504    ) -> tide::Result<proto::Ack> {
 505        let connection_ids = self.state_mut().update_worktree(
 506            request.sender_id,
 507            request.payload.project_id,
 508            request.payload.worktree_id,
 509            &request.payload.removed_entries,
 510            &request.payload.updated_entries,
 511        )?;
 512
 513        broadcast(request.sender_id, connection_ids, |connection_id| {
 514            self.peer
 515                .forward_send(request.sender_id, connection_id, request.payload.clone())
 516        })?;
 517
 518        Ok(proto::Ack {})
 519    }
 520
 521    async fn update_diagnostic_summary(
 522        mut self: Arc<Server>,
 523        request: TypedEnvelope<proto::UpdateDiagnosticSummary>,
 524    ) -> tide::Result<()> {
 525        let summary = request
 526            .payload
 527            .summary
 528            .clone()
 529            .ok_or_else(|| anyhow!("invalid summary"))?;
 530        let receiver_ids = self.state_mut().update_diagnostic_summary(
 531            request.payload.project_id,
 532            request.payload.worktree_id,
 533            request.sender_id,
 534            summary,
 535        )?;
 536
 537        broadcast(request.sender_id, receiver_ids, |connection_id| {
 538            self.peer
 539                .forward_send(request.sender_id, connection_id, request.payload.clone())
 540        })?;
 541        Ok(())
 542    }
 543
 544    async fn start_language_server(
 545        mut self: Arc<Server>,
 546        request: TypedEnvelope<proto::StartLanguageServer>,
 547    ) -> tide::Result<()> {
 548        let receiver_ids = self.state_mut().start_language_server(
 549            request.payload.project_id,
 550            request.sender_id,
 551            request
 552                .payload
 553                .server
 554                .clone()
 555                .ok_or_else(|| anyhow!("invalid language server"))?,
 556        )?;
 557        broadcast(request.sender_id, receiver_ids, |connection_id| {
 558            self.peer
 559                .forward_send(request.sender_id, connection_id, request.payload.clone())
 560        })?;
 561        Ok(())
 562    }
 563
 564    async fn update_language_server(
 565        self: Arc<Server>,
 566        request: TypedEnvelope<proto::UpdateLanguageServer>,
 567    ) -> tide::Result<()> {
 568        let receiver_ids = self
 569            .state()
 570            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 571        broadcast(request.sender_id, receiver_ids, |connection_id| {
 572            self.peer
 573                .forward_send(request.sender_id, connection_id, request.payload.clone())
 574        })?;
 575        Ok(())
 576    }
 577
 578    async fn forward_project_request<T>(
 579        self: Arc<Server>,
 580        request: TypedEnvelope<T>,
 581    ) -> tide::Result<T::Response>
 582    where
 583        T: EntityMessage + RequestMessage,
 584    {
 585        let host_connection_id = self
 586            .state()
 587            .read_project(request.payload.remote_entity_id(), request.sender_id)?
 588            .host_connection_id;
 589        Ok(self
 590            .peer
 591            .forward_request(request.sender_id, host_connection_id, request.payload)
 592            .await?)
 593    }
 594
 595    async fn save_buffer(
 596        self: Arc<Server>,
 597        request: TypedEnvelope<proto::SaveBuffer>,
 598    ) -> tide::Result<proto::BufferSaved> {
 599        let host;
 600        let mut guests;
 601        {
 602            let state = self.state();
 603            let project = state.read_project(request.payload.project_id, request.sender_id)?;
 604            host = project.host_connection_id;
 605            guests = project.guest_connection_ids()
 606        }
 607
 608        let response = self
 609            .peer
 610            .forward_request(request.sender_id, host, request.payload.clone())
 611            .await?;
 612
 613        guests.retain(|guest_connection_id| *guest_connection_id != request.sender_id);
 614        broadcast(host, guests, |conn_id| {
 615            self.peer.forward_send(host, conn_id, response.clone())
 616        })?;
 617
 618        Ok(response)
 619    }
 620
 621    async fn update_buffer(
 622        self: Arc<Server>,
 623        request: TypedEnvelope<proto::UpdateBuffer>,
 624    ) -> tide::Result<proto::Ack> {
 625        let receiver_ids = self
 626            .state()
 627            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 628        broadcast(request.sender_id, receiver_ids, |connection_id| {
 629            self.peer
 630                .forward_send(request.sender_id, connection_id, request.payload.clone())
 631        })?;
 632        Ok(proto::Ack {})
 633    }
 634
 635    async fn update_buffer_file(
 636        self: Arc<Server>,
 637        request: TypedEnvelope<proto::UpdateBufferFile>,
 638    ) -> tide::Result<()> {
 639        let receiver_ids = self
 640            .state()
 641            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 642        broadcast(request.sender_id, receiver_ids, |connection_id| {
 643            self.peer
 644                .forward_send(request.sender_id, connection_id, request.payload.clone())
 645        })?;
 646        Ok(())
 647    }
 648
 649    async fn buffer_reloaded(
 650        self: Arc<Server>,
 651        request: TypedEnvelope<proto::BufferReloaded>,
 652    ) -> tide::Result<()> {
 653        let receiver_ids = self
 654            .state()
 655            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 656        broadcast(request.sender_id, receiver_ids, |connection_id| {
 657            self.peer
 658                .forward_send(request.sender_id, connection_id, request.payload.clone())
 659        })?;
 660        Ok(())
 661    }
 662
 663    async fn buffer_saved(
 664        self: Arc<Server>,
 665        request: TypedEnvelope<proto::BufferSaved>,
 666    ) -> tide::Result<()> {
 667        let receiver_ids = self
 668            .state()
 669            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 670        broadcast(request.sender_id, receiver_ids, |connection_id| {
 671            self.peer
 672                .forward_send(request.sender_id, connection_id, request.payload.clone())
 673        })?;
 674        Ok(())
 675    }
 676
 677    async fn follow(
 678        self: Arc<Self>,
 679        request: TypedEnvelope<proto::Follow>,
 680    ) -> tide::Result<proto::FollowResponse> {
 681        let leader_id = ConnectionId(request.payload.leader_id);
 682        let follower_id = request.sender_id;
 683        if !self
 684            .state()
 685            .project_connection_ids(request.payload.project_id, follower_id)?
 686            .contains(&leader_id)
 687        {
 688            Err(anyhow!("no such peer"))?;
 689        }
 690        let mut response = self
 691            .peer
 692            .forward_request(request.sender_id, leader_id, request.payload)
 693            .await?;
 694        response
 695            .views
 696            .retain(|view| view.leader_id != Some(follower_id.0));
 697        Ok(response)
 698    }
 699
 700    async fn unfollow(
 701        self: Arc<Self>,
 702        request: TypedEnvelope<proto::Unfollow>,
 703    ) -> tide::Result<()> {
 704        let leader_id = ConnectionId(request.payload.leader_id);
 705        if !self
 706            .state()
 707            .project_connection_ids(request.payload.project_id, request.sender_id)?
 708            .contains(&leader_id)
 709        {
 710            Err(anyhow!("no such peer"))?;
 711        }
 712        self.peer
 713            .forward_send(request.sender_id, leader_id, request.payload)?;
 714        Ok(())
 715    }
 716
 717    async fn update_followers(
 718        self: Arc<Self>,
 719        request: TypedEnvelope<proto::UpdateFollowers>,
 720    ) -> tide::Result<()> {
 721        let connection_ids = self
 722            .state()
 723            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 724        let leader_id = request
 725            .payload
 726            .variant
 727            .as_ref()
 728            .and_then(|variant| match variant {
 729                proto::update_followers::Variant::CreateView(payload) => payload.leader_id,
 730                proto::update_followers::Variant::UpdateView(payload) => payload.leader_id,
 731                proto::update_followers::Variant::UpdateActiveView(payload) => payload.leader_id,
 732            });
 733        for follower_id in &request.payload.follower_ids {
 734            let follower_id = ConnectionId(*follower_id);
 735            if connection_ids.contains(&follower_id) && Some(follower_id.0) != leader_id {
 736                self.peer
 737                    .forward_send(request.sender_id, follower_id, request.payload.clone())?;
 738            }
 739        }
 740        Ok(())
 741    }
 742
 743    async fn get_channels(
 744        self: Arc<Server>,
 745        request: TypedEnvelope<proto::GetChannels>,
 746    ) -> tide::Result<proto::GetChannelsResponse> {
 747        let user_id = self.state().user_id_for_connection(request.sender_id)?;
 748        let channels = self.app_state.db.get_accessible_channels(user_id).await?;
 749        Ok(proto::GetChannelsResponse {
 750            channels: channels
 751                .into_iter()
 752                .map(|chan| proto::Channel {
 753                    id: chan.id.to_proto(),
 754                    name: chan.name,
 755                })
 756                .collect(),
 757        })
 758    }
 759
 760    async fn get_users(
 761        self: Arc<Server>,
 762        request: TypedEnvelope<proto::GetUsers>,
 763    ) -> tide::Result<proto::GetUsersResponse> {
 764        let user_ids = request
 765            .payload
 766            .user_ids
 767            .into_iter()
 768            .map(UserId::from_proto)
 769            .collect();
 770        let users = self
 771            .app_state
 772            .db
 773            .get_users_by_ids(user_ids)
 774            .await?
 775            .into_iter()
 776            .map(|user| proto::User {
 777                id: user.id.to_proto(),
 778                avatar_url: format!("https://github.com/{}.png?size=128", user.github_login),
 779                github_login: user.github_login,
 780            })
 781            .collect();
 782        Ok(proto::GetUsersResponse { users })
 783    }
 784
 785    fn update_contacts_for_users<'a>(
 786        self: &Arc<Server>,
 787        user_ids: impl IntoIterator<Item = &'a UserId>,
 788    ) -> anyhow::Result<()> {
 789        let mut result = Ok(());
 790        let state = self.state();
 791        for user_id in user_ids {
 792            let contacts = state.contacts_for_user(*user_id);
 793            for connection_id in state.connection_ids_for_user(*user_id) {
 794                if let Err(error) = self.peer.send(
 795                    connection_id,
 796                    proto::UpdateContacts {
 797                        contacts: contacts.clone(),
 798                    },
 799                ) {
 800                    result = Err(error);
 801                }
 802            }
 803        }
 804        result
 805    }
 806
 807    async fn join_channel(
 808        mut self: Arc<Self>,
 809        request: TypedEnvelope<proto::JoinChannel>,
 810    ) -> tide::Result<proto::JoinChannelResponse> {
 811        let user_id = self.state().user_id_for_connection(request.sender_id)?;
 812        let channel_id = ChannelId::from_proto(request.payload.channel_id);
 813        if !self
 814            .app_state
 815            .db
 816            .can_user_access_channel(user_id, channel_id)
 817            .await?
 818        {
 819            Err(anyhow!("access denied"))?;
 820        }
 821
 822        self.state_mut().join_channel(request.sender_id, channel_id);
 823        let messages = self
 824            .app_state
 825            .db
 826            .get_channel_messages(channel_id, MESSAGE_COUNT_PER_PAGE, None)
 827            .await?
 828            .into_iter()
 829            .map(|msg| proto::ChannelMessage {
 830                id: msg.id.to_proto(),
 831                body: msg.body,
 832                timestamp: msg.sent_at.unix_timestamp() as u64,
 833                sender_id: msg.sender_id.to_proto(),
 834                nonce: Some(msg.nonce.as_u128().into()),
 835            })
 836            .collect::<Vec<_>>();
 837        Ok(proto::JoinChannelResponse {
 838            done: messages.len() < MESSAGE_COUNT_PER_PAGE,
 839            messages,
 840        })
 841    }
 842
 843    async fn leave_channel(
 844        mut self: Arc<Self>,
 845        request: TypedEnvelope<proto::LeaveChannel>,
 846    ) -> tide::Result<()> {
 847        let user_id = self.state().user_id_for_connection(request.sender_id)?;
 848        let channel_id = ChannelId::from_proto(request.payload.channel_id);
 849        if !self
 850            .app_state
 851            .db
 852            .can_user_access_channel(user_id, channel_id)
 853            .await?
 854        {
 855            Err(anyhow!("access denied"))?;
 856        }
 857
 858        self.state_mut()
 859            .leave_channel(request.sender_id, channel_id);
 860
 861        Ok(())
 862    }
 863
 864    async fn send_channel_message(
 865        self: Arc<Self>,
 866        request: TypedEnvelope<proto::SendChannelMessage>,
 867    ) -> tide::Result<proto::SendChannelMessageResponse> {
 868        let channel_id = ChannelId::from_proto(request.payload.channel_id);
 869        let user_id;
 870        let connection_ids;
 871        {
 872            let state = self.state();
 873            user_id = state.user_id_for_connection(request.sender_id)?;
 874            connection_ids = state.channel_connection_ids(channel_id)?;
 875        }
 876
 877        // Validate the message body.
 878        let body = request.payload.body.trim().to_string();
 879        if body.len() > MAX_MESSAGE_LEN {
 880            return Err(anyhow!("message is too long"))?;
 881        }
 882        if body.is_empty() {
 883            return Err(anyhow!("message can't be blank"))?;
 884        }
 885
 886        let timestamp = OffsetDateTime::now_utc();
 887        let nonce = request
 888            .payload
 889            .nonce
 890            .ok_or_else(|| anyhow!("nonce can't be blank"))?;
 891
 892        let message_id = self
 893            .app_state
 894            .db
 895            .create_channel_message(channel_id, user_id, &body, timestamp, nonce.clone().into())
 896            .await?
 897            .to_proto();
 898        let message = proto::ChannelMessage {
 899            sender_id: user_id.to_proto(),
 900            id: message_id,
 901            body,
 902            timestamp: timestamp.unix_timestamp() as u64,
 903            nonce: Some(nonce),
 904        };
 905        broadcast(request.sender_id, connection_ids, |conn_id| {
 906            self.peer.send(
 907                conn_id,
 908                proto::ChannelMessageSent {
 909                    channel_id: channel_id.to_proto(),
 910                    message: Some(message.clone()),
 911                },
 912            )
 913        })?;
 914        Ok(proto::SendChannelMessageResponse {
 915            message: Some(message),
 916        })
 917    }
 918
 919    async fn get_channel_messages(
 920        self: Arc<Self>,
 921        request: TypedEnvelope<proto::GetChannelMessages>,
 922    ) -> tide::Result<proto::GetChannelMessagesResponse> {
 923        let user_id = self.state().user_id_for_connection(request.sender_id)?;
 924        let channel_id = ChannelId::from_proto(request.payload.channel_id);
 925        if !self
 926            .app_state
 927            .db
 928            .can_user_access_channel(user_id, channel_id)
 929            .await?
 930        {
 931            Err(anyhow!("access denied"))?;
 932        }
 933
 934        let messages = self
 935            .app_state
 936            .db
 937            .get_channel_messages(
 938                channel_id,
 939                MESSAGE_COUNT_PER_PAGE,
 940                Some(MessageId::from_proto(request.payload.before_message_id)),
 941            )
 942            .await?
 943            .into_iter()
 944            .map(|msg| proto::ChannelMessage {
 945                id: msg.id.to_proto(),
 946                body: msg.body,
 947                timestamp: msg.sent_at.unix_timestamp() as u64,
 948                sender_id: msg.sender_id.to_proto(),
 949                nonce: Some(msg.nonce.as_u128().into()),
 950            })
 951            .collect::<Vec<_>>();
 952
 953        Ok(proto::GetChannelMessagesResponse {
 954            done: messages.len() < MESSAGE_COUNT_PER_PAGE,
 955            messages,
 956        })
 957    }
 958
 959    fn state<'a>(self: &'a Arc<Self>) -> RwLockReadGuard<'a, Store> {
 960        self.store.read()
 961    }
 962
 963    fn state_mut<'a>(self: &'a mut Arc<Self>) -> RwLockWriteGuard<'a, Store> {
 964        self.store.write()
 965    }
 966}
 967
 968impl Executor for RealExecutor {
 969    type Timer = Timer;
 970
 971    fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F) {
 972        task::spawn(future);
 973    }
 974
 975    fn timer(&self, duration: Duration) -> Self::Timer {
 976        Timer::after(duration)
 977    }
 978}
 979
 980fn broadcast<F>(
 981    sender_id: ConnectionId,
 982    receiver_ids: Vec<ConnectionId>,
 983    mut f: F,
 984) -> anyhow::Result<()>
 985where
 986    F: FnMut(ConnectionId) -> anyhow::Result<()>,
 987{
 988    let mut result = Ok(());
 989    for receiver_id in receiver_ids {
 990        if receiver_id != sender_id {
 991            if let Err(error) = f(receiver_id) {
 992                if result.is_ok() {
 993                    result = Err(error);
 994                }
 995            }
 996        }
 997    }
 998    result
 999}
1000
1001pub fn add_routes(app: &mut tide::Server<Arc<AppState>>, rpc: &Arc<Peer>) {
1002    let server = Server::new(app.state().clone(), rpc.clone(), None);
1003    app.at("/rpc").get(move |request: Request<Arc<AppState>>| {
1004        let server = server.clone();
1005        async move {
1006            const WEBSOCKET_GUID: &str = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
1007
1008            let connection_upgrade = header_contains_ignore_case(&request, CONNECTION, "upgrade");
1009            let upgrade_to_websocket = header_contains_ignore_case(&request, UPGRADE, "websocket");
1010            let upgrade_requested = connection_upgrade && upgrade_to_websocket;
1011            let client_protocol_version: Option<u32> = request
1012                .header("X-Zed-Protocol-Version")
1013                .and_then(|v| v.as_str().parse().ok());
1014
1015            if !upgrade_requested || client_protocol_version != Some(rpc::PROTOCOL_VERSION) {
1016                return Ok(Response::new(StatusCode::UpgradeRequired));
1017            }
1018
1019            let header = match request.header("Sec-Websocket-Key") {
1020                Some(h) => h.as_str(),
1021                None => return Err(anyhow!("expected sec-websocket-key"))?,
1022            };
1023
1024            let user_id = process_auth_header(&request).await?;
1025
1026            let mut response = Response::new(StatusCode::SwitchingProtocols);
1027            response.insert_header(UPGRADE, "websocket");
1028            response.insert_header(CONNECTION, "Upgrade");
1029            let hash = Sha1::new().chain(header).chain(WEBSOCKET_GUID).finalize();
1030            response.insert_header("Sec-Websocket-Accept", base64::encode(&hash[..]));
1031            response.insert_header("Sec-Websocket-Version", "13");
1032
1033            let http_res: &mut tide::http::Response = response.as_mut();
1034            let upgrade_receiver = http_res.recv_upgrade().await;
1035            let addr = request.remote().unwrap_or("unknown").to_string();
1036            task::spawn(async move {
1037                if let Some(stream) = upgrade_receiver.await {
1038                    server
1039                        .handle_connection(
1040                            Connection::new(
1041                                WebSocketStream::from_raw_socket(stream, Role::Server, None).await,
1042                            ),
1043                            addr,
1044                            user_id,
1045                            None,
1046                            RealExecutor,
1047                        )
1048                        .await;
1049                }
1050            });
1051
1052            Ok(response)
1053        }
1054    });
1055}
1056
1057fn header_contains_ignore_case<T>(
1058    request: &tide::Request<T>,
1059    header_name: HeaderName,
1060    value: &str,
1061) -> bool {
1062    request
1063        .header(header_name)
1064        .map(|h| {
1065            h.as_str()
1066                .split(',')
1067                .any(|s| s.trim().eq_ignore_ascii_case(value.trim()))
1068        })
1069        .unwrap_or(false)
1070}
1071
1072#[cfg(test)]
1073mod tests {
1074    use super::*;
1075    use crate::{
1076        auth,
1077        db::{tests::TestDb, UserId},
1078        github, AppState, Config,
1079    };
1080    use ::rpc::Peer;
1081    use client::{
1082        self, test::FakeHttpClient, Channel, ChannelDetails, ChannelList, Client, Credentials,
1083        EstablishConnectionError, UserStore, RECEIVE_TIMEOUT,
1084    };
1085    use collections::BTreeMap;
1086    use editor::{
1087        self, ConfirmCodeAction, ConfirmCompletion, ConfirmRename, Editor, Input, Redo, Rename,
1088        ToOffset, ToggleCodeActions, Undo,
1089    };
1090    use gpui::{
1091        executor::{self, Deterministic},
1092        geometry::vector::vec2f,
1093        ModelHandle, TestAppContext, ViewHandle,
1094    };
1095    use language::{
1096        range_to_lsp, tree_sitter_rust, Diagnostic, DiagnosticEntry, FakeLspAdapter, Language,
1097        LanguageConfig, LanguageRegistry, OffsetRangeExt, Point, Rope,
1098    };
1099    use lsp::{self, FakeLanguageServer};
1100    use parking_lot::Mutex;
1101    use project::{
1102        fs::{FakeFs, Fs as _},
1103        search::SearchQuery,
1104        worktree::WorktreeHandle,
1105        DiagnosticSummary, Project, ProjectPath, WorktreeId,
1106    };
1107    use rand::prelude::*;
1108    use rpc::PeerId;
1109    use serde_json::json;
1110    use settings::Settings;
1111    use sqlx::types::time::OffsetDateTime;
1112    use std::{
1113        env,
1114        ops::Deref,
1115        path::{Path, PathBuf},
1116        rc::Rc,
1117        sync::{
1118            atomic::{AtomicBool, Ordering::SeqCst},
1119            Arc,
1120        },
1121        time::Duration,
1122    };
1123    use util::TryFutureExt;
1124    use workspace::{Item, SplitDirection, ToggleFollow, Workspace, WorkspaceParams};
1125
1126    #[cfg(test)]
1127    #[ctor::ctor]
1128    fn init_logger() {
1129        if std::env::var("RUST_LOG").is_ok() {
1130            env_logger::init();
1131        }
1132    }
1133
1134    #[gpui::test(iterations = 10)]
1135    async fn test_share_project(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
1136        let (window_b, _) = cx_b.add_window(|_| EmptyView);
1137        let lang_registry = Arc::new(LanguageRegistry::test());
1138        let fs = FakeFs::new(cx_a.background());
1139        cx_a.foreground().forbid_parking();
1140
1141        // Connect to a server as 2 clients.
1142        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1143        let client_a = server.create_client(cx_a, "user_a").await;
1144        let client_b = server.create_client(cx_b, "user_b").await;
1145
1146        // Share a project as client A
1147        fs.insert_tree(
1148            "/a",
1149            json!({
1150                ".zed.toml": r#"collaborators = ["user_b"]"#,
1151                "a.txt": "a-contents",
1152                "b.txt": "b-contents",
1153            }),
1154        )
1155        .await;
1156        let project_a = cx_a.update(|cx| {
1157            Project::local(
1158                client_a.clone(),
1159                client_a.user_store.clone(),
1160                lang_registry.clone(),
1161                fs.clone(),
1162                cx,
1163            )
1164        });
1165        let (worktree_a, _) = project_a
1166            .update(cx_a, |p, cx| {
1167                p.find_or_create_local_worktree("/a", true, cx)
1168            })
1169            .await
1170            .unwrap();
1171        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1172        worktree_a
1173            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1174            .await;
1175        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
1176        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
1177
1178        // Join that project as client B
1179        let project_b = Project::remote(
1180            project_id,
1181            client_b.clone(),
1182            client_b.user_store.clone(),
1183            lang_registry.clone(),
1184            fs.clone(),
1185            &mut cx_b.to_async(),
1186        )
1187        .await
1188        .unwrap();
1189
1190        let replica_id_b = project_b.read_with(cx_b, |project, _| {
1191            assert_eq!(
1192                project
1193                    .collaborators()
1194                    .get(&client_a.peer_id)
1195                    .unwrap()
1196                    .user
1197                    .github_login,
1198                "user_a"
1199            );
1200            project.replica_id()
1201        });
1202        project_a
1203            .condition(&cx_a, |tree, _| {
1204                tree.collaborators()
1205                    .get(&client_b.peer_id)
1206                    .map_or(false, |collaborator| {
1207                        collaborator.replica_id == replica_id_b
1208                            && collaborator.user.github_login == "user_b"
1209                    })
1210            })
1211            .await;
1212
1213        // Open the same file as client B and client A.
1214        let buffer_b = project_b
1215            .update(cx_b, |p, cx| p.open_buffer((worktree_id, "b.txt"), cx))
1216            .await
1217            .unwrap();
1218        buffer_b.read_with(cx_b, |buf, _| assert_eq!(buf.text(), "b-contents"));
1219        project_a.read_with(cx_a, |project, cx| {
1220            assert!(project.has_open_buffer((worktree_id, "b.txt"), cx))
1221        });
1222        let buffer_a = project_a
1223            .update(cx_a, |p, cx| p.open_buffer((worktree_id, "b.txt"), cx))
1224            .await
1225            .unwrap();
1226
1227        let editor_b = cx_b.add_view(window_b, |cx| Editor::for_buffer(buffer_b, None, cx));
1228
1229        // TODO
1230        // // Create a selection set as client B and see that selection set as client A.
1231        // buffer_a
1232        //     .condition(&cx_a, |buffer, _| buffer.selection_sets().count() == 1)
1233        //     .await;
1234
1235        // Edit the buffer as client B and see that edit as client A.
1236        editor_b.update(cx_b, |editor, cx| {
1237            editor.handle_input(&Input("ok, ".into()), cx)
1238        });
1239        buffer_a
1240            .condition(&cx_a, |buffer, _| buffer.text() == "ok, b-contents")
1241            .await;
1242
1243        // TODO
1244        // // Remove the selection set as client B, see those selections disappear as client A.
1245        cx_b.update(move |_| drop(editor_b));
1246        // buffer_a
1247        //     .condition(&cx_a, |buffer, _| buffer.selection_sets().count() == 0)
1248        //     .await;
1249
1250        // Dropping the client B's project removes client B from client A's collaborators.
1251        cx_b.update(move |_| drop(project_b));
1252        project_a
1253            .condition(&cx_a, |project, _| project.collaborators().is_empty())
1254            .await;
1255    }
1256
1257    #[gpui::test(iterations = 10)]
1258    async fn test_unshare_project(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
1259        let lang_registry = Arc::new(LanguageRegistry::test());
1260        let fs = FakeFs::new(cx_a.background());
1261        cx_a.foreground().forbid_parking();
1262
1263        // Connect to a server as 2 clients.
1264        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1265        let client_a = server.create_client(cx_a, "user_a").await;
1266        let client_b = server.create_client(cx_b, "user_b").await;
1267
1268        // Share a project as client A
1269        fs.insert_tree(
1270            "/a",
1271            json!({
1272                ".zed.toml": r#"collaborators = ["user_b"]"#,
1273                "a.txt": "a-contents",
1274                "b.txt": "b-contents",
1275            }),
1276        )
1277        .await;
1278        let project_a = cx_a.update(|cx| {
1279            Project::local(
1280                client_a.clone(),
1281                client_a.user_store.clone(),
1282                lang_registry.clone(),
1283                fs.clone(),
1284                cx,
1285            )
1286        });
1287        let (worktree_a, _) = project_a
1288            .update(cx_a, |p, cx| {
1289                p.find_or_create_local_worktree("/a", true, cx)
1290            })
1291            .await
1292            .unwrap();
1293        worktree_a
1294            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1295            .await;
1296        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
1297        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1298        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
1299        assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1300
1301        // Join that project as client B
1302        let project_b = Project::remote(
1303            project_id,
1304            client_b.clone(),
1305            client_b.user_store.clone(),
1306            lang_registry.clone(),
1307            fs.clone(),
1308            &mut cx_b.to_async(),
1309        )
1310        .await
1311        .unwrap();
1312        project_b
1313            .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1314            .await
1315            .unwrap();
1316
1317        // Unshare the project as client A
1318        project_a.update(cx_a, |project, cx| project.unshare(cx));
1319        project_b
1320            .condition(cx_b, |project, _| project.is_read_only())
1321            .await;
1322        assert!(worktree_a.read_with(cx_a, |tree, _| !tree.as_local().unwrap().is_shared()));
1323        cx_b.update(|_| {
1324            drop(project_b);
1325        });
1326
1327        // Share the project again and ensure guests can still join.
1328        project_a
1329            .update(cx_a, |project, cx| project.share(cx))
1330            .await
1331            .unwrap();
1332        assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1333
1334        let project_b2 = Project::remote(
1335            project_id,
1336            client_b.clone(),
1337            client_b.user_store.clone(),
1338            lang_registry.clone(),
1339            fs.clone(),
1340            &mut cx_b.to_async(),
1341        )
1342        .await
1343        .unwrap();
1344        project_b2
1345            .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1346            .await
1347            .unwrap();
1348    }
1349
1350    #[gpui::test(iterations = 10)]
1351    async fn test_host_disconnect(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
1352        let lang_registry = Arc::new(LanguageRegistry::test());
1353        let fs = FakeFs::new(cx_a.background());
1354        cx_a.foreground().forbid_parking();
1355
1356        // Connect to a server as 2 clients.
1357        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1358        let client_a = server.create_client(cx_a, "user_a").await;
1359        let client_b = server.create_client(cx_b, "user_b").await;
1360
1361        // Share a project as client A
1362        fs.insert_tree(
1363            "/a",
1364            json!({
1365                ".zed.toml": r#"collaborators = ["user_b"]"#,
1366                "a.txt": "a-contents",
1367                "b.txt": "b-contents",
1368            }),
1369        )
1370        .await;
1371        let project_a = cx_a.update(|cx| {
1372            Project::local(
1373                client_a.clone(),
1374                client_a.user_store.clone(),
1375                lang_registry.clone(),
1376                fs.clone(),
1377                cx,
1378            )
1379        });
1380        let (worktree_a, _) = project_a
1381            .update(cx_a, |p, cx| {
1382                p.find_or_create_local_worktree("/a", true, cx)
1383            })
1384            .await
1385            .unwrap();
1386        worktree_a
1387            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1388            .await;
1389        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
1390        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1391        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
1392        assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1393
1394        // Join that project as client B
1395        let project_b = Project::remote(
1396            project_id,
1397            client_b.clone(),
1398            client_b.user_store.clone(),
1399            lang_registry.clone(),
1400            fs.clone(),
1401            &mut cx_b.to_async(),
1402        )
1403        .await
1404        .unwrap();
1405        project_b
1406            .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1407            .await
1408            .unwrap();
1409
1410        // Drop client A's connection. Collaborators should disappear and the project should not be shown as shared.
1411        server.disconnect_client(client_a.current_user_id(cx_a));
1412        cx_a.foreground().advance_clock(rpc::RECEIVE_TIMEOUT);
1413        project_a
1414            .condition(cx_a, |project, _| project.collaborators().is_empty())
1415            .await;
1416        project_a.read_with(cx_a, |project, _| assert!(!project.is_shared()));
1417        project_b
1418            .condition(cx_b, |project, _| project.is_read_only())
1419            .await;
1420        assert!(worktree_a.read_with(cx_a, |tree, _| !tree.as_local().unwrap().is_shared()));
1421        cx_b.update(|_| {
1422            drop(project_b);
1423        });
1424
1425        // Await reconnection
1426        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
1427
1428        // Share the project again and ensure guests can still join.
1429        project_a
1430            .update(cx_a, |project, cx| project.share(cx))
1431            .await
1432            .unwrap();
1433        assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1434
1435        let project_b2 = Project::remote(
1436            project_id,
1437            client_b.clone(),
1438            client_b.user_store.clone(),
1439            lang_registry.clone(),
1440            fs.clone(),
1441            &mut cx_b.to_async(),
1442        )
1443        .await
1444        .unwrap();
1445        project_b2
1446            .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1447            .await
1448            .unwrap();
1449    }
1450
1451    #[gpui::test(iterations = 10)]
1452    async fn test_propagate_saves_and_fs_changes(
1453        cx_a: &mut TestAppContext,
1454        cx_b: &mut TestAppContext,
1455        cx_c: &mut TestAppContext,
1456    ) {
1457        let lang_registry = Arc::new(LanguageRegistry::test());
1458        let fs = FakeFs::new(cx_a.background());
1459        cx_a.foreground().forbid_parking();
1460
1461        // Connect to a server as 3 clients.
1462        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1463        let client_a = server.create_client(cx_a, "user_a").await;
1464        let client_b = server.create_client(cx_b, "user_b").await;
1465        let client_c = server.create_client(cx_c, "user_c").await;
1466
1467        // Share a worktree as client A.
1468        fs.insert_tree(
1469            "/a",
1470            json!({
1471                ".zed.toml": r#"collaborators = ["user_b", "user_c"]"#,
1472                "file1": "",
1473                "file2": ""
1474            }),
1475        )
1476        .await;
1477        let project_a = cx_a.update(|cx| {
1478            Project::local(
1479                client_a.clone(),
1480                client_a.user_store.clone(),
1481                lang_registry.clone(),
1482                fs.clone(),
1483                cx,
1484            )
1485        });
1486        let (worktree_a, _) = project_a
1487            .update(cx_a, |p, cx| {
1488                p.find_or_create_local_worktree("/a", true, cx)
1489            })
1490            .await
1491            .unwrap();
1492        worktree_a
1493            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1494            .await;
1495        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
1496        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1497        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
1498
1499        // Join that worktree as clients B and C.
1500        let project_b = Project::remote(
1501            project_id,
1502            client_b.clone(),
1503            client_b.user_store.clone(),
1504            lang_registry.clone(),
1505            fs.clone(),
1506            &mut cx_b.to_async(),
1507        )
1508        .await
1509        .unwrap();
1510        let project_c = Project::remote(
1511            project_id,
1512            client_c.clone(),
1513            client_c.user_store.clone(),
1514            lang_registry.clone(),
1515            fs.clone(),
1516            &mut cx_c.to_async(),
1517        )
1518        .await
1519        .unwrap();
1520        let worktree_b = project_b.read_with(cx_b, |p, cx| p.worktrees(cx).next().unwrap());
1521        let worktree_c = project_c.read_with(cx_c, |p, cx| p.worktrees(cx).next().unwrap());
1522
1523        // Open and edit a buffer as both guests B and C.
1524        let buffer_b = project_b
1525            .update(cx_b, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
1526            .await
1527            .unwrap();
1528        let buffer_c = project_c
1529            .update(cx_c, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
1530            .await
1531            .unwrap();
1532        buffer_b.update(cx_b, |buf, cx| buf.edit([0..0], "i-am-b, ", cx));
1533        buffer_c.update(cx_c, |buf, cx| buf.edit([0..0], "i-am-c, ", cx));
1534
1535        // Open and edit that buffer as the host.
1536        let buffer_a = project_a
1537            .update(cx_a, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
1538            .await
1539            .unwrap();
1540
1541        buffer_a
1542            .condition(cx_a, |buf, _| buf.text() == "i-am-c, i-am-b, ")
1543            .await;
1544        buffer_a.update(cx_a, |buf, cx| {
1545            buf.edit([buf.len()..buf.len()], "i-am-a", cx)
1546        });
1547
1548        // Wait for edits to propagate
1549        buffer_a
1550            .condition(cx_a, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
1551            .await;
1552        buffer_b
1553            .condition(cx_b, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
1554            .await;
1555        buffer_c
1556            .condition(cx_c, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
1557            .await;
1558
1559        // Edit the buffer as the host and concurrently save as guest B.
1560        let save_b = buffer_b.update(cx_b, |buf, cx| buf.save(cx));
1561        buffer_a.update(cx_a, |buf, cx| buf.edit([0..0], "hi-a, ", cx));
1562        save_b.await.unwrap();
1563        assert_eq!(
1564            fs.load("/a/file1".as_ref()).await.unwrap(),
1565            "hi-a, i-am-c, i-am-b, i-am-a"
1566        );
1567        buffer_a.read_with(cx_a, |buf, _| assert!(!buf.is_dirty()));
1568        buffer_b.read_with(cx_b, |buf, _| assert!(!buf.is_dirty()));
1569        buffer_c.condition(cx_c, |buf, _| !buf.is_dirty()).await;
1570
1571        worktree_a.flush_fs_events(cx_a).await;
1572
1573        // Make changes on host's file system, see those changes on guest worktrees.
1574        fs.rename(
1575            "/a/file1".as_ref(),
1576            "/a/file1-renamed".as_ref(),
1577            Default::default(),
1578        )
1579        .await
1580        .unwrap();
1581
1582        fs.rename("/a/file2".as_ref(), "/a/file3".as_ref(), Default::default())
1583            .await
1584            .unwrap();
1585        fs.insert_file(Path::new("/a/file4"), "4".into()).await;
1586
1587        worktree_a
1588            .condition(&cx_a, |tree, _| {
1589                tree.paths()
1590                    .map(|p| p.to_string_lossy())
1591                    .collect::<Vec<_>>()
1592                    == [".zed.toml", "file1-renamed", "file3", "file4"]
1593            })
1594            .await;
1595        worktree_b
1596            .condition(&cx_b, |tree, _| {
1597                tree.paths()
1598                    .map(|p| p.to_string_lossy())
1599                    .collect::<Vec<_>>()
1600                    == [".zed.toml", "file1-renamed", "file3", "file4"]
1601            })
1602            .await;
1603        worktree_c
1604            .condition(&cx_c, |tree, _| {
1605                tree.paths()
1606                    .map(|p| p.to_string_lossy())
1607                    .collect::<Vec<_>>()
1608                    == [".zed.toml", "file1-renamed", "file3", "file4"]
1609            })
1610            .await;
1611
1612        // Ensure buffer files are updated as well.
1613        buffer_a
1614            .condition(&cx_a, |buf, _| {
1615                buf.file().unwrap().path().to_str() == Some("file1-renamed")
1616            })
1617            .await;
1618        buffer_b
1619            .condition(&cx_b, |buf, _| {
1620                buf.file().unwrap().path().to_str() == Some("file1-renamed")
1621            })
1622            .await;
1623        buffer_c
1624            .condition(&cx_c, |buf, _| {
1625                buf.file().unwrap().path().to_str() == Some("file1-renamed")
1626            })
1627            .await;
1628    }
1629
1630    #[gpui::test(iterations = 10)]
1631    async fn test_buffer_conflict_after_save(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
1632        cx_a.foreground().forbid_parking();
1633        let lang_registry = Arc::new(LanguageRegistry::test());
1634        let fs = FakeFs::new(cx_a.background());
1635
1636        // Connect to a server as 2 clients.
1637        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1638        let client_a = server.create_client(cx_a, "user_a").await;
1639        let client_b = server.create_client(cx_b, "user_b").await;
1640
1641        // Share a project as client A
1642        fs.insert_tree(
1643            "/dir",
1644            json!({
1645                ".zed.toml": r#"collaborators = ["user_b", "user_c"]"#,
1646                "a.txt": "a-contents",
1647            }),
1648        )
1649        .await;
1650
1651        let project_a = cx_a.update(|cx| {
1652            Project::local(
1653                client_a.clone(),
1654                client_a.user_store.clone(),
1655                lang_registry.clone(),
1656                fs.clone(),
1657                cx,
1658            )
1659        });
1660        let (worktree_a, _) = project_a
1661            .update(cx_a, |p, cx| {
1662                p.find_or_create_local_worktree("/dir", true, cx)
1663            })
1664            .await
1665            .unwrap();
1666        worktree_a
1667            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1668            .await;
1669        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
1670        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1671        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
1672
1673        // Join that project as client B
1674        let project_b = Project::remote(
1675            project_id,
1676            client_b.clone(),
1677            client_b.user_store.clone(),
1678            lang_registry.clone(),
1679            fs.clone(),
1680            &mut cx_b.to_async(),
1681        )
1682        .await
1683        .unwrap();
1684
1685        // Open a buffer as client B
1686        let buffer_b = project_b
1687            .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1688            .await
1689            .unwrap();
1690
1691        buffer_b.update(cx_b, |buf, cx| buf.edit([0..0], "world ", cx));
1692        buffer_b.read_with(cx_b, |buf, _| {
1693            assert!(buf.is_dirty());
1694            assert!(!buf.has_conflict());
1695        });
1696
1697        buffer_b.update(cx_b, |buf, cx| buf.save(cx)).await.unwrap();
1698        buffer_b
1699            .condition(&cx_b, |buffer_b, _| !buffer_b.is_dirty())
1700            .await;
1701        buffer_b.read_with(cx_b, |buf, _| {
1702            assert!(!buf.has_conflict());
1703        });
1704
1705        buffer_b.update(cx_b, |buf, cx| buf.edit([0..0], "hello ", cx));
1706        buffer_b.read_with(cx_b, |buf, _| {
1707            assert!(buf.is_dirty());
1708            assert!(!buf.has_conflict());
1709        });
1710    }
1711
1712    #[gpui::test(iterations = 10)]
1713    async fn test_buffer_reloading(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
1714        cx_a.foreground().forbid_parking();
1715        let lang_registry = Arc::new(LanguageRegistry::test());
1716        let fs = FakeFs::new(cx_a.background());
1717
1718        // Connect to a server as 2 clients.
1719        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1720        let client_a = server.create_client(cx_a, "user_a").await;
1721        let client_b = server.create_client(cx_b, "user_b").await;
1722
1723        // Share a project as client A
1724        fs.insert_tree(
1725            "/dir",
1726            json!({
1727                ".zed.toml": r#"collaborators = ["user_b", "user_c"]"#,
1728                "a.txt": "a-contents",
1729            }),
1730        )
1731        .await;
1732
1733        let project_a = cx_a.update(|cx| {
1734            Project::local(
1735                client_a.clone(),
1736                client_a.user_store.clone(),
1737                lang_registry.clone(),
1738                fs.clone(),
1739                cx,
1740            )
1741        });
1742        let (worktree_a, _) = project_a
1743            .update(cx_a, |p, cx| {
1744                p.find_or_create_local_worktree("/dir", true, cx)
1745            })
1746            .await
1747            .unwrap();
1748        worktree_a
1749            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1750            .await;
1751        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
1752        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1753        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
1754
1755        // Join that project as client B
1756        let project_b = Project::remote(
1757            project_id,
1758            client_b.clone(),
1759            client_b.user_store.clone(),
1760            lang_registry.clone(),
1761            fs.clone(),
1762            &mut cx_b.to_async(),
1763        )
1764        .await
1765        .unwrap();
1766        let _worktree_b = project_b.update(cx_b, |p, cx| p.worktrees(cx).next().unwrap());
1767
1768        // Open a buffer as client B
1769        let buffer_b = project_b
1770            .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1771            .await
1772            .unwrap();
1773        buffer_b.read_with(cx_b, |buf, _| {
1774            assert!(!buf.is_dirty());
1775            assert!(!buf.has_conflict());
1776        });
1777
1778        fs.save(Path::new("/dir/a.txt"), &"new contents".into())
1779            .await
1780            .unwrap();
1781        buffer_b
1782            .condition(&cx_b, |buf, _| {
1783                buf.text() == "new contents" && !buf.is_dirty()
1784            })
1785            .await;
1786        buffer_b.read_with(cx_b, |buf, _| {
1787            assert!(!buf.has_conflict());
1788        });
1789    }
1790
1791    #[gpui::test(iterations = 10)]
1792    async fn test_editing_while_guest_opens_buffer(
1793        cx_a: &mut TestAppContext,
1794        cx_b: &mut TestAppContext,
1795    ) {
1796        cx_a.foreground().forbid_parking();
1797        let lang_registry = Arc::new(LanguageRegistry::test());
1798        let fs = FakeFs::new(cx_a.background());
1799
1800        // Connect to a server as 2 clients.
1801        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1802        let client_a = server.create_client(cx_a, "user_a").await;
1803        let client_b = server.create_client(cx_b, "user_b").await;
1804
1805        // Share a project as client A
1806        fs.insert_tree(
1807            "/dir",
1808            json!({
1809                ".zed.toml": r#"collaborators = ["user_b"]"#,
1810                "a.txt": "a-contents",
1811            }),
1812        )
1813        .await;
1814        let project_a = cx_a.update(|cx| {
1815            Project::local(
1816                client_a.clone(),
1817                client_a.user_store.clone(),
1818                lang_registry.clone(),
1819                fs.clone(),
1820                cx,
1821            )
1822        });
1823        let (worktree_a, _) = project_a
1824            .update(cx_a, |p, cx| {
1825                p.find_or_create_local_worktree("/dir", true, cx)
1826            })
1827            .await
1828            .unwrap();
1829        worktree_a
1830            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1831            .await;
1832        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
1833        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1834        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
1835
1836        // Join that project as client B
1837        let project_b = Project::remote(
1838            project_id,
1839            client_b.clone(),
1840            client_b.user_store.clone(),
1841            lang_registry.clone(),
1842            fs.clone(),
1843            &mut cx_b.to_async(),
1844        )
1845        .await
1846        .unwrap();
1847
1848        // Open a buffer as client A
1849        let buffer_a = project_a
1850            .update(cx_a, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1851            .await
1852            .unwrap();
1853
1854        // Start opening the same buffer as client B
1855        let buffer_b = cx_b
1856            .background()
1857            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx)));
1858
1859        // Edit the buffer as client A while client B is still opening it.
1860        cx_b.background().simulate_random_delay().await;
1861        buffer_a.update(cx_a, |buf, cx| buf.edit([0..0], "X", cx));
1862        cx_b.background().simulate_random_delay().await;
1863        buffer_a.update(cx_a, |buf, cx| buf.edit([1..1], "Y", cx));
1864
1865        let text = buffer_a.read_with(cx_a, |buf, _| buf.text());
1866        let buffer_b = buffer_b.await.unwrap();
1867        buffer_b.condition(&cx_b, |buf, _| buf.text() == text).await;
1868    }
1869
1870    #[gpui::test(iterations = 10)]
1871    async fn test_leaving_worktree_while_opening_buffer(
1872        cx_a: &mut TestAppContext,
1873        cx_b: &mut TestAppContext,
1874    ) {
1875        cx_a.foreground().forbid_parking();
1876        let lang_registry = Arc::new(LanguageRegistry::test());
1877        let fs = FakeFs::new(cx_a.background());
1878
1879        // Connect to a server as 2 clients.
1880        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1881        let client_a = server.create_client(cx_a, "user_a").await;
1882        let client_b = server.create_client(cx_b, "user_b").await;
1883
1884        // Share a project as client A
1885        fs.insert_tree(
1886            "/dir",
1887            json!({
1888                ".zed.toml": r#"collaborators = ["user_b"]"#,
1889                "a.txt": "a-contents",
1890            }),
1891        )
1892        .await;
1893        let project_a = cx_a.update(|cx| {
1894            Project::local(
1895                client_a.clone(),
1896                client_a.user_store.clone(),
1897                lang_registry.clone(),
1898                fs.clone(),
1899                cx,
1900            )
1901        });
1902        let (worktree_a, _) = project_a
1903            .update(cx_a, |p, cx| {
1904                p.find_or_create_local_worktree("/dir", true, cx)
1905            })
1906            .await
1907            .unwrap();
1908        worktree_a
1909            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1910            .await;
1911        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
1912        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1913        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
1914
1915        // Join that project as client B
1916        let project_b = Project::remote(
1917            project_id,
1918            client_b.clone(),
1919            client_b.user_store.clone(),
1920            lang_registry.clone(),
1921            fs.clone(),
1922            &mut cx_b.to_async(),
1923        )
1924        .await
1925        .unwrap();
1926
1927        // See that a guest has joined as client A.
1928        project_a
1929            .condition(&cx_a, |p, _| p.collaborators().len() == 1)
1930            .await;
1931
1932        // Begin opening a buffer as client B, but leave the project before the open completes.
1933        let buffer_b = cx_b
1934            .background()
1935            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx)));
1936        cx_b.update(|_| drop(project_b));
1937        drop(buffer_b);
1938
1939        // See that the guest has left.
1940        project_a
1941            .condition(&cx_a, |p, _| p.collaborators().len() == 0)
1942            .await;
1943    }
1944
1945    #[gpui::test(iterations = 10)]
1946    async fn test_leaving_project(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
1947        cx_a.foreground().forbid_parking();
1948        let lang_registry = Arc::new(LanguageRegistry::test());
1949        let fs = FakeFs::new(cx_a.background());
1950
1951        // Connect to a server as 2 clients.
1952        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1953        let client_a = server.create_client(cx_a, "user_a").await;
1954        let client_b = server.create_client(cx_b, "user_b").await;
1955
1956        // Share a project as client A
1957        fs.insert_tree(
1958            "/a",
1959            json!({
1960                ".zed.toml": r#"collaborators = ["user_b"]"#,
1961                "a.txt": "a-contents",
1962                "b.txt": "b-contents",
1963            }),
1964        )
1965        .await;
1966        let project_a = cx_a.update(|cx| {
1967            Project::local(
1968                client_a.clone(),
1969                client_a.user_store.clone(),
1970                lang_registry.clone(),
1971                fs.clone(),
1972                cx,
1973            )
1974        });
1975        let (worktree_a, _) = project_a
1976            .update(cx_a, |p, cx| {
1977                p.find_or_create_local_worktree("/a", true, cx)
1978            })
1979            .await
1980            .unwrap();
1981        worktree_a
1982            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1983            .await;
1984        let project_id = project_a
1985            .update(cx_a, |project, _| project.next_remote_id())
1986            .await;
1987        project_a
1988            .update(cx_a, |project, cx| project.share(cx))
1989            .await
1990            .unwrap();
1991
1992        // Join that project as client B
1993        let _project_b = Project::remote(
1994            project_id,
1995            client_b.clone(),
1996            client_b.user_store.clone(),
1997            lang_registry.clone(),
1998            fs.clone(),
1999            &mut cx_b.to_async(),
2000        )
2001        .await
2002        .unwrap();
2003
2004        // Client A sees that a guest has joined.
2005        project_a
2006            .condition(cx_a, |p, _| p.collaborators().len() == 1)
2007            .await;
2008
2009        // Drop client B's connection and ensure client A observes client B leaving the project.
2010        client_b.disconnect(&cx_b.to_async()).unwrap();
2011        project_a
2012            .condition(cx_a, |p, _| p.collaborators().len() == 0)
2013            .await;
2014
2015        // Rejoin the project as client B
2016        let _project_b = Project::remote(
2017            project_id,
2018            client_b.clone(),
2019            client_b.user_store.clone(),
2020            lang_registry.clone(),
2021            fs.clone(),
2022            &mut cx_b.to_async(),
2023        )
2024        .await
2025        .unwrap();
2026
2027        // Client A sees that a guest has re-joined.
2028        project_a
2029            .condition(cx_a, |p, _| p.collaborators().len() == 1)
2030            .await;
2031
2032        // Simulate connection loss for client B and ensure client A observes client B leaving the project.
2033        client_b.wait_for_current_user(cx_b).await;
2034        server.disconnect_client(client_b.current_user_id(cx_b));
2035        cx_a.foreground().advance_clock(Duration::from_secs(3));
2036        project_a
2037            .condition(cx_a, |p, _| p.collaborators().len() == 0)
2038            .await;
2039    }
2040
2041    #[gpui::test(iterations = 10)]
2042    async fn test_collaborating_with_diagnostics(
2043        cx_a: &mut TestAppContext,
2044        cx_b: &mut TestAppContext,
2045    ) {
2046        cx_a.foreground().forbid_parking();
2047        let lang_registry = Arc::new(LanguageRegistry::test());
2048        let fs = FakeFs::new(cx_a.background());
2049
2050        // Set up a fake language server.
2051        let mut language = Language::new(
2052            LanguageConfig {
2053                name: "Rust".into(),
2054                path_suffixes: vec!["rs".to_string()],
2055                ..Default::default()
2056            },
2057            Some(tree_sitter_rust::language()),
2058        );
2059        let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
2060        lang_registry.add(Arc::new(language));
2061
2062        // Connect to a server as 2 clients.
2063        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2064        let client_a = server.create_client(cx_a, "user_a").await;
2065        let client_b = server.create_client(cx_b, "user_b").await;
2066
2067        // Share a project as client A
2068        fs.insert_tree(
2069            "/a",
2070            json!({
2071                ".zed.toml": r#"collaborators = ["user_b"]"#,
2072                "a.rs": "let one = two",
2073                "other.rs": "",
2074            }),
2075        )
2076        .await;
2077        let project_a = cx_a.update(|cx| {
2078            Project::local(
2079                client_a.clone(),
2080                client_a.user_store.clone(),
2081                lang_registry.clone(),
2082                fs.clone(),
2083                cx,
2084            )
2085        });
2086        let (worktree_a, _) = project_a
2087            .update(cx_a, |p, cx| {
2088                p.find_or_create_local_worktree("/a", true, cx)
2089            })
2090            .await
2091            .unwrap();
2092        worktree_a
2093            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2094            .await;
2095        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
2096        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2097        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
2098
2099        // Cause the language server to start.
2100        let _ = cx_a
2101            .background()
2102            .spawn(project_a.update(cx_a, |project, cx| {
2103                project.open_buffer(
2104                    ProjectPath {
2105                        worktree_id,
2106                        path: Path::new("other.rs").into(),
2107                    },
2108                    cx,
2109                )
2110            }))
2111            .await
2112            .unwrap();
2113
2114        // Simulate a language server reporting errors for a file.
2115        let mut fake_language_server = fake_language_servers.next().await.unwrap();
2116        fake_language_server
2117            .receive_notification::<lsp::notification::DidOpenTextDocument>()
2118            .await;
2119        fake_language_server.notify::<lsp::notification::PublishDiagnostics>(
2120            lsp::PublishDiagnosticsParams {
2121                uri: lsp::Url::from_file_path("/a/a.rs").unwrap(),
2122                version: None,
2123                diagnostics: vec![lsp::Diagnostic {
2124                    severity: Some(lsp::DiagnosticSeverity::ERROR),
2125                    range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 7)),
2126                    message: "message 1".to_string(),
2127                    ..Default::default()
2128                }],
2129            },
2130        );
2131
2132        // Wait for server to see the diagnostics update.
2133        server
2134            .condition(|store| {
2135                let worktree = store
2136                    .project(project_id)
2137                    .unwrap()
2138                    .share
2139                    .as_ref()
2140                    .unwrap()
2141                    .worktrees
2142                    .get(&worktree_id.to_proto())
2143                    .unwrap();
2144
2145                !worktree.diagnostic_summaries.is_empty()
2146            })
2147            .await;
2148
2149        // Join the worktree as client B.
2150        let project_b = Project::remote(
2151            project_id,
2152            client_b.clone(),
2153            client_b.user_store.clone(),
2154            lang_registry.clone(),
2155            fs.clone(),
2156            &mut cx_b.to_async(),
2157        )
2158        .await
2159        .unwrap();
2160
2161        project_b.read_with(cx_b, |project, cx| {
2162            assert_eq!(
2163                project.diagnostic_summaries(cx).collect::<Vec<_>>(),
2164                &[(
2165                    ProjectPath {
2166                        worktree_id,
2167                        path: Arc::from(Path::new("a.rs")),
2168                    },
2169                    DiagnosticSummary {
2170                        error_count: 1,
2171                        warning_count: 0,
2172                        ..Default::default()
2173                    },
2174                )]
2175            )
2176        });
2177
2178        // Simulate a language server reporting more errors for a file.
2179        fake_language_server.notify::<lsp::notification::PublishDiagnostics>(
2180            lsp::PublishDiagnosticsParams {
2181                uri: lsp::Url::from_file_path("/a/a.rs").unwrap(),
2182                version: None,
2183                diagnostics: vec![
2184                    lsp::Diagnostic {
2185                        severity: Some(lsp::DiagnosticSeverity::ERROR),
2186                        range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 7)),
2187                        message: "message 1".to_string(),
2188                        ..Default::default()
2189                    },
2190                    lsp::Diagnostic {
2191                        severity: Some(lsp::DiagnosticSeverity::WARNING),
2192                        range: lsp::Range::new(
2193                            lsp::Position::new(0, 10),
2194                            lsp::Position::new(0, 13),
2195                        ),
2196                        message: "message 2".to_string(),
2197                        ..Default::default()
2198                    },
2199                ],
2200            },
2201        );
2202
2203        // Client b gets the updated summaries
2204        project_b
2205            .condition(&cx_b, |project, cx| {
2206                project.diagnostic_summaries(cx).collect::<Vec<_>>()
2207                    == &[(
2208                        ProjectPath {
2209                            worktree_id,
2210                            path: Arc::from(Path::new("a.rs")),
2211                        },
2212                        DiagnosticSummary {
2213                            error_count: 1,
2214                            warning_count: 1,
2215                            ..Default::default()
2216                        },
2217                    )]
2218            })
2219            .await;
2220
2221        // Open the file with the errors on client B. They should be present.
2222        let buffer_b = cx_b
2223            .background()
2224            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
2225            .await
2226            .unwrap();
2227
2228        buffer_b.read_with(cx_b, |buffer, _| {
2229            assert_eq!(
2230                buffer
2231                    .snapshot()
2232                    .diagnostics_in_range::<_, Point>(0..buffer.len(), false)
2233                    .map(|entry| entry)
2234                    .collect::<Vec<_>>(),
2235                &[
2236                    DiagnosticEntry {
2237                        range: Point::new(0, 4)..Point::new(0, 7),
2238                        diagnostic: Diagnostic {
2239                            group_id: 0,
2240                            message: "message 1".to_string(),
2241                            severity: lsp::DiagnosticSeverity::ERROR,
2242                            is_primary: true,
2243                            ..Default::default()
2244                        }
2245                    },
2246                    DiagnosticEntry {
2247                        range: Point::new(0, 10)..Point::new(0, 13),
2248                        diagnostic: Diagnostic {
2249                            group_id: 1,
2250                            severity: lsp::DiagnosticSeverity::WARNING,
2251                            message: "message 2".to_string(),
2252                            is_primary: true,
2253                            ..Default::default()
2254                        }
2255                    }
2256                ]
2257            );
2258        });
2259    }
2260
2261    #[gpui::test(iterations = 10)]
2262    async fn test_collaborating_with_completion(
2263        cx_a: &mut TestAppContext,
2264        cx_b: &mut TestAppContext,
2265    ) {
2266        cx_a.foreground().forbid_parking();
2267        let lang_registry = Arc::new(LanguageRegistry::test());
2268        let fs = FakeFs::new(cx_a.background());
2269
2270        // Set up a fake language server.
2271        let mut language = Language::new(
2272            LanguageConfig {
2273                name: "Rust".into(),
2274                path_suffixes: vec!["rs".to_string()],
2275                ..Default::default()
2276            },
2277            Some(tree_sitter_rust::language()),
2278        );
2279        let mut fake_language_servers = language.set_fake_lsp_adapter(FakeLspAdapter {
2280            capabilities: lsp::ServerCapabilities {
2281                completion_provider: Some(lsp::CompletionOptions {
2282                    trigger_characters: Some(vec![".".to_string()]),
2283                    ..Default::default()
2284                }),
2285                ..Default::default()
2286            },
2287            ..Default::default()
2288        });
2289        lang_registry.add(Arc::new(language));
2290
2291        // Connect to a server as 2 clients.
2292        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2293        let client_a = server.create_client(cx_a, "user_a").await;
2294        let client_b = server.create_client(cx_b, "user_b").await;
2295
2296        // Share a project as client A
2297        fs.insert_tree(
2298            "/a",
2299            json!({
2300                ".zed.toml": r#"collaborators = ["user_b"]"#,
2301                "main.rs": "fn main() { a }",
2302                "other.rs": "",
2303            }),
2304        )
2305        .await;
2306        let project_a = cx_a.update(|cx| {
2307            Project::local(
2308                client_a.clone(),
2309                client_a.user_store.clone(),
2310                lang_registry.clone(),
2311                fs.clone(),
2312                cx,
2313            )
2314        });
2315        let (worktree_a, _) = project_a
2316            .update(cx_a, |p, cx| {
2317                p.find_or_create_local_worktree("/a", true, cx)
2318            })
2319            .await
2320            .unwrap();
2321        worktree_a
2322            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2323            .await;
2324        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
2325        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2326        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
2327
2328        // Join the worktree as client B.
2329        let project_b = Project::remote(
2330            project_id,
2331            client_b.clone(),
2332            client_b.user_store.clone(),
2333            lang_registry.clone(),
2334            fs.clone(),
2335            &mut cx_b.to_async(),
2336        )
2337        .await
2338        .unwrap();
2339
2340        // Open a file in an editor as the guest.
2341        let buffer_b = project_b
2342            .update(cx_b, |p, cx| p.open_buffer((worktree_id, "main.rs"), cx))
2343            .await
2344            .unwrap();
2345        let (window_b, _) = cx_b.add_window(|_| EmptyView);
2346        let editor_b = cx_b.add_view(window_b, |cx| {
2347            Editor::for_buffer(buffer_b.clone(), Some(project_b.clone()), cx)
2348        });
2349
2350        let fake_language_server = fake_language_servers.next().await.unwrap();
2351        buffer_b
2352            .condition(&cx_b, |buffer, _| !buffer.completion_triggers().is_empty())
2353            .await;
2354
2355        // Type a completion trigger character as the guest.
2356        editor_b.update(cx_b, |editor, cx| {
2357            editor.select_ranges([13..13], None, cx);
2358            editor.handle_input(&Input(".".into()), cx);
2359            cx.focus(&editor_b);
2360        });
2361
2362        // Receive a completion request as the host's language server.
2363        // Return some completions from the host's language server.
2364        cx_a.foreground().start_waiting();
2365        fake_language_server
2366            .handle_request::<lsp::request::Completion, _, _>(|params, _| async move {
2367                assert_eq!(
2368                    params.text_document_position.text_document.uri,
2369                    lsp::Url::from_file_path("/a/main.rs").unwrap(),
2370                );
2371                assert_eq!(
2372                    params.text_document_position.position,
2373                    lsp::Position::new(0, 14),
2374                );
2375
2376                Ok(Some(lsp::CompletionResponse::Array(vec![
2377                    lsp::CompletionItem {
2378                        label: "first_method(…)".into(),
2379                        detail: Some("fn(&mut self, B) -> C".into()),
2380                        text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
2381                            new_text: "first_method($1)".to_string(),
2382                            range: lsp::Range::new(
2383                                lsp::Position::new(0, 14),
2384                                lsp::Position::new(0, 14),
2385                            ),
2386                        })),
2387                        insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
2388                        ..Default::default()
2389                    },
2390                    lsp::CompletionItem {
2391                        label: "second_method(…)".into(),
2392                        detail: Some("fn(&mut self, C) -> D<E>".into()),
2393                        text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
2394                            new_text: "second_method()".to_string(),
2395                            range: lsp::Range::new(
2396                                lsp::Position::new(0, 14),
2397                                lsp::Position::new(0, 14),
2398                            ),
2399                        })),
2400                        insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
2401                        ..Default::default()
2402                    },
2403                ])))
2404            })
2405            .next()
2406            .await
2407            .unwrap();
2408        cx_a.foreground().finish_waiting();
2409
2410        // Open the buffer on the host.
2411        let buffer_a = project_a
2412            .update(cx_a, |p, cx| p.open_buffer((worktree_id, "main.rs"), cx))
2413            .await
2414            .unwrap();
2415        buffer_a
2416            .condition(&cx_a, |buffer, _| buffer.text() == "fn main() { a. }")
2417            .await;
2418
2419        // Confirm a completion on the guest.
2420        editor_b
2421            .condition(&cx_b, |editor, _| editor.context_menu_visible())
2422            .await;
2423        editor_b.update(cx_b, |editor, cx| {
2424            editor.confirm_completion(&ConfirmCompletion(Some(0)), cx);
2425            assert_eq!(editor.text(cx), "fn main() { a.first_method() }");
2426        });
2427
2428        // Return a resolved completion from the host's language server.
2429        // The resolved completion has an additional text edit.
2430        fake_language_server.handle_request::<lsp::request::ResolveCompletionItem, _, _>(
2431            |params, _| async move {
2432                assert_eq!(params.label, "first_method(…)");
2433                Ok(lsp::CompletionItem {
2434                    label: "first_method(…)".into(),
2435                    detail: Some("fn(&mut self, B) -> C".into()),
2436                    text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
2437                        new_text: "first_method($1)".to_string(),
2438                        range: lsp::Range::new(
2439                            lsp::Position::new(0, 14),
2440                            lsp::Position::new(0, 14),
2441                        ),
2442                    })),
2443                    additional_text_edits: Some(vec![lsp::TextEdit {
2444                        new_text: "use d::SomeTrait;\n".to_string(),
2445                        range: lsp::Range::new(lsp::Position::new(0, 0), lsp::Position::new(0, 0)),
2446                    }]),
2447                    insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
2448                    ..Default::default()
2449                })
2450            },
2451        );
2452
2453        // The additional edit is applied.
2454        buffer_a
2455            .condition(&cx_a, |buffer, _| {
2456                buffer.text() == "use d::SomeTrait;\nfn main() { a.first_method() }"
2457            })
2458            .await;
2459        buffer_b
2460            .condition(&cx_b, |buffer, _| {
2461                buffer.text() == "use d::SomeTrait;\nfn main() { a.first_method() }"
2462            })
2463            .await;
2464    }
2465
2466    #[gpui::test(iterations = 10)]
2467    async fn test_reloading_buffer_manually(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2468        cx_a.foreground().forbid_parking();
2469        let lang_registry = Arc::new(LanguageRegistry::test());
2470        let fs = FakeFs::new(cx_a.background());
2471
2472        // Connect to a server as 2 clients.
2473        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2474        let client_a = server.create_client(cx_a, "user_a").await;
2475        let client_b = server.create_client(cx_b, "user_b").await;
2476
2477        // Share a project as client A
2478        fs.insert_tree(
2479            "/a",
2480            json!({
2481                ".zed.toml": r#"collaborators = ["user_b"]"#,
2482                "a.rs": "let one = 1;",
2483            }),
2484        )
2485        .await;
2486        let project_a = cx_a.update(|cx| {
2487            Project::local(
2488                client_a.clone(),
2489                client_a.user_store.clone(),
2490                lang_registry.clone(),
2491                fs.clone(),
2492                cx,
2493            )
2494        });
2495        let (worktree_a, _) = project_a
2496            .update(cx_a, |p, cx| {
2497                p.find_or_create_local_worktree("/a", true, cx)
2498            })
2499            .await
2500            .unwrap();
2501        worktree_a
2502            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2503            .await;
2504        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
2505        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2506        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
2507        let buffer_a = project_a
2508            .update(cx_a, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx))
2509            .await
2510            .unwrap();
2511
2512        // Join the worktree as client B.
2513        let project_b = Project::remote(
2514            project_id,
2515            client_b.clone(),
2516            client_b.user_store.clone(),
2517            lang_registry.clone(),
2518            fs.clone(),
2519            &mut cx_b.to_async(),
2520        )
2521        .await
2522        .unwrap();
2523
2524        let buffer_b = cx_b
2525            .background()
2526            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
2527            .await
2528            .unwrap();
2529        buffer_b.update(cx_b, |buffer, cx| {
2530            buffer.edit([4..7], "six", cx);
2531            buffer.edit([10..11], "6", cx);
2532            assert_eq!(buffer.text(), "let six = 6;");
2533            assert!(buffer.is_dirty());
2534            assert!(!buffer.has_conflict());
2535        });
2536        buffer_a
2537            .condition(cx_a, |buffer, _| buffer.text() == "let six = 6;")
2538            .await;
2539
2540        fs.save(Path::new("/a/a.rs"), &Rope::from("let seven = 7;"))
2541            .await
2542            .unwrap();
2543        buffer_a
2544            .condition(cx_a, |buffer, _| buffer.has_conflict())
2545            .await;
2546        buffer_b
2547            .condition(cx_b, |buffer, _| buffer.has_conflict())
2548            .await;
2549
2550        project_b
2551            .update(cx_b, |project, cx| {
2552                project.reload_buffers(HashSet::from_iter([buffer_b.clone()]), true, cx)
2553            })
2554            .await
2555            .unwrap();
2556        buffer_a.read_with(cx_a, |buffer, _| {
2557            assert_eq!(buffer.text(), "let seven = 7;");
2558            assert!(!buffer.is_dirty());
2559            assert!(!buffer.has_conflict());
2560        });
2561        buffer_b.read_with(cx_b, |buffer, _| {
2562            assert_eq!(buffer.text(), "let seven = 7;");
2563            assert!(!buffer.is_dirty());
2564            assert!(!buffer.has_conflict());
2565        });
2566
2567        buffer_a.update(cx_a, |buffer, cx| {
2568            // Undoing on the host is a no-op when the reload was initiated by the guest.
2569            buffer.undo(cx);
2570            assert_eq!(buffer.text(), "let seven = 7;");
2571            assert!(!buffer.is_dirty());
2572            assert!(!buffer.has_conflict());
2573        });
2574        buffer_b.update(cx_b, |buffer, cx| {
2575            // Undoing on the guest rolls back the buffer to before it was reloaded but the conflict gets cleared.
2576            buffer.undo(cx);
2577            assert_eq!(buffer.text(), "let six = 6;");
2578            assert!(buffer.is_dirty());
2579            assert!(!buffer.has_conflict());
2580        });
2581    }
2582
2583    #[gpui::test(iterations = 10)]
2584    async fn test_formatting_buffer(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2585        cx_a.foreground().forbid_parking();
2586        let lang_registry = Arc::new(LanguageRegistry::test());
2587        let fs = FakeFs::new(cx_a.background());
2588
2589        // Set up a fake language server.
2590        let mut language = Language::new(
2591            LanguageConfig {
2592                name: "Rust".into(),
2593                path_suffixes: vec!["rs".to_string()],
2594                ..Default::default()
2595            },
2596            Some(tree_sitter_rust::language()),
2597        );
2598        let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
2599        lang_registry.add(Arc::new(language));
2600
2601        // Connect to a server as 2 clients.
2602        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2603        let client_a = server.create_client(cx_a, "user_a").await;
2604        let client_b = server.create_client(cx_b, "user_b").await;
2605
2606        // Share a project as client A
2607        fs.insert_tree(
2608            "/a",
2609            json!({
2610                ".zed.toml": r#"collaborators = ["user_b"]"#,
2611                "a.rs": "let one = two",
2612            }),
2613        )
2614        .await;
2615        let project_a = cx_a.update(|cx| {
2616            Project::local(
2617                client_a.clone(),
2618                client_a.user_store.clone(),
2619                lang_registry.clone(),
2620                fs.clone(),
2621                cx,
2622            )
2623        });
2624        let (worktree_a, _) = project_a
2625            .update(cx_a, |p, cx| {
2626                p.find_or_create_local_worktree("/a", true, cx)
2627            })
2628            .await
2629            .unwrap();
2630        worktree_a
2631            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2632            .await;
2633        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
2634        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2635        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
2636
2637        // Join the worktree as client B.
2638        let project_b = Project::remote(
2639            project_id,
2640            client_b.clone(),
2641            client_b.user_store.clone(),
2642            lang_registry.clone(),
2643            fs.clone(),
2644            &mut cx_b.to_async(),
2645        )
2646        .await
2647        .unwrap();
2648
2649        let buffer_b = cx_b
2650            .background()
2651            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
2652            .await
2653            .unwrap();
2654
2655        let fake_language_server = fake_language_servers.next().await.unwrap();
2656        fake_language_server.handle_request::<lsp::request::Formatting, _, _>(|_, _| async move {
2657            Ok(Some(vec![
2658                lsp::TextEdit {
2659                    range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 4)),
2660                    new_text: "h".to_string(),
2661                },
2662                lsp::TextEdit {
2663                    range: lsp::Range::new(lsp::Position::new(0, 7), lsp::Position::new(0, 7)),
2664                    new_text: "y".to_string(),
2665                },
2666            ]))
2667        });
2668
2669        project_b
2670            .update(cx_b, |project, cx| {
2671                project.format(HashSet::from_iter([buffer_b.clone()]), true, cx)
2672            })
2673            .await
2674            .unwrap();
2675        assert_eq!(
2676            buffer_b.read_with(cx_b, |buffer, _| buffer.text()),
2677            "let honey = two"
2678        );
2679    }
2680
2681    #[gpui::test(iterations = 10)]
2682    async fn test_definition(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2683        cx_a.foreground().forbid_parking();
2684        let lang_registry = Arc::new(LanguageRegistry::test());
2685        let fs = FakeFs::new(cx_a.background());
2686        fs.insert_tree(
2687            "/root-1",
2688            json!({
2689                ".zed.toml": r#"collaborators = ["user_b"]"#,
2690                "a.rs": "const ONE: usize = b::TWO + b::THREE;",
2691            }),
2692        )
2693        .await;
2694        fs.insert_tree(
2695            "/root-2",
2696            json!({
2697                "b.rs": "const TWO: usize = 2;\nconst THREE: usize = 3;",
2698            }),
2699        )
2700        .await;
2701
2702        // Set up a fake language server.
2703        let mut language = Language::new(
2704            LanguageConfig {
2705                name: "Rust".into(),
2706                path_suffixes: vec!["rs".to_string()],
2707                ..Default::default()
2708            },
2709            Some(tree_sitter_rust::language()),
2710        );
2711        let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
2712        lang_registry.add(Arc::new(language));
2713
2714        // Connect to a server as 2 clients.
2715        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2716        let client_a = server.create_client(cx_a, "user_a").await;
2717        let client_b = server.create_client(cx_b, "user_b").await;
2718
2719        // Share a project as client A
2720        let project_a = cx_a.update(|cx| {
2721            Project::local(
2722                client_a.clone(),
2723                client_a.user_store.clone(),
2724                lang_registry.clone(),
2725                fs.clone(),
2726                cx,
2727            )
2728        });
2729        let (worktree_a, _) = project_a
2730            .update(cx_a, |p, cx| {
2731                p.find_or_create_local_worktree("/root-1", true, cx)
2732            })
2733            .await
2734            .unwrap();
2735        worktree_a
2736            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2737            .await;
2738        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
2739        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2740        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
2741
2742        // Join the worktree as client B.
2743        let project_b = Project::remote(
2744            project_id,
2745            client_b.clone(),
2746            client_b.user_store.clone(),
2747            lang_registry.clone(),
2748            fs.clone(),
2749            &mut cx_b.to_async(),
2750        )
2751        .await
2752        .unwrap();
2753
2754        // Open the file on client B.
2755        let buffer_b = cx_b
2756            .background()
2757            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
2758            .await
2759            .unwrap();
2760
2761        // Request the definition of a symbol as the guest.
2762        let fake_language_server = fake_language_servers.next().await.unwrap();
2763        fake_language_server.handle_request::<lsp::request::GotoDefinition, _, _>(
2764            |_, _| async move {
2765                Ok(Some(lsp::GotoDefinitionResponse::Scalar(
2766                    lsp::Location::new(
2767                        lsp::Url::from_file_path("/root-2/b.rs").unwrap(),
2768                        lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
2769                    ),
2770                )))
2771            },
2772        );
2773
2774        let definitions_1 = project_b
2775            .update(cx_b, |p, cx| p.definition(&buffer_b, 23, cx))
2776            .await
2777            .unwrap();
2778        cx_b.read(|cx| {
2779            assert_eq!(definitions_1.len(), 1);
2780            assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
2781            let target_buffer = definitions_1[0].buffer.read(cx);
2782            assert_eq!(
2783                target_buffer.text(),
2784                "const TWO: usize = 2;\nconst THREE: usize = 3;"
2785            );
2786            assert_eq!(
2787                definitions_1[0].range.to_point(target_buffer),
2788                Point::new(0, 6)..Point::new(0, 9)
2789            );
2790        });
2791
2792        // Try getting more definitions for the same buffer, ensuring the buffer gets reused from
2793        // the previous call to `definition`.
2794        fake_language_server.handle_request::<lsp::request::GotoDefinition, _, _>(
2795            |_, _| async move {
2796                Ok(Some(lsp::GotoDefinitionResponse::Scalar(
2797                    lsp::Location::new(
2798                        lsp::Url::from_file_path("/root-2/b.rs").unwrap(),
2799                        lsp::Range::new(lsp::Position::new(1, 6), lsp::Position::new(1, 11)),
2800                    ),
2801                )))
2802            },
2803        );
2804
2805        let definitions_2 = project_b
2806            .update(cx_b, |p, cx| p.definition(&buffer_b, 33, cx))
2807            .await
2808            .unwrap();
2809        cx_b.read(|cx| {
2810            assert_eq!(definitions_2.len(), 1);
2811            assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
2812            let target_buffer = definitions_2[0].buffer.read(cx);
2813            assert_eq!(
2814                target_buffer.text(),
2815                "const TWO: usize = 2;\nconst THREE: usize = 3;"
2816            );
2817            assert_eq!(
2818                definitions_2[0].range.to_point(target_buffer),
2819                Point::new(1, 6)..Point::new(1, 11)
2820            );
2821        });
2822        assert_eq!(definitions_1[0].buffer, definitions_2[0].buffer);
2823    }
2824
2825    #[gpui::test(iterations = 10)]
2826    async fn test_references(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2827        cx_a.foreground().forbid_parking();
2828        let lang_registry = Arc::new(LanguageRegistry::test());
2829        let fs = FakeFs::new(cx_a.background());
2830        fs.insert_tree(
2831            "/root-1",
2832            json!({
2833                ".zed.toml": r#"collaborators = ["user_b"]"#,
2834                "one.rs": "const ONE: usize = 1;",
2835                "two.rs": "const TWO: usize = one::ONE + one::ONE;",
2836            }),
2837        )
2838        .await;
2839        fs.insert_tree(
2840            "/root-2",
2841            json!({
2842                "three.rs": "const THREE: usize = two::TWO + one::ONE;",
2843            }),
2844        )
2845        .await;
2846
2847        // Set up a fake language server.
2848        let mut language = Language::new(
2849            LanguageConfig {
2850                name: "Rust".into(),
2851                path_suffixes: vec!["rs".to_string()],
2852                ..Default::default()
2853            },
2854            Some(tree_sitter_rust::language()),
2855        );
2856        let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
2857        lang_registry.add(Arc::new(language));
2858
2859        // Connect to a server as 2 clients.
2860        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2861        let client_a = server.create_client(cx_a, "user_a").await;
2862        let client_b = server.create_client(cx_b, "user_b").await;
2863
2864        // Share a project as client A
2865        let project_a = cx_a.update(|cx| {
2866            Project::local(
2867                client_a.clone(),
2868                client_a.user_store.clone(),
2869                lang_registry.clone(),
2870                fs.clone(),
2871                cx,
2872            )
2873        });
2874        let (worktree_a, _) = project_a
2875            .update(cx_a, |p, cx| {
2876                p.find_or_create_local_worktree("/root-1", true, cx)
2877            })
2878            .await
2879            .unwrap();
2880        worktree_a
2881            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2882            .await;
2883        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
2884        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2885        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
2886
2887        // Join the worktree as client B.
2888        let project_b = Project::remote(
2889            project_id,
2890            client_b.clone(),
2891            client_b.user_store.clone(),
2892            lang_registry.clone(),
2893            fs.clone(),
2894            &mut cx_b.to_async(),
2895        )
2896        .await
2897        .unwrap();
2898
2899        // Open the file on client B.
2900        let buffer_b = cx_b
2901            .background()
2902            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "one.rs"), cx)))
2903            .await
2904            .unwrap();
2905
2906        // Request references to a symbol as the guest.
2907        let fake_language_server = fake_language_servers.next().await.unwrap();
2908        fake_language_server.handle_request::<lsp::request::References, _, _>(
2909            |params, _| async move {
2910                assert_eq!(
2911                    params.text_document_position.text_document.uri.as_str(),
2912                    "file:///root-1/one.rs"
2913                );
2914                Ok(Some(vec![
2915                    lsp::Location {
2916                        uri: lsp::Url::from_file_path("/root-1/two.rs").unwrap(),
2917                        range: lsp::Range::new(
2918                            lsp::Position::new(0, 24),
2919                            lsp::Position::new(0, 27),
2920                        ),
2921                    },
2922                    lsp::Location {
2923                        uri: lsp::Url::from_file_path("/root-1/two.rs").unwrap(),
2924                        range: lsp::Range::new(
2925                            lsp::Position::new(0, 35),
2926                            lsp::Position::new(0, 38),
2927                        ),
2928                    },
2929                    lsp::Location {
2930                        uri: lsp::Url::from_file_path("/root-2/three.rs").unwrap(),
2931                        range: lsp::Range::new(
2932                            lsp::Position::new(0, 37),
2933                            lsp::Position::new(0, 40),
2934                        ),
2935                    },
2936                ]))
2937            },
2938        );
2939
2940        let references = project_b
2941            .update(cx_b, |p, cx| p.references(&buffer_b, 7, cx))
2942            .await
2943            .unwrap();
2944        cx_b.read(|cx| {
2945            assert_eq!(references.len(), 3);
2946            assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
2947
2948            let two_buffer = references[0].buffer.read(cx);
2949            let three_buffer = references[2].buffer.read(cx);
2950            assert_eq!(
2951                two_buffer.file().unwrap().path().as_ref(),
2952                Path::new("two.rs")
2953            );
2954            assert_eq!(references[1].buffer, references[0].buffer);
2955            assert_eq!(
2956                three_buffer.file().unwrap().full_path(cx),
2957                Path::new("three.rs")
2958            );
2959
2960            assert_eq!(references[0].range.to_offset(&two_buffer), 24..27);
2961            assert_eq!(references[1].range.to_offset(&two_buffer), 35..38);
2962            assert_eq!(references[2].range.to_offset(&three_buffer), 37..40);
2963        });
2964    }
2965
2966    #[gpui::test(iterations = 10)]
2967    async fn test_project_search(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2968        cx_a.foreground().forbid_parking();
2969        let lang_registry = Arc::new(LanguageRegistry::test());
2970        let fs = FakeFs::new(cx_a.background());
2971        fs.insert_tree(
2972            "/root-1",
2973            json!({
2974                ".zed.toml": r#"collaborators = ["user_b"]"#,
2975                "a": "hello world",
2976                "b": "goodnight moon",
2977                "c": "a world of goo",
2978                "d": "world champion of clown world",
2979            }),
2980        )
2981        .await;
2982        fs.insert_tree(
2983            "/root-2",
2984            json!({
2985                "e": "disney world is fun",
2986            }),
2987        )
2988        .await;
2989
2990        // Connect to a server as 2 clients.
2991        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2992        let client_a = server.create_client(cx_a, "user_a").await;
2993        let client_b = server.create_client(cx_b, "user_b").await;
2994
2995        // Share a project as client A
2996        let project_a = cx_a.update(|cx| {
2997            Project::local(
2998                client_a.clone(),
2999                client_a.user_store.clone(),
3000                lang_registry.clone(),
3001                fs.clone(),
3002                cx,
3003            )
3004        });
3005        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
3006
3007        let (worktree_1, _) = project_a
3008            .update(cx_a, |p, cx| {
3009                p.find_or_create_local_worktree("/root-1", true, cx)
3010            })
3011            .await
3012            .unwrap();
3013        worktree_1
3014            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3015            .await;
3016        let (worktree_2, _) = project_a
3017            .update(cx_a, |p, cx| {
3018                p.find_or_create_local_worktree("/root-2", true, cx)
3019            })
3020            .await
3021            .unwrap();
3022        worktree_2
3023            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3024            .await;
3025
3026        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
3027
3028        // Join the worktree as client B.
3029        let project_b = Project::remote(
3030            project_id,
3031            client_b.clone(),
3032            client_b.user_store.clone(),
3033            lang_registry.clone(),
3034            fs.clone(),
3035            &mut cx_b.to_async(),
3036        )
3037        .await
3038        .unwrap();
3039
3040        let results = project_b
3041            .update(cx_b, |project, cx| {
3042                project.search(SearchQuery::text("world", false, false), cx)
3043            })
3044            .await
3045            .unwrap();
3046
3047        let mut ranges_by_path = results
3048            .into_iter()
3049            .map(|(buffer, ranges)| {
3050                buffer.read_with(cx_b, |buffer, cx| {
3051                    let path = buffer.file().unwrap().full_path(cx);
3052                    let offset_ranges = ranges
3053                        .into_iter()
3054                        .map(|range| range.to_offset(buffer))
3055                        .collect::<Vec<_>>();
3056                    (path, offset_ranges)
3057                })
3058            })
3059            .collect::<Vec<_>>();
3060        ranges_by_path.sort_by_key(|(path, _)| path.clone());
3061
3062        assert_eq!(
3063            ranges_by_path,
3064            &[
3065                (PathBuf::from("root-1/a"), vec![6..11]),
3066                (PathBuf::from("root-1/c"), vec![2..7]),
3067                (PathBuf::from("root-1/d"), vec![0..5, 24..29]),
3068                (PathBuf::from("root-2/e"), vec![7..12]),
3069            ]
3070        );
3071    }
3072
3073    #[gpui::test(iterations = 10)]
3074    async fn test_document_highlights(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3075        cx_a.foreground().forbid_parking();
3076        let lang_registry = Arc::new(LanguageRegistry::test());
3077        let fs = FakeFs::new(cx_a.background());
3078        fs.insert_tree(
3079            "/root-1",
3080            json!({
3081                ".zed.toml": r#"collaborators = ["user_b"]"#,
3082                "main.rs": "fn double(number: i32) -> i32 { number + number }",
3083            }),
3084        )
3085        .await;
3086
3087        // Set up a fake language server.
3088        let mut language = Language::new(
3089            LanguageConfig {
3090                name: "Rust".into(),
3091                path_suffixes: vec!["rs".to_string()],
3092                ..Default::default()
3093            },
3094            Some(tree_sitter_rust::language()),
3095        );
3096        let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3097        lang_registry.add(Arc::new(language));
3098
3099        // Connect to a server as 2 clients.
3100        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3101        let client_a = server.create_client(cx_a, "user_a").await;
3102        let client_b = server.create_client(cx_b, "user_b").await;
3103
3104        // Share a project as client A
3105        let project_a = cx_a.update(|cx| {
3106            Project::local(
3107                client_a.clone(),
3108                client_a.user_store.clone(),
3109                lang_registry.clone(),
3110                fs.clone(),
3111                cx,
3112            )
3113        });
3114        let (worktree_a, _) = project_a
3115            .update(cx_a, |p, cx| {
3116                p.find_or_create_local_worktree("/root-1", true, cx)
3117            })
3118            .await
3119            .unwrap();
3120        worktree_a
3121            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3122            .await;
3123        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
3124        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3125        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
3126
3127        // Join the worktree as client B.
3128        let project_b = Project::remote(
3129            project_id,
3130            client_b.clone(),
3131            client_b.user_store.clone(),
3132            lang_registry.clone(),
3133            fs.clone(),
3134            &mut cx_b.to_async(),
3135        )
3136        .await
3137        .unwrap();
3138
3139        // Open the file on client B.
3140        let buffer_b = cx_b
3141            .background()
3142            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "main.rs"), cx)))
3143            .await
3144            .unwrap();
3145
3146        // Request document highlights as the guest.
3147        let fake_language_server = fake_language_servers.next().await.unwrap();
3148        fake_language_server.handle_request::<lsp::request::DocumentHighlightRequest, _, _>(
3149            |params, _| async move {
3150                assert_eq!(
3151                    params
3152                        .text_document_position_params
3153                        .text_document
3154                        .uri
3155                        .as_str(),
3156                    "file:///root-1/main.rs"
3157                );
3158                assert_eq!(
3159                    params.text_document_position_params.position,
3160                    lsp::Position::new(0, 34)
3161                );
3162                Ok(Some(vec![
3163                    lsp::DocumentHighlight {
3164                        kind: Some(lsp::DocumentHighlightKind::WRITE),
3165                        range: lsp::Range::new(
3166                            lsp::Position::new(0, 10),
3167                            lsp::Position::new(0, 16),
3168                        ),
3169                    },
3170                    lsp::DocumentHighlight {
3171                        kind: Some(lsp::DocumentHighlightKind::READ),
3172                        range: lsp::Range::new(
3173                            lsp::Position::new(0, 32),
3174                            lsp::Position::new(0, 38),
3175                        ),
3176                    },
3177                    lsp::DocumentHighlight {
3178                        kind: Some(lsp::DocumentHighlightKind::READ),
3179                        range: lsp::Range::new(
3180                            lsp::Position::new(0, 41),
3181                            lsp::Position::new(0, 47),
3182                        ),
3183                    },
3184                ]))
3185            },
3186        );
3187
3188        let highlights = project_b
3189            .update(cx_b, |p, cx| p.document_highlights(&buffer_b, 34, cx))
3190            .await
3191            .unwrap();
3192        buffer_b.read_with(cx_b, |buffer, _| {
3193            let snapshot = buffer.snapshot();
3194
3195            let highlights = highlights
3196                .into_iter()
3197                .map(|highlight| (highlight.kind, highlight.range.to_offset(&snapshot)))
3198                .collect::<Vec<_>>();
3199            assert_eq!(
3200                highlights,
3201                &[
3202                    (lsp::DocumentHighlightKind::WRITE, 10..16),
3203                    (lsp::DocumentHighlightKind::READ, 32..38),
3204                    (lsp::DocumentHighlightKind::READ, 41..47)
3205                ]
3206            )
3207        });
3208    }
3209
3210    #[gpui::test(iterations = 10)]
3211    async fn test_project_symbols(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3212        cx_a.foreground().forbid_parking();
3213        let lang_registry = Arc::new(LanguageRegistry::test());
3214        let fs = FakeFs::new(cx_a.background());
3215        fs.insert_tree(
3216            "/code",
3217            json!({
3218                "crate-1": {
3219                    ".zed.toml": r#"collaborators = ["user_b"]"#,
3220                    "one.rs": "const ONE: usize = 1;",
3221                },
3222                "crate-2": {
3223                    "two.rs": "const TWO: usize = 2; const THREE: usize = 3;",
3224                },
3225                "private": {
3226                    "passwords.txt": "the-password",
3227                }
3228            }),
3229        )
3230        .await;
3231
3232        // Set up a fake language server.
3233        let mut language = Language::new(
3234            LanguageConfig {
3235                name: "Rust".into(),
3236                path_suffixes: vec!["rs".to_string()],
3237                ..Default::default()
3238            },
3239            Some(tree_sitter_rust::language()),
3240        );
3241        let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3242        lang_registry.add(Arc::new(language));
3243
3244        // Connect to a server as 2 clients.
3245        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3246        let client_a = server.create_client(cx_a, "user_a").await;
3247        let client_b = server.create_client(cx_b, "user_b").await;
3248
3249        // Share a project as client A
3250        let project_a = cx_a.update(|cx| {
3251            Project::local(
3252                client_a.clone(),
3253                client_a.user_store.clone(),
3254                lang_registry.clone(),
3255                fs.clone(),
3256                cx,
3257            )
3258        });
3259        let (worktree_a, _) = project_a
3260            .update(cx_a, |p, cx| {
3261                p.find_or_create_local_worktree("/code/crate-1", true, cx)
3262            })
3263            .await
3264            .unwrap();
3265        worktree_a
3266            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3267            .await;
3268        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
3269        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3270        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
3271
3272        // Join the worktree as client B.
3273        let project_b = Project::remote(
3274            project_id,
3275            client_b.clone(),
3276            client_b.user_store.clone(),
3277            lang_registry.clone(),
3278            fs.clone(),
3279            &mut cx_b.to_async(),
3280        )
3281        .await
3282        .unwrap();
3283
3284        // Cause the language server to start.
3285        let _buffer = cx_b
3286            .background()
3287            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "one.rs"), cx)))
3288            .await
3289            .unwrap();
3290
3291        let fake_language_server = fake_language_servers.next().await.unwrap();
3292        fake_language_server.handle_request::<lsp::request::WorkspaceSymbol, _, _>(
3293            |_, _| async move {
3294                #[allow(deprecated)]
3295                Ok(Some(vec![lsp::SymbolInformation {
3296                    name: "TWO".into(),
3297                    location: lsp::Location {
3298                        uri: lsp::Url::from_file_path("/code/crate-2/two.rs").unwrap(),
3299                        range: lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
3300                    },
3301                    kind: lsp::SymbolKind::CONSTANT,
3302                    tags: None,
3303                    container_name: None,
3304                    deprecated: None,
3305                }]))
3306            },
3307        );
3308
3309        // Request the definition of a symbol as the guest.
3310        let symbols = project_b
3311            .update(cx_b, |p, cx| p.symbols("two", cx))
3312            .await
3313            .unwrap();
3314        assert_eq!(symbols.len(), 1);
3315        assert_eq!(symbols[0].name, "TWO");
3316
3317        // Open one of the returned symbols.
3318        let buffer_b_2 = project_b
3319            .update(cx_b, |project, cx| {
3320                project.open_buffer_for_symbol(&symbols[0], cx)
3321            })
3322            .await
3323            .unwrap();
3324        buffer_b_2.read_with(cx_b, |buffer, _| {
3325            assert_eq!(
3326                buffer.file().unwrap().path().as_ref(),
3327                Path::new("../crate-2/two.rs")
3328            );
3329        });
3330
3331        // Attempt to craft a symbol and violate host's privacy by opening an arbitrary file.
3332        let mut fake_symbol = symbols[0].clone();
3333        fake_symbol.path = Path::new("/code/secrets").into();
3334        let error = project_b
3335            .update(cx_b, |project, cx| {
3336                project.open_buffer_for_symbol(&fake_symbol, cx)
3337            })
3338            .await
3339            .unwrap_err();
3340        assert!(error.to_string().contains("invalid symbol signature"));
3341    }
3342
3343    #[gpui::test(iterations = 10)]
3344    async fn test_open_buffer_while_getting_definition_pointing_to_it(
3345        cx_a: &mut TestAppContext,
3346        cx_b: &mut TestAppContext,
3347        mut rng: StdRng,
3348    ) {
3349        cx_a.foreground().forbid_parking();
3350        let lang_registry = Arc::new(LanguageRegistry::test());
3351        let fs = FakeFs::new(cx_a.background());
3352        fs.insert_tree(
3353            "/root",
3354            json!({
3355                ".zed.toml": r#"collaborators = ["user_b"]"#,
3356                "a.rs": "const ONE: usize = b::TWO;",
3357                "b.rs": "const TWO: usize = 2",
3358            }),
3359        )
3360        .await;
3361
3362        // Set up a fake language server.
3363        let mut language = Language::new(
3364            LanguageConfig {
3365                name: "Rust".into(),
3366                path_suffixes: vec!["rs".to_string()],
3367                ..Default::default()
3368            },
3369            Some(tree_sitter_rust::language()),
3370        );
3371        let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3372        lang_registry.add(Arc::new(language));
3373
3374        // Connect to a server as 2 clients.
3375        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3376        let client_a = server.create_client(cx_a, "user_a").await;
3377        let client_b = server.create_client(cx_b, "user_b").await;
3378
3379        // Share a project as client A
3380        let project_a = cx_a.update(|cx| {
3381            Project::local(
3382                client_a.clone(),
3383                client_a.user_store.clone(),
3384                lang_registry.clone(),
3385                fs.clone(),
3386                cx,
3387            )
3388        });
3389
3390        let (worktree_a, _) = project_a
3391            .update(cx_a, |p, cx| {
3392                p.find_or_create_local_worktree("/root", true, cx)
3393            })
3394            .await
3395            .unwrap();
3396        worktree_a
3397            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3398            .await;
3399        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
3400        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3401        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
3402
3403        // Join the worktree as client B.
3404        let project_b = Project::remote(
3405            project_id,
3406            client_b.clone(),
3407            client_b.user_store.clone(),
3408            lang_registry.clone(),
3409            fs.clone(),
3410            &mut cx_b.to_async(),
3411        )
3412        .await
3413        .unwrap();
3414
3415        let buffer_b1 = cx_b
3416            .background()
3417            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
3418            .await
3419            .unwrap();
3420
3421        let fake_language_server = fake_language_servers.next().await.unwrap();
3422        fake_language_server.handle_request::<lsp::request::GotoDefinition, _, _>(
3423            |_, _| async move {
3424                Ok(Some(lsp::GotoDefinitionResponse::Scalar(
3425                    lsp::Location::new(
3426                        lsp::Url::from_file_path("/root/b.rs").unwrap(),
3427                        lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
3428                    ),
3429                )))
3430            },
3431        );
3432
3433        let definitions;
3434        let buffer_b2;
3435        if rng.gen() {
3436            definitions = project_b.update(cx_b, |p, cx| p.definition(&buffer_b1, 23, cx));
3437            buffer_b2 = project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "b.rs"), cx));
3438        } else {
3439            buffer_b2 = project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "b.rs"), cx));
3440            definitions = project_b.update(cx_b, |p, cx| p.definition(&buffer_b1, 23, cx));
3441        }
3442
3443        let buffer_b2 = buffer_b2.await.unwrap();
3444        let definitions = definitions.await.unwrap();
3445        assert_eq!(definitions.len(), 1);
3446        assert_eq!(definitions[0].buffer, buffer_b2);
3447    }
3448
3449    #[gpui::test(iterations = 10)]
3450    async fn test_collaborating_with_code_actions(
3451        cx_a: &mut TestAppContext,
3452        cx_b: &mut TestAppContext,
3453    ) {
3454        cx_a.foreground().forbid_parking();
3455        let lang_registry = Arc::new(LanguageRegistry::test());
3456        let fs = FakeFs::new(cx_a.background());
3457        cx_b.update(|cx| editor::init(cx));
3458
3459        // Set up a fake language server.
3460        let mut language = Language::new(
3461            LanguageConfig {
3462                name: "Rust".into(),
3463                path_suffixes: vec!["rs".to_string()],
3464                ..Default::default()
3465            },
3466            Some(tree_sitter_rust::language()),
3467        );
3468        let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3469        lang_registry.add(Arc::new(language));
3470
3471        // Connect to a server as 2 clients.
3472        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3473        let client_a = server.create_client(cx_a, "user_a").await;
3474        let client_b = server.create_client(cx_b, "user_b").await;
3475
3476        // Share a project as client A
3477        fs.insert_tree(
3478            "/a",
3479            json!({
3480                ".zed.toml": r#"collaborators = ["user_b"]"#,
3481                "main.rs": "mod other;\nfn main() { let foo = other::foo(); }",
3482                "other.rs": "pub fn foo() -> usize { 4 }",
3483            }),
3484        )
3485        .await;
3486        let project_a = cx_a.update(|cx| {
3487            Project::local(
3488                client_a.clone(),
3489                client_a.user_store.clone(),
3490                lang_registry.clone(),
3491                fs.clone(),
3492                cx,
3493            )
3494        });
3495        let (worktree_a, _) = project_a
3496            .update(cx_a, |p, cx| {
3497                p.find_or_create_local_worktree("/a", true, cx)
3498            })
3499            .await
3500            .unwrap();
3501        worktree_a
3502            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3503            .await;
3504        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
3505        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3506        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
3507
3508        // Join the worktree as client B.
3509        let project_b = Project::remote(
3510            project_id,
3511            client_b.clone(),
3512            client_b.user_store.clone(),
3513            lang_registry.clone(),
3514            fs.clone(),
3515            &mut cx_b.to_async(),
3516        )
3517        .await
3518        .unwrap();
3519        let mut params = cx_b.update(WorkspaceParams::test);
3520        params.languages = lang_registry.clone();
3521        params.client = client_b.client.clone();
3522        params.user_store = client_b.user_store.clone();
3523        params.project = project_b;
3524
3525        let (_window_b, workspace_b) = cx_b.add_window(|cx| Workspace::new(&params, cx));
3526        let editor_b = workspace_b
3527            .update(cx_b, |workspace, cx| {
3528                workspace.open_path((worktree_id, "main.rs"), cx)
3529            })
3530            .await
3531            .unwrap()
3532            .downcast::<Editor>()
3533            .unwrap();
3534
3535        let mut fake_language_server = fake_language_servers.next().await.unwrap();
3536        fake_language_server
3537            .handle_request::<lsp::request::CodeActionRequest, _, _>(|params, _| async move {
3538                assert_eq!(
3539                    params.text_document.uri,
3540                    lsp::Url::from_file_path("/a/main.rs").unwrap(),
3541                );
3542                assert_eq!(params.range.start, lsp::Position::new(0, 0));
3543                assert_eq!(params.range.end, lsp::Position::new(0, 0));
3544                Ok(None)
3545            })
3546            .next()
3547            .await;
3548
3549        // Move cursor to a location that contains code actions.
3550        editor_b.update(cx_b, |editor, cx| {
3551            editor.select_ranges([Point::new(1, 31)..Point::new(1, 31)], None, cx);
3552            cx.focus(&editor_b);
3553        });
3554
3555        fake_language_server
3556            .handle_request::<lsp::request::CodeActionRequest, _, _>(|params, _| async move {
3557                assert_eq!(
3558                    params.text_document.uri,
3559                    lsp::Url::from_file_path("/a/main.rs").unwrap(),
3560                );
3561                assert_eq!(params.range.start, lsp::Position::new(1, 31));
3562                assert_eq!(params.range.end, lsp::Position::new(1, 31));
3563
3564                Ok(Some(vec![lsp::CodeActionOrCommand::CodeAction(
3565                    lsp::CodeAction {
3566                        title: "Inline into all callers".to_string(),
3567                        edit: Some(lsp::WorkspaceEdit {
3568                            changes: Some(
3569                                [
3570                                    (
3571                                        lsp::Url::from_file_path("/a/main.rs").unwrap(),
3572                                        vec![lsp::TextEdit::new(
3573                                            lsp::Range::new(
3574                                                lsp::Position::new(1, 22),
3575                                                lsp::Position::new(1, 34),
3576                                            ),
3577                                            "4".to_string(),
3578                                        )],
3579                                    ),
3580                                    (
3581                                        lsp::Url::from_file_path("/a/other.rs").unwrap(),
3582                                        vec![lsp::TextEdit::new(
3583                                            lsp::Range::new(
3584                                                lsp::Position::new(0, 0),
3585                                                lsp::Position::new(0, 27),
3586                                            ),
3587                                            "".to_string(),
3588                                        )],
3589                                    ),
3590                                ]
3591                                .into_iter()
3592                                .collect(),
3593                            ),
3594                            ..Default::default()
3595                        }),
3596                        data: Some(json!({
3597                            "codeActionParams": {
3598                                "range": {
3599                                    "start": {"line": 1, "column": 31},
3600                                    "end": {"line": 1, "column": 31},
3601                                }
3602                            }
3603                        })),
3604                        ..Default::default()
3605                    },
3606                )]))
3607            })
3608            .next()
3609            .await;
3610
3611        // Toggle code actions and wait for them to display.
3612        editor_b.update(cx_b, |editor, cx| {
3613            editor.toggle_code_actions(&ToggleCodeActions(false), cx);
3614        });
3615        editor_b
3616            .condition(&cx_b, |editor, _| editor.context_menu_visible())
3617            .await;
3618
3619        fake_language_server.remove_request_handler::<lsp::request::CodeActionRequest>();
3620
3621        // Confirming the code action will trigger a resolve request.
3622        let confirm_action = workspace_b
3623            .update(cx_b, |workspace, cx| {
3624                Editor::confirm_code_action(workspace, &ConfirmCodeAction(Some(0)), cx)
3625            })
3626            .unwrap();
3627        fake_language_server.handle_request::<lsp::request::CodeActionResolveRequest, _, _>(
3628            |_, _| async move {
3629                Ok(lsp::CodeAction {
3630                    title: "Inline into all callers".to_string(),
3631                    edit: Some(lsp::WorkspaceEdit {
3632                        changes: Some(
3633                            [
3634                                (
3635                                    lsp::Url::from_file_path("/a/main.rs").unwrap(),
3636                                    vec![lsp::TextEdit::new(
3637                                        lsp::Range::new(
3638                                            lsp::Position::new(1, 22),
3639                                            lsp::Position::new(1, 34),
3640                                        ),
3641                                        "4".to_string(),
3642                                    )],
3643                                ),
3644                                (
3645                                    lsp::Url::from_file_path("/a/other.rs").unwrap(),
3646                                    vec![lsp::TextEdit::new(
3647                                        lsp::Range::new(
3648                                            lsp::Position::new(0, 0),
3649                                            lsp::Position::new(0, 27),
3650                                        ),
3651                                        "".to_string(),
3652                                    )],
3653                                ),
3654                            ]
3655                            .into_iter()
3656                            .collect(),
3657                        ),
3658                        ..Default::default()
3659                    }),
3660                    ..Default::default()
3661                })
3662            },
3663        );
3664
3665        // After the action is confirmed, an editor containing both modified files is opened.
3666        confirm_action.await.unwrap();
3667        let code_action_editor = workspace_b.read_with(cx_b, |workspace, cx| {
3668            workspace
3669                .active_item(cx)
3670                .unwrap()
3671                .downcast::<Editor>()
3672                .unwrap()
3673        });
3674        code_action_editor.update(cx_b, |editor, cx| {
3675            assert_eq!(editor.text(cx), "\nmod other;\nfn main() { let foo = 4; }");
3676            editor.undo(&Undo, cx);
3677            assert_eq!(
3678                editor.text(cx),
3679                "pub fn foo() -> usize { 4 }\nmod other;\nfn main() { let foo = other::foo(); }"
3680            );
3681            editor.redo(&Redo, cx);
3682            assert_eq!(editor.text(cx), "\nmod other;\nfn main() { let foo = 4; }");
3683        });
3684    }
3685
3686    #[gpui::test(iterations = 10)]
3687    async fn test_collaborating_with_renames(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3688        cx_a.foreground().forbid_parking();
3689        let lang_registry = Arc::new(LanguageRegistry::test());
3690        let fs = FakeFs::new(cx_a.background());
3691        cx_b.update(|cx| editor::init(cx));
3692
3693        // Set up a fake language server.
3694        let mut language = Language::new(
3695            LanguageConfig {
3696                name: "Rust".into(),
3697                path_suffixes: vec!["rs".to_string()],
3698                ..Default::default()
3699            },
3700            Some(tree_sitter_rust::language()),
3701        );
3702        let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3703        lang_registry.add(Arc::new(language));
3704
3705        // Connect to a server as 2 clients.
3706        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3707        let client_a = server.create_client(cx_a, "user_a").await;
3708        let client_b = server.create_client(cx_b, "user_b").await;
3709
3710        // Share a project as client A
3711        fs.insert_tree(
3712            "/dir",
3713            json!({
3714                ".zed.toml": r#"collaborators = ["user_b"]"#,
3715                "one.rs": "const ONE: usize = 1;",
3716                "two.rs": "const TWO: usize = one::ONE + one::ONE;"
3717            }),
3718        )
3719        .await;
3720        let project_a = cx_a.update(|cx| {
3721            Project::local(
3722                client_a.clone(),
3723                client_a.user_store.clone(),
3724                lang_registry.clone(),
3725                fs.clone(),
3726                cx,
3727            )
3728        });
3729        let (worktree_a, _) = project_a
3730            .update(cx_a, |p, cx| {
3731                p.find_or_create_local_worktree("/dir", true, cx)
3732            })
3733            .await
3734            .unwrap();
3735        worktree_a
3736            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3737            .await;
3738        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
3739        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3740        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
3741
3742        // Join the worktree as client B.
3743        let project_b = Project::remote(
3744            project_id,
3745            client_b.clone(),
3746            client_b.user_store.clone(),
3747            lang_registry.clone(),
3748            fs.clone(),
3749            &mut cx_b.to_async(),
3750        )
3751        .await
3752        .unwrap();
3753        let mut params = cx_b.update(WorkspaceParams::test);
3754        params.languages = lang_registry.clone();
3755        params.client = client_b.client.clone();
3756        params.user_store = client_b.user_store.clone();
3757        params.project = project_b;
3758
3759        let (_window_b, workspace_b) = cx_b.add_window(|cx| Workspace::new(&params, cx));
3760        let editor_b = workspace_b
3761            .update(cx_b, |workspace, cx| {
3762                workspace.open_path((worktree_id, "one.rs"), cx)
3763            })
3764            .await
3765            .unwrap()
3766            .downcast::<Editor>()
3767            .unwrap();
3768        let fake_language_server = fake_language_servers.next().await.unwrap();
3769
3770        // Move cursor to a location that can be renamed.
3771        let prepare_rename = editor_b.update(cx_b, |editor, cx| {
3772            editor.select_ranges([7..7], None, cx);
3773            editor.rename(&Rename, cx).unwrap()
3774        });
3775
3776        fake_language_server
3777            .handle_request::<lsp::request::PrepareRenameRequest, _, _>(|params, _| async move {
3778                assert_eq!(params.text_document.uri.as_str(), "file:///dir/one.rs");
3779                assert_eq!(params.position, lsp::Position::new(0, 7));
3780                Ok(Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
3781                    lsp::Position::new(0, 6),
3782                    lsp::Position::new(0, 9),
3783                ))))
3784            })
3785            .next()
3786            .await
3787            .unwrap();
3788        prepare_rename.await.unwrap();
3789        editor_b.update(cx_b, |editor, cx| {
3790            let rename = editor.pending_rename().unwrap();
3791            let buffer = editor.buffer().read(cx).snapshot(cx);
3792            assert_eq!(
3793                rename.range.start.to_offset(&buffer)..rename.range.end.to_offset(&buffer),
3794                6..9
3795            );
3796            rename.editor.update(cx, |rename_editor, cx| {
3797                rename_editor.buffer().update(cx, |rename_buffer, cx| {
3798                    rename_buffer.edit([0..3], "THREE", cx);
3799                });
3800            });
3801        });
3802
3803        let confirm_rename = workspace_b.update(cx_b, |workspace, cx| {
3804            Editor::confirm_rename(workspace, &ConfirmRename, cx).unwrap()
3805        });
3806        fake_language_server
3807            .handle_request::<lsp::request::Rename, _, _>(|params, _| async move {
3808                assert_eq!(
3809                    params.text_document_position.text_document.uri.as_str(),
3810                    "file:///dir/one.rs"
3811                );
3812                assert_eq!(
3813                    params.text_document_position.position,
3814                    lsp::Position::new(0, 6)
3815                );
3816                assert_eq!(params.new_name, "THREE");
3817                Ok(Some(lsp::WorkspaceEdit {
3818                    changes: Some(
3819                        [
3820                            (
3821                                lsp::Url::from_file_path("/dir/one.rs").unwrap(),
3822                                vec![lsp::TextEdit::new(
3823                                    lsp::Range::new(
3824                                        lsp::Position::new(0, 6),
3825                                        lsp::Position::new(0, 9),
3826                                    ),
3827                                    "THREE".to_string(),
3828                                )],
3829                            ),
3830                            (
3831                                lsp::Url::from_file_path("/dir/two.rs").unwrap(),
3832                                vec![
3833                                    lsp::TextEdit::new(
3834                                        lsp::Range::new(
3835                                            lsp::Position::new(0, 24),
3836                                            lsp::Position::new(0, 27),
3837                                        ),
3838                                        "THREE".to_string(),
3839                                    ),
3840                                    lsp::TextEdit::new(
3841                                        lsp::Range::new(
3842                                            lsp::Position::new(0, 35),
3843                                            lsp::Position::new(0, 38),
3844                                        ),
3845                                        "THREE".to_string(),
3846                                    ),
3847                                ],
3848                            ),
3849                        ]
3850                        .into_iter()
3851                        .collect(),
3852                    ),
3853                    ..Default::default()
3854                }))
3855            })
3856            .next()
3857            .await
3858            .unwrap();
3859        confirm_rename.await.unwrap();
3860
3861        let rename_editor = workspace_b.read_with(cx_b, |workspace, cx| {
3862            workspace
3863                .active_item(cx)
3864                .unwrap()
3865                .downcast::<Editor>()
3866                .unwrap()
3867        });
3868        rename_editor.update(cx_b, |editor, cx| {
3869            assert_eq!(
3870                editor.text(cx),
3871                "const TWO: usize = one::THREE + one::THREE;\nconst THREE: usize = 1;"
3872            );
3873            editor.undo(&Undo, cx);
3874            assert_eq!(
3875                editor.text(cx),
3876                "const TWO: usize = one::ONE + one::ONE;\nconst ONE: usize = 1;"
3877            );
3878            editor.redo(&Redo, cx);
3879            assert_eq!(
3880                editor.text(cx),
3881                "const TWO: usize = one::THREE + one::THREE;\nconst THREE: usize = 1;"
3882            );
3883        });
3884
3885        // Ensure temporary rename edits cannot be undone/redone.
3886        editor_b.update(cx_b, |editor, cx| {
3887            editor.undo(&Undo, cx);
3888            assert_eq!(editor.text(cx), "const ONE: usize = 1;");
3889            editor.undo(&Undo, cx);
3890            assert_eq!(editor.text(cx), "const ONE: usize = 1;");
3891            editor.redo(&Redo, cx);
3892            assert_eq!(editor.text(cx), "const THREE: usize = 1;");
3893        })
3894    }
3895
3896    #[gpui::test(iterations = 10)]
3897    async fn test_basic_chat(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3898        cx_a.foreground().forbid_parking();
3899
3900        // Connect to a server as 2 clients.
3901        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3902        let client_a = server.create_client(cx_a, "user_a").await;
3903        let client_b = server.create_client(cx_b, "user_b").await;
3904
3905        // Create an org that includes these 2 users.
3906        let db = &server.app_state.db;
3907        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
3908        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
3909            .await
3910            .unwrap();
3911        db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
3912            .await
3913            .unwrap();
3914
3915        // Create a channel that includes all the users.
3916        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
3917        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
3918            .await
3919            .unwrap();
3920        db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
3921            .await
3922            .unwrap();
3923        db.create_channel_message(
3924            channel_id,
3925            client_b.current_user_id(&cx_b),
3926            "hello A, it's B.",
3927            OffsetDateTime::now_utc(),
3928            1,
3929        )
3930        .await
3931        .unwrap();
3932
3933        let channels_a = cx_a
3934            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
3935        channels_a
3936            .condition(cx_a, |list, _| list.available_channels().is_some())
3937            .await;
3938        channels_a.read_with(cx_a, |list, _| {
3939            assert_eq!(
3940                list.available_channels().unwrap(),
3941                &[ChannelDetails {
3942                    id: channel_id.to_proto(),
3943                    name: "test-channel".to_string()
3944                }]
3945            )
3946        });
3947        let channel_a = channels_a.update(cx_a, |this, cx| {
3948            this.get_channel(channel_id.to_proto(), cx).unwrap()
3949        });
3950        channel_a.read_with(cx_a, |channel, _| assert!(channel.messages().is_empty()));
3951        channel_a
3952            .condition(&cx_a, |channel, _| {
3953                channel_messages(channel)
3954                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
3955            })
3956            .await;
3957
3958        let channels_b = cx_b
3959            .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
3960        channels_b
3961            .condition(cx_b, |list, _| list.available_channels().is_some())
3962            .await;
3963        channels_b.read_with(cx_b, |list, _| {
3964            assert_eq!(
3965                list.available_channels().unwrap(),
3966                &[ChannelDetails {
3967                    id: channel_id.to_proto(),
3968                    name: "test-channel".to_string()
3969                }]
3970            )
3971        });
3972
3973        let channel_b = channels_b.update(cx_b, |this, cx| {
3974            this.get_channel(channel_id.to_proto(), cx).unwrap()
3975        });
3976        channel_b.read_with(cx_b, |channel, _| assert!(channel.messages().is_empty()));
3977        channel_b
3978            .condition(&cx_b, |channel, _| {
3979                channel_messages(channel)
3980                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
3981            })
3982            .await;
3983
3984        channel_a
3985            .update(cx_a, |channel, cx| {
3986                channel
3987                    .send_message("oh, hi B.".to_string(), cx)
3988                    .unwrap()
3989                    .detach();
3990                let task = channel.send_message("sup".to_string(), cx).unwrap();
3991                assert_eq!(
3992                    channel_messages(channel),
3993                    &[
3994                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
3995                        ("user_a".to_string(), "oh, hi B.".to_string(), true),
3996                        ("user_a".to_string(), "sup".to_string(), true)
3997                    ]
3998                );
3999                task
4000            })
4001            .await
4002            .unwrap();
4003
4004        channel_b
4005            .condition(&cx_b, |channel, _| {
4006                channel_messages(channel)
4007                    == [
4008                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4009                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4010                        ("user_a".to_string(), "sup".to_string(), false),
4011                    ]
4012            })
4013            .await;
4014
4015        assert_eq!(
4016            server
4017                .state()
4018                .await
4019                .channel(channel_id)
4020                .unwrap()
4021                .connection_ids
4022                .len(),
4023            2
4024        );
4025        cx_b.update(|_| drop(channel_b));
4026        server
4027            .condition(|state| state.channel(channel_id).unwrap().connection_ids.len() == 1)
4028            .await;
4029
4030        cx_a.update(|_| drop(channel_a));
4031        server
4032            .condition(|state| state.channel(channel_id).is_none())
4033            .await;
4034    }
4035
4036    #[gpui::test(iterations = 10)]
4037    async fn test_chat_message_validation(cx_a: &mut TestAppContext) {
4038        cx_a.foreground().forbid_parking();
4039
4040        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4041        let client_a = server.create_client(cx_a, "user_a").await;
4042
4043        let db = &server.app_state.db;
4044        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
4045        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
4046        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
4047            .await
4048            .unwrap();
4049        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
4050            .await
4051            .unwrap();
4052
4053        let channels_a = cx_a
4054            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
4055        channels_a
4056            .condition(cx_a, |list, _| list.available_channels().is_some())
4057            .await;
4058        let channel_a = channels_a.update(cx_a, |this, cx| {
4059            this.get_channel(channel_id.to_proto(), cx).unwrap()
4060        });
4061
4062        // Messages aren't allowed to be too long.
4063        channel_a
4064            .update(cx_a, |channel, cx| {
4065                let long_body = "this is long.\n".repeat(1024);
4066                channel.send_message(long_body, cx).unwrap()
4067            })
4068            .await
4069            .unwrap_err();
4070
4071        // Messages aren't allowed to be blank.
4072        channel_a.update(cx_a, |channel, cx| {
4073            channel.send_message(String::new(), cx).unwrap_err()
4074        });
4075
4076        // Leading and trailing whitespace are trimmed.
4077        channel_a
4078            .update(cx_a, |channel, cx| {
4079                channel
4080                    .send_message("\n surrounded by whitespace  \n".to_string(), cx)
4081                    .unwrap()
4082            })
4083            .await
4084            .unwrap();
4085        assert_eq!(
4086            db.get_channel_messages(channel_id, 10, None)
4087                .await
4088                .unwrap()
4089                .iter()
4090                .map(|m| &m.body)
4091                .collect::<Vec<_>>(),
4092            &["surrounded by whitespace"]
4093        );
4094    }
4095
4096    #[gpui::test(iterations = 10)]
4097    async fn test_chat_reconnection(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4098        cx_a.foreground().forbid_parking();
4099
4100        // Connect to a server as 2 clients.
4101        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4102        let client_a = server.create_client(cx_a, "user_a").await;
4103        let client_b = server.create_client(cx_b, "user_b").await;
4104        let mut status_b = client_b.status();
4105
4106        // Create an org that includes these 2 users.
4107        let db = &server.app_state.db;
4108        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
4109        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
4110            .await
4111            .unwrap();
4112        db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
4113            .await
4114            .unwrap();
4115
4116        // Create a channel that includes all the users.
4117        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
4118        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
4119            .await
4120            .unwrap();
4121        db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
4122            .await
4123            .unwrap();
4124        db.create_channel_message(
4125            channel_id,
4126            client_b.current_user_id(&cx_b),
4127            "hello A, it's B.",
4128            OffsetDateTime::now_utc(),
4129            2,
4130        )
4131        .await
4132        .unwrap();
4133
4134        let channels_a = cx_a
4135            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
4136        channels_a
4137            .condition(cx_a, |list, _| list.available_channels().is_some())
4138            .await;
4139
4140        channels_a.read_with(cx_a, |list, _| {
4141            assert_eq!(
4142                list.available_channels().unwrap(),
4143                &[ChannelDetails {
4144                    id: channel_id.to_proto(),
4145                    name: "test-channel".to_string()
4146                }]
4147            )
4148        });
4149        let channel_a = channels_a.update(cx_a, |this, cx| {
4150            this.get_channel(channel_id.to_proto(), cx).unwrap()
4151        });
4152        channel_a.read_with(cx_a, |channel, _| assert!(channel.messages().is_empty()));
4153        channel_a
4154            .condition(&cx_a, |channel, _| {
4155                channel_messages(channel)
4156                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4157            })
4158            .await;
4159
4160        let channels_b = cx_b
4161            .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
4162        channels_b
4163            .condition(cx_b, |list, _| list.available_channels().is_some())
4164            .await;
4165        channels_b.read_with(cx_b, |list, _| {
4166            assert_eq!(
4167                list.available_channels().unwrap(),
4168                &[ChannelDetails {
4169                    id: channel_id.to_proto(),
4170                    name: "test-channel".to_string()
4171                }]
4172            )
4173        });
4174
4175        let channel_b = channels_b.update(cx_b, |this, cx| {
4176            this.get_channel(channel_id.to_proto(), cx).unwrap()
4177        });
4178        channel_b.read_with(cx_b, |channel, _| assert!(channel.messages().is_empty()));
4179        channel_b
4180            .condition(&cx_b, |channel, _| {
4181                channel_messages(channel)
4182                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4183            })
4184            .await;
4185
4186        // Disconnect client B, ensuring we can still access its cached channel data.
4187        server.forbid_connections();
4188        server.disconnect_client(client_b.current_user_id(&cx_b));
4189        cx_b.foreground().advance_clock(Duration::from_secs(3));
4190        while !matches!(
4191            status_b.next().await,
4192            Some(client::Status::ReconnectionError { .. })
4193        ) {}
4194
4195        channels_b.read_with(cx_b, |channels, _| {
4196            assert_eq!(
4197                channels.available_channels().unwrap(),
4198                [ChannelDetails {
4199                    id: channel_id.to_proto(),
4200                    name: "test-channel".to_string()
4201                }]
4202            )
4203        });
4204        channel_b.read_with(cx_b, |channel, _| {
4205            assert_eq!(
4206                channel_messages(channel),
4207                [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4208            )
4209        });
4210
4211        // Send a message from client B while it is disconnected.
4212        channel_b
4213            .update(cx_b, |channel, cx| {
4214                let task = channel
4215                    .send_message("can you see this?".to_string(), cx)
4216                    .unwrap();
4217                assert_eq!(
4218                    channel_messages(channel),
4219                    &[
4220                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4221                        ("user_b".to_string(), "can you see this?".to_string(), true)
4222                    ]
4223                );
4224                task
4225            })
4226            .await
4227            .unwrap_err();
4228
4229        // Send a message from client A while B is disconnected.
4230        channel_a
4231            .update(cx_a, |channel, cx| {
4232                channel
4233                    .send_message("oh, hi B.".to_string(), cx)
4234                    .unwrap()
4235                    .detach();
4236                let task = channel.send_message("sup".to_string(), cx).unwrap();
4237                assert_eq!(
4238                    channel_messages(channel),
4239                    &[
4240                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4241                        ("user_a".to_string(), "oh, hi B.".to_string(), true),
4242                        ("user_a".to_string(), "sup".to_string(), true)
4243                    ]
4244                );
4245                task
4246            })
4247            .await
4248            .unwrap();
4249
4250        // Give client B a chance to reconnect.
4251        server.allow_connections();
4252        cx_b.foreground().advance_clock(Duration::from_secs(10));
4253
4254        // Verify that B sees the new messages upon reconnection, as well as the message client B
4255        // sent while offline.
4256        channel_b
4257            .condition(&cx_b, |channel, _| {
4258                channel_messages(channel)
4259                    == [
4260                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4261                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4262                        ("user_a".to_string(), "sup".to_string(), false),
4263                        ("user_b".to_string(), "can you see this?".to_string(), false),
4264                    ]
4265            })
4266            .await;
4267
4268        // Ensure client A and B can communicate normally after reconnection.
4269        channel_a
4270            .update(cx_a, |channel, cx| {
4271                channel.send_message("you online?".to_string(), cx).unwrap()
4272            })
4273            .await
4274            .unwrap();
4275        channel_b
4276            .condition(&cx_b, |channel, _| {
4277                channel_messages(channel)
4278                    == [
4279                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4280                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4281                        ("user_a".to_string(), "sup".to_string(), false),
4282                        ("user_b".to_string(), "can you see this?".to_string(), false),
4283                        ("user_a".to_string(), "you online?".to_string(), false),
4284                    ]
4285            })
4286            .await;
4287
4288        channel_b
4289            .update(cx_b, |channel, cx| {
4290                channel.send_message("yep".to_string(), cx).unwrap()
4291            })
4292            .await
4293            .unwrap();
4294        channel_a
4295            .condition(&cx_a, |channel, _| {
4296                channel_messages(channel)
4297                    == [
4298                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4299                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4300                        ("user_a".to_string(), "sup".to_string(), false),
4301                        ("user_b".to_string(), "can you see this?".to_string(), false),
4302                        ("user_a".to_string(), "you online?".to_string(), false),
4303                        ("user_b".to_string(), "yep".to_string(), false),
4304                    ]
4305            })
4306            .await;
4307    }
4308
4309    #[gpui::test(iterations = 10)]
4310    async fn test_contacts(
4311        cx_a: &mut TestAppContext,
4312        cx_b: &mut TestAppContext,
4313        cx_c: &mut TestAppContext,
4314    ) {
4315        cx_a.foreground().forbid_parking();
4316        let lang_registry = Arc::new(LanguageRegistry::test());
4317        let fs = FakeFs::new(cx_a.background());
4318
4319        // Connect to a server as 3 clients.
4320        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4321        let client_a = server.create_client(cx_a, "user_a").await;
4322        let client_b = server.create_client(cx_b, "user_b").await;
4323        let client_c = server.create_client(cx_c, "user_c").await;
4324
4325        // Share a worktree as client A.
4326        fs.insert_tree(
4327            "/a",
4328            json!({
4329                ".zed.toml": r#"collaborators = ["user_b", "user_c"]"#,
4330            }),
4331        )
4332        .await;
4333
4334        let project_a = cx_a.update(|cx| {
4335            Project::local(
4336                client_a.clone(),
4337                client_a.user_store.clone(),
4338                lang_registry.clone(),
4339                fs.clone(),
4340                cx,
4341            )
4342        });
4343        let (worktree_a, _) = project_a
4344            .update(cx_a, |p, cx| {
4345                p.find_or_create_local_worktree("/a", true, cx)
4346            })
4347            .await
4348            .unwrap();
4349        worktree_a
4350            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
4351            .await;
4352
4353        client_a
4354            .user_store
4355            .condition(&cx_a, |user_store, _| {
4356                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4357            })
4358            .await;
4359        client_b
4360            .user_store
4361            .condition(&cx_b, |user_store, _| {
4362                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4363            })
4364            .await;
4365        client_c
4366            .user_store
4367            .condition(&cx_c, |user_store, _| {
4368                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4369            })
4370            .await;
4371
4372        let project_id = project_a
4373            .update(cx_a, |project, _| project.next_remote_id())
4374            .await;
4375        project_a
4376            .update(cx_a, |project, cx| project.share(cx))
4377            .await
4378            .unwrap();
4379
4380        let _project_b = Project::remote(
4381            project_id,
4382            client_b.clone(),
4383            client_b.user_store.clone(),
4384            lang_registry.clone(),
4385            fs.clone(),
4386            &mut cx_b.to_async(),
4387        )
4388        .await
4389        .unwrap();
4390
4391        client_a
4392            .user_store
4393            .condition(&cx_a, |user_store, _| {
4394                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4395            })
4396            .await;
4397        client_b
4398            .user_store
4399            .condition(&cx_b, |user_store, _| {
4400                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4401            })
4402            .await;
4403        client_c
4404            .user_store
4405            .condition(&cx_c, |user_store, _| {
4406                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4407            })
4408            .await;
4409
4410        project_a
4411            .condition(&cx_a, |project, _| {
4412                project.collaborators().contains_key(&client_b.peer_id)
4413            })
4414            .await;
4415
4416        cx_a.update(move |_| drop(project_a));
4417        client_a
4418            .user_store
4419            .condition(&cx_a, |user_store, _| contacts(user_store) == vec![])
4420            .await;
4421        client_b
4422            .user_store
4423            .condition(&cx_b, |user_store, _| contacts(user_store) == vec![])
4424            .await;
4425        client_c
4426            .user_store
4427            .condition(&cx_c, |user_store, _| contacts(user_store) == vec![])
4428            .await;
4429
4430        fn contacts(user_store: &UserStore) -> Vec<(&str, Vec<(&str, Vec<&str>)>)> {
4431            user_store
4432                .contacts()
4433                .iter()
4434                .map(|contact| {
4435                    let worktrees = contact
4436                        .projects
4437                        .iter()
4438                        .map(|p| {
4439                            (
4440                                p.worktree_root_names[0].as_str(),
4441                                p.guests.iter().map(|p| p.github_login.as_str()).collect(),
4442                            )
4443                        })
4444                        .collect();
4445                    (contact.user.github_login.as_str(), worktrees)
4446                })
4447                .collect()
4448        }
4449    }
4450
4451    #[gpui::test(iterations = 10)]
4452    async fn test_following(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4453        cx_a.foreground().forbid_parking();
4454        let fs = FakeFs::new(cx_a.background());
4455
4456        // 2 clients connect to a server.
4457        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4458        let mut client_a = server.create_client(cx_a, "user_a").await;
4459        let mut client_b = server.create_client(cx_b, "user_b").await;
4460        cx_a.update(editor::init);
4461        cx_b.update(editor::init);
4462
4463        // Client A shares a project.
4464        fs.insert_tree(
4465            "/a",
4466            json!({
4467                ".zed.toml": r#"collaborators = ["user_b"]"#,
4468                "1.txt": "one",
4469                "2.txt": "two",
4470                "3.txt": "three",
4471            }),
4472        )
4473        .await;
4474        let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
4475        project_a
4476            .update(cx_a, |project, cx| project.share(cx))
4477            .await
4478            .unwrap();
4479
4480        // Client B joins the project.
4481        let project_b = client_b
4482            .build_remote_project(
4483                project_a
4484                    .read_with(cx_a, |project, _| project.remote_id())
4485                    .unwrap(),
4486                cx_b,
4487            )
4488            .await;
4489
4490        // Client A opens some editors.
4491        let workspace_a = client_a.build_workspace(&project_a, cx_a);
4492        let pane_a = workspace_a.read_with(cx_a, |workspace, _| workspace.active_pane().clone());
4493        let editor_a1 = workspace_a
4494            .update(cx_a, |workspace, cx| {
4495                workspace.open_path((worktree_id, "1.txt"), cx)
4496            })
4497            .await
4498            .unwrap()
4499            .downcast::<Editor>()
4500            .unwrap();
4501        let editor_a2 = workspace_a
4502            .update(cx_a, |workspace, cx| {
4503                workspace.open_path((worktree_id, "2.txt"), cx)
4504            })
4505            .await
4506            .unwrap()
4507            .downcast::<Editor>()
4508            .unwrap();
4509
4510        // Client B opens an editor.
4511        let workspace_b = client_b.build_workspace(&project_b, cx_b);
4512        let editor_b1 = workspace_b
4513            .update(cx_b, |workspace, cx| {
4514                workspace.open_path((worktree_id, "1.txt"), cx)
4515            })
4516            .await
4517            .unwrap()
4518            .downcast::<Editor>()
4519            .unwrap();
4520
4521        let client_a_id = project_b.read_with(cx_b, |project, _| {
4522            project.collaborators().values().next().unwrap().peer_id
4523        });
4524        let client_b_id = project_a.read_with(cx_a, |project, _| {
4525            project.collaborators().values().next().unwrap().peer_id
4526        });
4527
4528        // When client B starts following client A, all visible view states are replicated to client B.
4529        editor_a1.update(cx_a, |editor, cx| editor.select_ranges([0..1], None, cx));
4530        editor_a2.update(cx_a, |editor, cx| editor.select_ranges([2..3], None, cx));
4531        workspace_b
4532            .update(cx_b, |workspace, cx| {
4533                workspace
4534                    .toggle_follow(&ToggleFollow(client_a_id), cx)
4535                    .unwrap()
4536            })
4537            .await
4538            .unwrap();
4539        let editor_b2 = workspace_b.read_with(cx_b, |workspace, cx| {
4540            workspace
4541                .active_item(cx)
4542                .unwrap()
4543                .downcast::<Editor>()
4544                .unwrap()
4545        });
4546        assert!(cx_b.read(|cx| editor_b2.is_focused(cx)));
4547        assert_eq!(
4548            editor_b2.read_with(cx_b, |editor, cx| editor.project_path(cx)),
4549            Some((worktree_id, "2.txt").into())
4550        );
4551        assert_eq!(
4552            editor_b2.read_with(cx_b, |editor, cx| editor.selected_ranges(cx)),
4553            vec![2..3]
4554        );
4555        assert_eq!(
4556            editor_b1.read_with(cx_b, |editor, cx| editor.selected_ranges(cx)),
4557            vec![0..1]
4558        );
4559
4560        // When client A activates a different editor, client B does so as well.
4561        workspace_a.update(cx_a, |workspace, cx| {
4562            workspace.activate_item(&editor_a1, cx)
4563        });
4564        workspace_b
4565            .condition(cx_b, |workspace, cx| {
4566                workspace.active_item(cx).unwrap().id() == editor_b1.id()
4567            })
4568            .await;
4569
4570        // When client A navigates back and forth, client B does so as well.
4571        workspace_a
4572            .update(cx_a, |workspace, cx| {
4573                workspace::Pane::go_back(workspace, None, cx)
4574            })
4575            .await;
4576        workspace_b
4577            .condition(cx_b, |workspace, cx| {
4578                workspace.active_item(cx).unwrap().id() == editor_b2.id()
4579            })
4580            .await;
4581
4582        workspace_a
4583            .update(cx_a, |workspace, cx| {
4584                workspace::Pane::go_forward(workspace, None, cx)
4585            })
4586            .await;
4587        workspace_b
4588            .condition(cx_b, |workspace, cx| {
4589                workspace.active_item(cx).unwrap().id() == editor_b1.id()
4590            })
4591            .await;
4592
4593        // Changes to client A's editor are reflected on client B.
4594        editor_a1.update(cx_a, |editor, cx| {
4595            editor.select_ranges([1..1, 2..2], None, cx);
4596        });
4597        editor_b1
4598            .condition(cx_b, |editor, cx| {
4599                editor.selected_ranges(cx) == vec![1..1, 2..2]
4600            })
4601            .await;
4602
4603        editor_a1.update(cx_a, |editor, cx| editor.set_text("TWO", cx));
4604        editor_b1
4605            .condition(cx_b, |editor, cx| editor.text(cx) == "TWO")
4606            .await;
4607
4608        editor_a1.update(cx_a, |editor, cx| {
4609            editor.select_ranges([3..3], None, cx);
4610            editor.set_scroll_position(vec2f(0., 100.), cx);
4611        });
4612        editor_b1
4613            .condition(cx_b, |editor, cx| editor.selected_ranges(cx) == vec![3..3])
4614            .await;
4615
4616        // After unfollowing, client B stops receiving updates from client A.
4617        workspace_b.update(cx_b, |workspace, cx| {
4618            workspace.unfollow(&workspace.active_pane().clone(), cx)
4619        });
4620        workspace_a.update(cx_a, |workspace, cx| {
4621            workspace.activate_item(&editor_a2, cx)
4622        });
4623        cx_a.foreground().run_until_parked();
4624        assert_eq!(
4625            workspace_b.read_with(cx_b, |workspace, cx| workspace
4626                .active_item(cx)
4627                .unwrap()
4628                .id()),
4629            editor_b1.id()
4630        );
4631
4632        // Client A starts following client B.
4633        workspace_a
4634            .update(cx_a, |workspace, cx| {
4635                workspace
4636                    .toggle_follow(&ToggleFollow(client_b_id), cx)
4637                    .unwrap()
4638            })
4639            .await
4640            .unwrap();
4641        assert_eq!(
4642            workspace_a.read_with(cx_a, |workspace, _| workspace.leader_for_pane(&pane_a)),
4643            Some(client_b_id)
4644        );
4645        assert_eq!(
4646            workspace_a.read_with(cx_a, |workspace, cx| workspace
4647                .active_item(cx)
4648                .unwrap()
4649                .id()),
4650            editor_a1.id()
4651        );
4652
4653        // Following interrupts when client B disconnects.
4654        client_b.disconnect(&cx_b.to_async()).unwrap();
4655        cx_a.foreground().run_until_parked();
4656        assert_eq!(
4657            workspace_a.read_with(cx_a, |workspace, _| workspace.leader_for_pane(&pane_a)),
4658            None
4659        );
4660    }
4661
4662    #[gpui::test(iterations = 10)]
4663    async fn test_peers_following_each_other(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4664        cx_a.foreground().forbid_parking();
4665        let fs = FakeFs::new(cx_a.background());
4666
4667        // 2 clients connect to a server.
4668        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4669        let mut client_a = server.create_client(cx_a, "user_a").await;
4670        let mut client_b = server.create_client(cx_b, "user_b").await;
4671        cx_a.update(editor::init);
4672        cx_b.update(editor::init);
4673
4674        // Client A shares a project.
4675        fs.insert_tree(
4676            "/a",
4677            json!({
4678                ".zed.toml": r#"collaborators = ["user_b"]"#,
4679                "1.txt": "one",
4680                "2.txt": "two",
4681                "3.txt": "three",
4682                "4.txt": "four",
4683            }),
4684        )
4685        .await;
4686        let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
4687        project_a
4688            .update(cx_a, |project, cx| project.share(cx))
4689            .await
4690            .unwrap();
4691
4692        // Client B joins the project.
4693        let project_b = client_b
4694            .build_remote_project(
4695                project_a
4696                    .read_with(cx_a, |project, _| project.remote_id())
4697                    .unwrap(),
4698                cx_b,
4699            )
4700            .await;
4701
4702        // Client A opens some editors.
4703        let workspace_a = client_a.build_workspace(&project_a, cx_a);
4704        let pane_a1 = workspace_a.read_with(cx_a, |workspace, _| workspace.active_pane().clone());
4705        let _editor_a1 = workspace_a
4706            .update(cx_a, |workspace, cx| {
4707                workspace.open_path((worktree_id, "1.txt"), cx)
4708            })
4709            .await
4710            .unwrap()
4711            .downcast::<Editor>()
4712            .unwrap();
4713
4714        // Client B opens an editor.
4715        let workspace_b = client_b.build_workspace(&project_b, cx_b);
4716        let pane_b1 = workspace_b.read_with(cx_b, |workspace, _| workspace.active_pane().clone());
4717        let _editor_b1 = workspace_b
4718            .update(cx_b, |workspace, cx| {
4719                workspace.open_path((worktree_id, "2.txt"), cx)
4720            })
4721            .await
4722            .unwrap()
4723            .downcast::<Editor>()
4724            .unwrap();
4725
4726        // Clients A and B follow each other in split panes
4727        workspace_a
4728            .update(cx_a, |workspace, cx| {
4729                workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
4730                assert_ne!(*workspace.active_pane(), pane_a1);
4731                let leader_id = *project_a.read(cx).collaborators().keys().next().unwrap();
4732                workspace
4733                    .toggle_follow(&workspace::ToggleFollow(leader_id), cx)
4734                    .unwrap()
4735            })
4736            .await
4737            .unwrap();
4738        workspace_b
4739            .update(cx_b, |workspace, cx| {
4740                workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
4741                assert_ne!(*workspace.active_pane(), pane_b1);
4742                let leader_id = *project_b.read(cx).collaborators().keys().next().unwrap();
4743                workspace
4744                    .toggle_follow(&workspace::ToggleFollow(leader_id), cx)
4745                    .unwrap()
4746            })
4747            .await
4748            .unwrap();
4749
4750        workspace_a
4751            .update(cx_a, |workspace, cx| {
4752                workspace.activate_next_pane(cx);
4753                assert_eq!(*workspace.active_pane(), pane_a1);
4754                workspace.open_path((worktree_id, "3.txt"), cx)
4755            })
4756            .await
4757            .unwrap();
4758        workspace_b
4759            .update(cx_b, |workspace, cx| {
4760                workspace.activate_next_pane(cx);
4761                assert_eq!(*workspace.active_pane(), pane_b1);
4762                workspace.open_path((worktree_id, "4.txt"), cx)
4763            })
4764            .await
4765            .unwrap();
4766        cx_a.foreground().run_until_parked();
4767
4768        // Ensure leader updates don't change the active pane of followers
4769        workspace_a.read_with(cx_a, |workspace, _| {
4770            assert_eq!(*workspace.active_pane(), pane_a1);
4771        });
4772        workspace_b.read_with(cx_b, |workspace, _| {
4773            assert_eq!(*workspace.active_pane(), pane_b1);
4774        });
4775
4776        // Ensure peers following each other doesn't cause an infinite loop.
4777        assert_eq!(
4778            workspace_a.read_with(cx_a, |workspace, cx| workspace
4779                .active_item(cx)
4780                .unwrap()
4781                .project_path(cx)),
4782            Some((worktree_id, "3.txt").into())
4783        );
4784        workspace_a.update(cx_a, |workspace, cx| {
4785            assert_eq!(
4786                workspace.active_item(cx).unwrap().project_path(cx),
4787                Some((worktree_id, "3.txt").into())
4788            );
4789            workspace.activate_next_pane(cx);
4790            assert_eq!(
4791                workspace.active_item(cx).unwrap().project_path(cx),
4792                Some((worktree_id, "4.txt").into())
4793            );
4794        });
4795        workspace_b.update(cx_b, |workspace, cx| {
4796            assert_eq!(
4797                workspace.active_item(cx).unwrap().project_path(cx),
4798                Some((worktree_id, "4.txt").into())
4799            );
4800            workspace.activate_next_pane(cx);
4801            assert_eq!(
4802                workspace.active_item(cx).unwrap().project_path(cx),
4803                Some((worktree_id, "3.txt").into())
4804            );
4805        });
4806    }
4807
4808    #[gpui::test(iterations = 10)]
4809    async fn test_auto_unfollowing(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4810        cx_a.foreground().forbid_parking();
4811        let fs = FakeFs::new(cx_a.background());
4812
4813        // 2 clients connect to a server.
4814        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4815        let mut client_a = server.create_client(cx_a, "user_a").await;
4816        let mut client_b = server.create_client(cx_b, "user_b").await;
4817        cx_a.update(editor::init);
4818        cx_b.update(editor::init);
4819
4820        // Client A shares a project.
4821        fs.insert_tree(
4822            "/a",
4823            json!({
4824                ".zed.toml": r#"collaborators = ["user_b"]"#,
4825                "1.txt": "one",
4826                "2.txt": "two",
4827                "3.txt": "three",
4828            }),
4829        )
4830        .await;
4831        let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
4832        project_a
4833            .update(cx_a, |project, cx| project.share(cx))
4834            .await
4835            .unwrap();
4836
4837        // Client B joins the project.
4838        let project_b = client_b
4839            .build_remote_project(
4840                project_a
4841                    .read_with(cx_a, |project, _| project.remote_id())
4842                    .unwrap(),
4843                cx_b,
4844            )
4845            .await;
4846
4847        // Client A opens some editors.
4848        let workspace_a = client_a.build_workspace(&project_a, cx_a);
4849        let _editor_a1 = workspace_a
4850            .update(cx_a, |workspace, cx| {
4851                workspace.open_path((worktree_id, "1.txt"), cx)
4852            })
4853            .await
4854            .unwrap()
4855            .downcast::<Editor>()
4856            .unwrap();
4857
4858        // Client B starts following client A.
4859        let workspace_b = client_b.build_workspace(&project_b, cx_b);
4860        let pane_b = workspace_b.read_with(cx_b, |workspace, _| workspace.active_pane().clone());
4861        let leader_id = project_b.read_with(cx_b, |project, _| {
4862            project.collaborators().values().next().unwrap().peer_id
4863        });
4864        workspace_b
4865            .update(cx_b, |workspace, cx| {
4866                workspace
4867                    .toggle_follow(&ToggleFollow(leader_id), cx)
4868                    .unwrap()
4869            })
4870            .await
4871            .unwrap();
4872        assert_eq!(
4873            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4874            Some(leader_id)
4875        );
4876        let editor_b2 = workspace_b.read_with(cx_b, |workspace, cx| {
4877            workspace
4878                .active_item(cx)
4879                .unwrap()
4880                .downcast::<Editor>()
4881                .unwrap()
4882        });
4883
4884        // When client B moves, it automatically stops following client A.
4885        editor_b2.update(cx_b, |editor, cx| editor.move_right(&editor::MoveRight, cx));
4886        assert_eq!(
4887            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4888            None
4889        );
4890
4891        workspace_b
4892            .update(cx_b, |workspace, cx| {
4893                workspace
4894                    .toggle_follow(&ToggleFollow(leader_id), cx)
4895                    .unwrap()
4896            })
4897            .await
4898            .unwrap();
4899        assert_eq!(
4900            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4901            Some(leader_id)
4902        );
4903
4904        // When client B edits, it automatically stops following client A.
4905        editor_b2.update(cx_b, |editor, cx| editor.insert("X", cx));
4906        assert_eq!(
4907            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4908            None
4909        );
4910
4911        workspace_b
4912            .update(cx_b, |workspace, cx| {
4913                workspace
4914                    .toggle_follow(&ToggleFollow(leader_id), cx)
4915                    .unwrap()
4916            })
4917            .await
4918            .unwrap();
4919        assert_eq!(
4920            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4921            Some(leader_id)
4922        );
4923
4924        // When client B scrolls, it automatically stops following client A.
4925        editor_b2.update(cx_b, |editor, cx| {
4926            editor.set_scroll_position(vec2f(0., 3.), cx)
4927        });
4928        assert_eq!(
4929            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4930            None
4931        );
4932
4933        workspace_b
4934            .update(cx_b, |workspace, cx| {
4935                workspace
4936                    .toggle_follow(&ToggleFollow(leader_id), cx)
4937                    .unwrap()
4938            })
4939            .await
4940            .unwrap();
4941        assert_eq!(
4942            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4943            Some(leader_id)
4944        );
4945
4946        // When client B activates a different pane, it continues following client A in the original pane.
4947        workspace_b.update(cx_b, |workspace, cx| {
4948            workspace.split_pane(pane_b.clone(), SplitDirection::Right, cx)
4949        });
4950        assert_eq!(
4951            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4952            Some(leader_id)
4953        );
4954
4955        workspace_b.update(cx_b, |workspace, cx| workspace.activate_next_pane(cx));
4956        assert_eq!(
4957            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4958            Some(leader_id)
4959        );
4960
4961        // When client B activates a different item in the original pane, it automatically stops following client A.
4962        workspace_b
4963            .update(cx_b, |workspace, cx| {
4964                workspace.open_path((worktree_id, "2.txt"), cx)
4965            })
4966            .await
4967            .unwrap();
4968        assert_eq!(
4969            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4970            None
4971        );
4972    }
4973
4974    #[gpui::test(iterations = 100)]
4975    async fn test_random_collaboration(
4976        cx: &mut TestAppContext,
4977        deterministic: Arc<Deterministic>,
4978        rng: StdRng,
4979    ) {
4980        cx.foreground().forbid_parking();
4981        let max_peers = env::var("MAX_PEERS")
4982            .map(|i| i.parse().expect("invalid `MAX_PEERS` variable"))
4983            .unwrap_or(5);
4984        let max_operations = env::var("OPERATIONS")
4985            .map(|i| i.parse().expect("invalid `OPERATIONS` variable"))
4986            .unwrap_or(10);
4987
4988        let rng = Arc::new(Mutex::new(rng));
4989
4990        let guest_lang_registry = Arc::new(LanguageRegistry::test());
4991        let host_language_registry = Arc::new(LanguageRegistry::test());
4992
4993        let fs = FakeFs::new(cx.background());
4994        fs.insert_tree(
4995            "/_collab",
4996            json!({
4997                ".zed.toml": r#"collaborators = ["guest-1", "guest-2", "guest-3", "guest-4", "guest-5"]"#
4998            }),
4999        )
5000        .await;
5001
5002        let mut server = TestServer::start(cx.foreground(), cx.background()).await;
5003        let mut clients = Vec::new();
5004        let mut user_ids = Vec::new();
5005        let mut op_start_signals = Vec::new();
5006        let files = Arc::new(Mutex::new(Vec::new()));
5007
5008        let mut next_entity_id = 100000;
5009        let mut host_cx = TestAppContext::new(
5010            cx.foreground_platform(),
5011            cx.platform(),
5012            deterministic.build_foreground(next_entity_id),
5013            deterministic.build_background(),
5014            cx.font_cache(),
5015            cx.leak_detector(),
5016            next_entity_id,
5017        );
5018        let host = server.create_client(&mut host_cx, "host").await;
5019        let host_project = host_cx.update(|cx| {
5020            Project::local(
5021                host.client.clone(),
5022                host.user_store.clone(),
5023                host_language_registry.clone(),
5024                fs.clone(),
5025                cx,
5026            )
5027        });
5028        let host_project_id = host_project
5029            .update(&mut host_cx, |p, _| p.next_remote_id())
5030            .await;
5031
5032        let (collab_worktree, _) = host_project
5033            .update(&mut host_cx, |project, cx| {
5034                project.find_or_create_local_worktree("/_collab", true, cx)
5035            })
5036            .await
5037            .unwrap();
5038        collab_worktree
5039            .read_with(&host_cx, |tree, _| tree.as_local().unwrap().scan_complete())
5040            .await;
5041        host_project
5042            .update(&mut host_cx, |project, cx| project.share(cx))
5043            .await
5044            .unwrap();
5045
5046        // Set up fake language servers.
5047        let mut language = Language::new(
5048            LanguageConfig {
5049                name: "Rust".into(),
5050                path_suffixes: vec!["rs".to_string()],
5051                ..Default::default()
5052            },
5053            None,
5054        );
5055        let _fake_servers = language.set_fake_lsp_adapter(FakeLspAdapter {
5056            name: "the-fake-language-server",
5057            capabilities: lsp::LanguageServer::full_capabilities(),
5058            initializer: Some(Box::new({
5059                let rng = rng.clone();
5060                let files = files.clone();
5061                let project = host_project.downgrade();
5062                move |fake_server: &mut FakeLanguageServer| {
5063                    fake_server.handle_request::<lsp::request::Completion, _, _>(
5064                        |_, _| async move {
5065                            Ok(Some(lsp::CompletionResponse::Array(vec![
5066                                lsp::CompletionItem {
5067                                    text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
5068                                        range: lsp::Range::new(
5069                                            lsp::Position::new(0, 0),
5070                                            lsp::Position::new(0, 0),
5071                                        ),
5072                                        new_text: "the-new-text".to_string(),
5073                                    })),
5074                                    ..Default::default()
5075                                },
5076                            ])))
5077                        },
5078                    );
5079
5080                    fake_server.handle_request::<lsp::request::CodeActionRequest, _, _>(
5081                        |_, _| async move {
5082                            Ok(Some(vec![lsp::CodeActionOrCommand::CodeAction(
5083                                lsp::CodeAction {
5084                                    title: "the-code-action".to_string(),
5085                                    ..Default::default()
5086                                },
5087                            )]))
5088                        },
5089                    );
5090
5091                    fake_server.handle_request::<lsp::request::PrepareRenameRequest, _, _>(
5092                        |params, _| async move {
5093                            Ok(Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
5094                                params.position,
5095                                params.position,
5096                            ))))
5097                        },
5098                    );
5099
5100                    fake_server.handle_request::<lsp::request::GotoDefinition, _, _>({
5101                        let files = files.clone();
5102                        let rng = rng.clone();
5103                        move |_, _| {
5104                            let files = files.clone();
5105                            let rng = rng.clone();
5106                            async move {
5107                                let files = files.lock();
5108                                let mut rng = rng.lock();
5109                                let count = rng.gen_range::<usize, _>(1..3);
5110                                let files = (0..count)
5111                                    .map(|_| files.choose(&mut *rng).unwrap())
5112                                    .collect::<Vec<_>>();
5113                                log::info!("LSP: Returning definitions in files {:?}", &files);
5114                                Ok(Some(lsp::GotoDefinitionResponse::Array(
5115                                    files
5116                                        .into_iter()
5117                                        .map(|file| lsp::Location {
5118                                            uri: lsp::Url::from_file_path(file).unwrap(),
5119                                            range: Default::default(),
5120                                        })
5121                                        .collect(),
5122                                )))
5123                            }
5124                        }
5125                    });
5126
5127                    fake_server.handle_request::<lsp::request::DocumentHighlightRequest, _, _>({
5128                        let rng = rng.clone();
5129                        let project = project.clone();
5130                        move |params, mut cx| {
5131                            let highlights = if let Some(project) = project.upgrade(&cx) {
5132                                project.update(&mut cx, |project, cx| {
5133                                    let path = params
5134                                        .text_document_position_params
5135                                        .text_document
5136                                        .uri
5137                                        .to_file_path()
5138                                        .unwrap();
5139                                    let (worktree, relative_path) =
5140                                        project.find_local_worktree(&path, cx)?;
5141                                    let project_path =
5142                                        ProjectPath::from((worktree.read(cx).id(), relative_path));
5143                                    let buffer =
5144                                        project.get_open_buffer(&project_path, cx)?.read(cx);
5145
5146                                    let mut highlights = Vec::new();
5147                                    let highlight_count = rng.lock().gen_range(1..=5);
5148                                    let mut prev_end = 0;
5149                                    for _ in 0..highlight_count {
5150                                        let range =
5151                                            buffer.random_byte_range(prev_end, &mut *rng.lock());
5152
5153                                        highlights.push(lsp::DocumentHighlight {
5154                                            range: range_to_lsp(range.to_point_utf16(buffer)),
5155                                            kind: Some(lsp::DocumentHighlightKind::READ),
5156                                        });
5157                                        prev_end = range.end;
5158                                    }
5159                                    Some(highlights)
5160                                })
5161                            } else {
5162                                None
5163                            };
5164                            async move { Ok(highlights) }
5165                        }
5166                    });
5167                }
5168            })),
5169            ..Default::default()
5170        });
5171        host_language_registry.add(Arc::new(language));
5172
5173        let host_disconnected = Rc::new(AtomicBool::new(false));
5174        let op_start_signal = futures::channel::mpsc::unbounded();
5175        user_ids.push(host.current_user_id(&host_cx));
5176        op_start_signals.push(op_start_signal.0);
5177        clients.push(host_cx.foreground().spawn(host.simulate_host(
5178            host_project,
5179            files,
5180            op_start_signal.1,
5181            rng.clone(),
5182            host_cx,
5183        )));
5184
5185        let disconnect_host_at = if rng.lock().gen_bool(0.2) {
5186            rng.lock().gen_range(0..max_operations)
5187        } else {
5188            max_operations
5189        };
5190        let mut operations = 0;
5191        while operations < max_operations {
5192            if operations == disconnect_host_at {
5193                host_disconnected.store(true, SeqCst);
5194                server.disconnect_client(user_ids[0]);
5195                cx.foreground().advance_clock(RECEIVE_TIMEOUT);
5196                drop(op_start_signals);
5197                let mut clients = futures::future::join_all(clients).await;
5198                cx.foreground().run_until_parked();
5199
5200                let (host, mut host_cx) = clients.remove(0);
5201                host.project
5202                    .as_ref()
5203                    .unwrap()
5204                    .read_with(&host_cx, |project, _| assert!(!project.is_shared()));
5205                for (guest, mut guest_cx) in clients {
5206                    let contacts = server
5207                        .store
5208                        .read()
5209                        .contacts_for_user(guest.current_user_id(&guest_cx));
5210                    assert!(!contacts
5211                        .iter()
5212                        .flat_map(|contact| &contact.projects)
5213                        .any(|project| project.id == host_project_id));
5214                    guest
5215                        .project
5216                        .as_ref()
5217                        .unwrap()
5218                        .read_with(&guest_cx, |project, _| assert!(project.is_read_only()));
5219                    guest_cx.update(|_| drop(guest));
5220                }
5221                host_cx.update(|_| drop(host));
5222
5223                return;
5224            }
5225
5226            let distribution = rng.lock().gen_range(0..100);
5227            match distribution {
5228                0..=19 if clients.len() < max_peers => {
5229                    let guest_id = clients.len();
5230                    log::info!("Adding guest {}", guest_id);
5231                    next_entity_id += 100000;
5232                    let mut guest_cx = TestAppContext::new(
5233                        cx.foreground_platform(),
5234                        cx.platform(),
5235                        deterministic.build_foreground(next_entity_id),
5236                        deterministic.build_background(),
5237                        cx.font_cache(),
5238                        cx.leak_detector(),
5239                        next_entity_id,
5240                    );
5241                    let guest = server
5242                        .create_client(&mut guest_cx, &format!("guest-{}", guest_id))
5243                        .await;
5244                    let guest_project = Project::remote(
5245                        host_project_id,
5246                        guest.client.clone(),
5247                        guest.user_store.clone(),
5248                        guest_lang_registry.clone(),
5249                        FakeFs::new(cx.background()),
5250                        &mut guest_cx.to_async(),
5251                    )
5252                    .await
5253                    .unwrap();
5254                    let op_start_signal = futures::channel::mpsc::unbounded();
5255                    user_ids.push(guest.current_user_id(&guest_cx));
5256                    op_start_signals.push(op_start_signal.0);
5257                    clients.push(guest_cx.foreground().spawn(guest.simulate_guest(
5258                        guest_id,
5259                        guest_project,
5260                        op_start_signal.1,
5261                        rng.clone(),
5262                        host_disconnected.clone(),
5263                        guest_cx,
5264                    )));
5265
5266                    log::info!("Guest {} added", guest_id);
5267                    operations += 1;
5268                }
5269                _ => {
5270                    while operations < max_operations && rng.lock().gen_bool(0.7) {
5271                        op_start_signals
5272                            .choose(&mut *rng.lock())
5273                            .unwrap()
5274                            .unbounded_send(())
5275                            .unwrap();
5276                        operations += 1;
5277                    }
5278
5279                    if rng.lock().gen_bool(0.8) {
5280                        cx.foreground().run_until_parked();
5281                    }
5282                }
5283            }
5284        }
5285
5286        drop(op_start_signals);
5287        let mut clients = futures::future::join_all(clients).await;
5288        cx.foreground().run_until_parked();
5289
5290        let (host_client, mut host_cx) = clients.remove(0);
5291        let host_project = host_client.project.as_ref().unwrap();
5292        let host_worktree_snapshots = host_project.read_with(&host_cx, |project, cx| {
5293            project
5294                .worktrees(cx)
5295                .map(|worktree| {
5296                    let snapshot = worktree.read(cx).snapshot();
5297                    (snapshot.id(), snapshot)
5298                })
5299                .collect::<BTreeMap<_, _>>()
5300        });
5301
5302        host_client
5303            .project
5304            .as_ref()
5305            .unwrap()
5306            .read_with(&host_cx, |project, cx| project.check_invariants(cx));
5307
5308        for (guest_client, mut guest_cx) in clients.into_iter() {
5309            let guest_id = guest_client.client.id();
5310            let worktree_snapshots =
5311                guest_client
5312                    .project
5313                    .as_ref()
5314                    .unwrap()
5315                    .read_with(&guest_cx, |project, cx| {
5316                        project
5317                            .worktrees(cx)
5318                            .map(|worktree| {
5319                                let worktree = worktree.read(cx);
5320                                (worktree.id(), worktree.snapshot())
5321                            })
5322                            .collect::<BTreeMap<_, _>>()
5323                    });
5324
5325            assert_eq!(
5326                worktree_snapshots.keys().collect::<Vec<_>>(),
5327                host_worktree_snapshots.keys().collect::<Vec<_>>(),
5328                "guest {} has different worktrees than the host",
5329                guest_id
5330            );
5331            for (id, host_snapshot) in &host_worktree_snapshots {
5332                let guest_snapshot = &worktree_snapshots[id];
5333                assert_eq!(
5334                    guest_snapshot.root_name(),
5335                    host_snapshot.root_name(),
5336                    "guest {} has different root name than the host for worktree {}",
5337                    guest_id,
5338                    id
5339                );
5340                assert_eq!(
5341                    guest_snapshot.entries(false).collect::<Vec<_>>(),
5342                    host_snapshot.entries(false).collect::<Vec<_>>(),
5343                    "guest {} has different snapshot than the host for worktree {}",
5344                    guest_id,
5345                    id
5346                );
5347            }
5348
5349            guest_client
5350                .project
5351                .as_ref()
5352                .unwrap()
5353                .read_with(&guest_cx, |project, cx| project.check_invariants(cx));
5354
5355            for guest_buffer in &guest_client.buffers {
5356                let buffer_id = guest_buffer.read_with(&guest_cx, |buffer, _| buffer.remote_id());
5357                let host_buffer = host_project.read_with(&host_cx, |project, cx| {
5358                    project.buffer_for_id(buffer_id, cx).expect(&format!(
5359                        "host does not have buffer for guest:{}, peer:{}, id:{}",
5360                        guest_id, guest_client.peer_id, buffer_id
5361                    ))
5362                });
5363                let path = host_buffer
5364                    .read_with(&host_cx, |buffer, cx| buffer.file().unwrap().full_path(cx));
5365
5366                assert_eq!(
5367                    guest_buffer.read_with(&guest_cx, |buffer, _| buffer.deferred_ops_len()),
5368                    0,
5369                    "guest {}, buffer {}, path {:?} has deferred operations",
5370                    guest_id,
5371                    buffer_id,
5372                    path,
5373                );
5374                assert_eq!(
5375                    guest_buffer.read_with(&guest_cx, |buffer, _| buffer.text()),
5376                    host_buffer.read_with(&host_cx, |buffer, _| buffer.text()),
5377                    "guest {}, buffer {}, path {:?}, differs from the host's buffer",
5378                    guest_id,
5379                    buffer_id,
5380                    path
5381                );
5382            }
5383
5384            guest_cx.update(|_| drop(guest_client));
5385        }
5386
5387        host_cx.update(|_| drop(host_client));
5388    }
5389
5390    struct TestServer {
5391        peer: Arc<Peer>,
5392        app_state: Arc<AppState>,
5393        server: Arc<Server>,
5394        foreground: Rc<executor::Foreground>,
5395        notifications: mpsc::UnboundedReceiver<()>,
5396        connection_killers: Arc<Mutex<HashMap<UserId, Arc<AtomicBool>>>>,
5397        forbid_connections: Arc<AtomicBool>,
5398        _test_db: TestDb,
5399    }
5400
5401    impl TestServer {
5402        async fn start(
5403            foreground: Rc<executor::Foreground>,
5404            background: Arc<executor::Background>,
5405        ) -> Self {
5406            let test_db = TestDb::fake(background);
5407            let app_state = Self::build_app_state(&test_db).await;
5408            let peer = Peer::new();
5409            let notifications = mpsc::unbounded();
5410            let server = Server::new(app_state.clone(), peer.clone(), Some(notifications.0));
5411            Self {
5412                peer,
5413                app_state,
5414                server,
5415                foreground,
5416                notifications: notifications.1,
5417                connection_killers: Default::default(),
5418                forbid_connections: Default::default(),
5419                _test_db: test_db,
5420            }
5421        }
5422
5423        async fn create_client(&mut self, cx: &mut TestAppContext, name: &str) -> TestClient {
5424            cx.update(|cx| {
5425                let settings = Settings::test(cx);
5426                cx.set_global(settings);
5427            });
5428
5429            let http = FakeHttpClient::with_404_response();
5430            let user_id = self.app_state.db.create_user(name, false).await.unwrap();
5431            let client_name = name.to_string();
5432            let mut client = Client::new(http.clone());
5433            let server = self.server.clone();
5434            let connection_killers = self.connection_killers.clone();
5435            let forbid_connections = self.forbid_connections.clone();
5436            let (connection_id_tx, mut connection_id_rx) = mpsc::channel(16);
5437
5438            Arc::get_mut(&mut client)
5439                .unwrap()
5440                .override_authenticate(move |cx| {
5441                    cx.spawn(|_| async move {
5442                        let access_token = "the-token".to_string();
5443                        Ok(Credentials {
5444                            user_id: user_id.0 as u64,
5445                            access_token,
5446                        })
5447                    })
5448                })
5449                .override_establish_connection(move |credentials, cx| {
5450                    assert_eq!(credentials.user_id, user_id.0 as u64);
5451                    assert_eq!(credentials.access_token, "the-token");
5452
5453                    let server = server.clone();
5454                    let connection_killers = connection_killers.clone();
5455                    let forbid_connections = forbid_connections.clone();
5456                    let client_name = client_name.clone();
5457                    let connection_id_tx = connection_id_tx.clone();
5458                    cx.spawn(move |cx| async move {
5459                        if forbid_connections.load(SeqCst) {
5460                            Err(EstablishConnectionError::other(anyhow!(
5461                                "server is forbidding connections"
5462                            )))
5463                        } else {
5464                            let (client_conn, server_conn, killed) =
5465                                Connection::in_memory(cx.background());
5466                            connection_killers.lock().insert(user_id, killed);
5467                            cx.background()
5468                                .spawn(server.handle_connection(
5469                                    server_conn,
5470                                    client_name,
5471                                    user_id,
5472                                    Some(connection_id_tx),
5473                                    cx.background(),
5474                                ))
5475                                .detach();
5476                            Ok(client_conn)
5477                        }
5478                    })
5479                });
5480
5481            client
5482                .authenticate_and_connect(false, &cx.to_async())
5483                .await
5484                .unwrap();
5485
5486            Channel::init(&client);
5487            Project::init(&client);
5488            cx.update(|cx| {
5489                workspace::init(&client, cx);
5490            });
5491
5492            let peer_id = PeerId(connection_id_rx.next().await.unwrap().0);
5493            let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http, cx));
5494
5495            let client = TestClient {
5496                client,
5497                peer_id,
5498                user_store,
5499                language_registry: Arc::new(LanguageRegistry::test()),
5500                project: Default::default(),
5501                buffers: Default::default(),
5502            };
5503            client.wait_for_current_user(cx).await;
5504            client
5505        }
5506
5507        fn disconnect_client(&self, user_id: UserId) {
5508            self.connection_killers
5509                .lock()
5510                .remove(&user_id)
5511                .unwrap()
5512                .store(true, SeqCst);
5513        }
5514
5515        fn forbid_connections(&self) {
5516            self.forbid_connections.store(true, SeqCst);
5517        }
5518
5519        fn allow_connections(&self) {
5520            self.forbid_connections.store(false, SeqCst);
5521        }
5522
5523        async fn build_app_state(test_db: &TestDb) -> Arc<AppState> {
5524            let mut config = Config::default();
5525            config.session_secret = "a".repeat(32);
5526            config.database_url = test_db.url.clone();
5527            let github_client = github::AppClient::test();
5528            Arc::new(AppState {
5529                db: test_db.db().clone(),
5530                handlebars: Default::default(),
5531                auth_client: auth::build_client("", ""),
5532                repo_client: github::RepoClient::test(&github_client),
5533                github_client,
5534                config,
5535            })
5536        }
5537
5538        async fn state<'a>(&'a self) -> RwLockReadGuard<'a, Store> {
5539            self.server.store.read()
5540        }
5541
5542        async fn condition<F>(&mut self, mut predicate: F)
5543        where
5544            F: FnMut(&Store) -> bool,
5545        {
5546            async_std::future::timeout(Duration::from_millis(500), async {
5547                while !(predicate)(&*self.server.store.read()) {
5548                    self.foreground.start_waiting();
5549                    self.notifications.next().await;
5550                    self.foreground.finish_waiting();
5551                }
5552            })
5553            .await
5554            .expect("condition timed out");
5555        }
5556    }
5557
5558    impl Deref for TestServer {
5559        type Target = Server;
5560
5561        fn deref(&self) -> &Self::Target {
5562            &self.server
5563        }
5564    }
5565
5566    impl Drop for TestServer {
5567        fn drop(&mut self) {
5568            self.peer.reset();
5569        }
5570    }
5571
5572    struct TestClient {
5573        client: Arc<Client>,
5574        pub peer_id: PeerId,
5575        pub user_store: ModelHandle<UserStore>,
5576        language_registry: Arc<LanguageRegistry>,
5577        project: Option<ModelHandle<Project>>,
5578        buffers: HashSet<ModelHandle<language::Buffer>>,
5579    }
5580
5581    impl Deref for TestClient {
5582        type Target = Arc<Client>;
5583
5584        fn deref(&self) -> &Self::Target {
5585            &self.client
5586        }
5587    }
5588
5589    impl TestClient {
5590        pub fn current_user_id(&self, cx: &TestAppContext) -> UserId {
5591            UserId::from_proto(
5592                self.user_store
5593                    .read_with(cx, |user_store, _| user_store.current_user().unwrap().id),
5594            )
5595        }
5596
5597        async fn wait_for_current_user(&self, cx: &TestAppContext) {
5598            let mut authed_user = self
5599                .user_store
5600                .read_with(cx, |user_store, _| user_store.watch_current_user());
5601            while authed_user.next().await.unwrap().is_none() {}
5602        }
5603
5604        async fn build_local_project(
5605            &mut self,
5606            fs: Arc<FakeFs>,
5607            root_path: impl AsRef<Path>,
5608            cx: &mut TestAppContext,
5609        ) -> (ModelHandle<Project>, WorktreeId) {
5610            let project = cx.update(|cx| {
5611                Project::local(
5612                    self.client.clone(),
5613                    self.user_store.clone(),
5614                    self.language_registry.clone(),
5615                    fs,
5616                    cx,
5617                )
5618            });
5619            self.project = Some(project.clone());
5620            let (worktree, _) = project
5621                .update(cx, |p, cx| {
5622                    p.find_or_create_local_worktree(root_path, true, cx)
5623                })
5624                .await
5625                .unwrap();
5626            worktree
5627                .read_with(cx, |tree, _| tree.as_local().unwrap().scan_complete())
5628                .await;
5629            project
5630                .update(cx, |project, _| project.next_remote_id())
5631                .await;
5632            (project, worktree.read_with(cx, |tree, _| tree.id()))
5633        }
5634
5635        async fn build_remote_project(
5636            &mut self,
5637            project_id: u64,
5638            cx: &mut TestAppContext,
5639        ) -> ModelHandle<Project> {
5640            let project = Project::remote(
5641                project_id,
5642                self.client.clone(),
5643                self.user_store.clone(),
5644                self.language_registry.clone(),
5645                FakeFs::new(cx.background()),
5646                &mut cx.to_async(),
5647            )
5648            .await
5649            .unwrap();
5650            self.project = Some(project.clone());
5651            project
5652        }
5653
5654        fn build_workspace(
5655            &self,
5656            project: &ModelHandle<Project>,
5657            cx: &mut TestAppContext,
5658        ) -> ViewHandle<Workspace> {
5659            let (window_id, _) = cx.add_window(|_| EmptyView);
5660            cx.add_view(window_id, |cx| {
5661                let fs = project.read(cx).fs().clone();
5662                Workspace::new(
5663                    &WorkspaceParams {
5664                        fs,
5665                        project: project.clone(),
5666                        user_store: self.user_store.clone(),
5667                        languages: self.language_registry.clone(),
5668                        channel_list: cx.add_model(|cx| {
5669                            ChannelList::new(self.user_store.clone(), self.client.clone(), cx)
5670                        }),
5671                        client: self.client.clone(),
5672                    },
5673                    cx,
5674                )
5675            })
5676        }
5677
5678        async fn simulate_host(
5679            mut self,
5680            project: ModelHandle<Project>,
5681            files: Arc<Mutex<Vec<PathBuf>>>,
5682            op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
5683            rng: Arc<Mutex<StdRng>>,
5684            mut cx: TestAppContext,
5685        ) -> (Self, TestAppContext) {
5686            async fn simulate_host_internal(
5687                client: &mut TestClient,
5688                project: ModelHandle<Project>,
5689                files: Arc<Mutex<Vec<PathBuf>>>,
5690                mut op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
5691                rng: Arc<Mutex<StdRng>>,
5692                cx: &mut TestAppContext,
5693            ) -> anyhow::Result<()> {
5694                let fs = project.read_with(cx, |project, _| project.fs().clone());
5695
5696                while op_start_signal.next().await.is_some() {
5697                    let distribution = rng.lock().gen_range::<usize, _>(0..100);
5698                    match distribution {
5699                        0..=20 if !files.lock().is_empty() => {
5700                            let path = files.lock().choose(&mut *rng.lock()).unwrap().clone();
5701                            let mut path = path.as_path();
5702                            while let Some(parent_path) = path.parent() {
5703                                path = parent_path;
5704                                if rng.lock().gen() {
5705                                    break;
5706                                }
5707                            }
5708
5709                            log::info!("Host: find/create local worktree {:?}", path);
5710                            let find_or_create_worktree = project.update(cx, |project, cx| {
5711                                project.find_or_create_local_worktree(path, true, cx)
5712                            });
5713                            if rng.lock().gen() {
5714                                cx.background().spawn(find_or_create_worktree).detach();
5715                            } else {
5716                                find_or_create_worktree.await?;
5717                            }
5718                        }
5719                        10..=80 if !files.lock().is_empty() => {
5720                            let buffer = if client.buffers.is_empty() || rng.lock().gen() {
5721                                let file = files.lock().choose(&mut *rng.lock()).unwrap().clone();
5722                                let (worktree, path) = project
5723                                    .update(cx, |project, cx| {
5724                                        project.find_or_create_local_worktree(
5725                                            file.clone(),
5726                                            true,
5727                                            cx,
5728                                        )
5729                                    })
5730                                    .await?;
5731                                let project_path =
5732                                    worktree.read_with(cx, |worktree, _| (worktree.id(), path));
5733                                log::info!(
5734                                    "Host: opening path {:?}, worktree {}, relative_path {:?}",
5735                                    file,
5736                                    project_path.0,
5737                                    project_path.1
5738                                );
5739                                let buffer = project
5740                                    .update(cx, |project, cx| project.open_buffer(project_path, cx))
5741                                    .await
5742                                    .unwrap();
5743                                client.buffers.insert(buffer.clone());
5744                                buffer
5745                            } else {
5746                                client
5747                                    .buffers
5748                                    .iter()
5749                                    .choose(&mut *rng.lock())
5750                                    .unwrap()
5751                                    .clone()
5752                            };
5753
5754                            if rng.lock().gen_bool(0.1) {
5755                                cx.update(|cx| {
5756                                    log::info!(
5757                                        "Host: dropping buffer {:?}",
5758                                        buffer.read(cx).file().unwrap().full_path(cx)
5759                                    );
5760                                    client.buffers.remove(&buffer);
5761                                    drop(buffer);
5762                                });
5763                            } else {
5764                                buffer.update(cx, |buffer, cx| {
5765                                    log::info!(
5766                                        "Host: updating buffer {:?} ({})",
5767                                        buffer.file().unwrap().full_path(cx),
5768                                        buffer.remote_id()
5769                                    );
5770                                    buffer.randomly_edit(&mut *rng.lock(), 5, cx)
5771                                });
5772                            }
5773                        }
5774                        _ => loop {
5775                            let path_component_count = rng.lock().gen_range::<usize, _>(1..=5);
5776                            let mut path = PathBuf::new();
5777                            path.push("/");
5778                            for _ in 0..path_component_count {
5779                                let letter = rng.lock().gen_range(b'a'..=b'z');
5780                                path.push(std::str::from_utf8(&[letter]).unwrap());
5781                            }
5782                            path.set_extension("rs");
5783                            let parent_path = path.parent().unwrap();
5784
5785                            log::info!("Host: creating file {:?}", path,);
5786
5787                            if fs.create_dir(&parent_path).await.is_ok()
5788                                && fs.create_file(&path, Default::default()).await.is_ok()
5789                            {
5790                                files.lock().push(path);
5791                                break;
5792                            } else {
5793                                log::info!("Host: cannot create file");
5794                            }
5795                        },
5796                    }
5797
5798                    cx.background().simulate_random_delay().await;
5799                }
5800
5801                Ok(())
5802            }
5803
5804            simulate_host_internal(
5805                &mut self,
5806                project.clone(),
5807                files,
5808                op_start_signal,
5809                rng,
5810                &mut cx,
5811            )
5812            .log_err()
5813            .await;
5814            log::info!("Host done");
5815            self.project = Some(project);
5816            (self, cx)
5817        }
5818
5819        pub async fn simulate_guest(
5820            mut self,
5821            guest_id: usize,
5822            project: ModelHandle<Project>,
5823            op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
5824            rng: Arc<Mutex<StdRng>>,
5825            host_disconnected: Rc<AtomicBool>,
5826            mut cx: TestAppContext,
5827        ) -> (Self, TestAppContext) {
5828            async fn simulate_guest_internal(
5829                client: &mut TestClient,
5830                guest_id: usize,
5831                project: ModelHandle<Project>,
5832                mut op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
5833                rng: Arc<Mutex<StdRng>>,
5834                cx: &mut TestAppContext,
5835            ) -> anyhow::Result<()> {
5836                while op_start_signal.next().await.is_some() {
5837                    let buffer = if client.buffers.is_empty() || rng.lock().gen() {
5838                        let worktree = if let Some(worktree) =
5839                            project.read_with(cx, |project, cx| {
5840                                project
5841                                    .worktrees(&cx)
5842                                    .filter(|worktree| {
5843                                        let worktree = worktree.read(cx);
5844                                        worktree.is_visible()
5845                                            && worktree.entries(false).any(|e| e.is_file())
5846                                    })
5847                                    .choose(&mut *rng.lock())
5848                            }) {
5849                            worktree
5850                        } else {
5851                            cx.background().simulate_random_delay().await;
5852                            continue;
5853                        };
5854
5855                        let (worktree_root_name, project_path) =
5856                            worktree.read_with(cx, |worktree, _| {
5857                                let entry = worktree
5858                                    .entries(false)
5859                                    .filter(|e| e.is_file())
5860                                    .choose(&mut *rng.lock())
5861                                    .unwrap();
5862                                (
5863                                    worktree.root_name().to_string(),
5864                                    (worktree.id(), entry.path.clone()),
5865                                )
5866                            });
5867                        log::info!(
5868                            "Guest {}: opening path {:?} in worktree {} ({})",
5869                            guest_id,
5870                            project_path.1,
5871                            project_path.0,
5872                            worktree_root_name,
5873                        );
5874                        let buffer = project
5875                            .update(cx, |project, cx| {
5876                                project.open_buffer(project_path.clone(), cx)
5877                            })
5878                            .await?;
5879                        log::info!(
5880                            "Guest {}: opened path {:?} in worktree {} ({}) with buffer id {}",
5881                            guest_id,
5882                            project_path.1,
5883                            project_path.0,
5884                            worktree_root_name,
5885                            buffer.read_with(cx, |buffer, _| buffer.remote_id())
5886                        );
5887                        client.buffers.insert(buffer.clone());
5888                        buffer
5889                    } else {
5890                        client
5891                            .buffers
5892                            .iter()
5893                            .choose(&mut *rng.lock())
5894                            .unwrap()
5895                            .clone()
5896                    };
5897
5898                    let choice = rng.lock().gen_range(0..100);
5899                    match choice {
5900                        0..=9 => {
5901                            cx.update(|cx| {
5902                                log::info!(
5903                                    "Guest {}: dropping buffer {:?}",
5904                                    guest_id,
5905                                    buffer.read(cx).file().unwrap().full_path(cx)
5906                                );
5907                                client.buffers.remove(&buffer);
5908                                drop(buffer);
5909                            });
5910                        }
5911                        10..=19 => {
5912                            let completions = project.update(cx, |project, cx| {
5913                                log::info!(
5914                                    "Guest {}: requesting completions for buffer {} ({:?})",
5915                                    guest_id,
5916                                    buffer.read(cx).remote_id(),
5917                                    buffer.read(cx).file().unwrap().full_path(cx)
5918                                );
5919                                let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
5920                                project.completions(&buffer, offset, cx)
5921                            });
5922                            let completions = cx.background().spawn(async move {
5923                                completions
5924                                    .await
5925                                    .map_err(|err| anyhow!("completions request failed: {:?}", err))
5926                            });
5927                            if rng.lock().gen_bool(0.3) {
5928                                log::info!("Guest {}: detaching completions request", guest_id);
5929                                cx.update(|cx| completions.detach_and_log_err(cx));
5930                            } else {
5931                                completions.await?;
5932                            }
5933                        }
5934                        20..=29 => {
5935                            let code_actions = project.update(cx, |project, cx| {
5936                                log::info!(
5937                                    "Guest {}: requesting code actions for buffer {} ({:?})",
5938                                    guest_id,
5939                                    buffer.read(cx).remote_id(),
5940                                    buffer.read(cx).file().unwrap().full_path(cx)
5941                                );
5942                                let range = buffer.read(cx).random_byte_range(0, &mut *rng.lock());
5943                                project.code_actions(&buffer, range, cx)
5944                            });
5945                            let code_actions = cx.background().spawn(async move {
5946                                code_actions.await.map_err(|err| {
5947                                    anyhow!("code actions request failed: {:?}", err)
5948                                })
5949                            });
5950                            if rng.lock().gen_bool(0.3) {
5951                                log::info!("Guest {}: detaching code actions request", guest_id);
5952                                cx.update(|cx| code_actions.detach_and_log_err(cx));
5953                            } else {
5954                                code_actions.await?;
5955                            }
5956                        }
5957                        30..=39 if buffer.read_with(cx, |buffer, _| buffer.is_dirty()) => {
5958                            let (requested_version, save) = buffer.update(cx, |buffer, cx| {
5959                                log::info!(
5960                                    "Guest {}: saving buffer {} ({:?})",
5961                                    guest_id,
5962                                    buffer.remote_id(),
5963                                    buffer.file().unwrap().full_path(cx)
5964                                );
5965                                (buffer.version(), buffer.save(cx))
5966                            });
5967                            let save = cx.background().spawn(async move {
5968                                let (saved_version, _) = save
5969                                    .await
5970                                    .map_err(|err| anyhow!("save request failed: {:?}", err))?;
5971                                assert!(saved_version.observed_all(&requested_version));
5972                                Ok::<_, anyhow::Error>(())
5973                            });
5974                            if rng.lock().gen_bool(0.3) {
5975                                log::info!("Guest {}: detaching save request", guest_id);
5976                                cx.update(|cx| save.detach_and_log_err(cx));
5977                            } else {
5978                                save.await?;
5979                            }
5980                        }
5981                        40..=44 => {
5982                            let prepare_rename = project.update(cx, |project, cx| {
5983                                log::info!(
5984                                    "Guest {}: preparing rename for buffer {} ({:?})",
5985                                    guest_id,
5986                                    buffer.read(cx).remote_id(),
5987                                    buffer.read(cx).file().unwrap().full_path(cx)
5988                                );
5989                                let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
5990                                project.prepare_rename(buffer, offset, cx)
5991                            });
5992                            let prepare_rename = cx.background().spawn(async move {
5993                                prepare_rename.await.map_err(|err| {
5994                                    anyhow!("prepare rename request failed: {:?}", err)
5995                                })
5996                            });
5997                            if rng.lock().gen_bool(0.3) {
5998                                log::info!("Guest {}: detaching prepare rename request", guest_id);
5999                                cx.update(|cx| prepare_rename.detach_and_log_err(cx));
6000                            } else {
6001                                prepare_rename.await?;
6002                            }
6003                        }
6004                        45..=49 => {
6005                            let definitions = project.update(cx, |project, cx| {
6006                                log::info!(
6007                                    "Guest {}: requesting definitions for buffer {} ({:?})",
6008                                    guest_id,
6009                                    buffer.read(cx).remote_id(),
6010                                    buffer.read(cx).file().unwrap().full_path(cx)
6011                                );
6012                                let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
6013                                project.definition(&buffer, offset, cx)
6014                            });
6015                            let definitions = cx.background().spawn(async move {
6016                                definitions
6017                                    .await
6018                                    .map_err(|err| anyhow!("definitions request failed: {:?}", err))
6019                            });
6020                            if rng.lock().gen_bool(0.3) {
6021                                log::info!("Guest {}: detaching definitions request", guest_id);
6022                                cx.update(|cx| definitions.detach_and_log_err(cx));
6023                            } else {
6024                                client
6025                                    .buffers
6026                                    .extend(definitions.await?.into_iter().map(|loc| loc.buffer));
6027                            }
6028                        }
6029                        50..=54 => {
6030                            let highlights = project.update(cx, |project, cx| {
6031                                log::info!(
6032                                    "Guest {}: requesting highlights for buffer {} ({:?})",
6033                                    guest_id,
6034                                    buffer.read(cx).remote_id(),
6035                                    buffer.read(cx).file().unwrap().full_path(cx)
6036                                );
6037                                let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
6038                                project.document_highlights(&buffer, offset, cx)
6039                            });
6040                            let highlights = cx.background().spawn(async move {
6041                                highlights
6042                                    .await
6043                                    .map_err(|err| anyhow!("highlights request failed: {:?}", err))
6044                            });
6045                            if rng.lock().gen_bool(0.3) {
6046                                log::info!("Guest {}: detaching highlights request", guest_id);
6047                                cx.update(|cx| highlights.detach_and_log_err(cx));
6048                            } else {
6049                                highlights.await?;
6050                            }
6051                        }
6052                        55..=59 => {
6053                            let search = project.update(cx, |project, cx| {
6054                                let query = rng.lock().gen_range('a'..='z');
6055                                log::info!("Guest {}: project-wide search {:?}", guest_id, query);
6056                                project.search(SearchQuery::text(query, false, false), cx)
6057                            });
6058                            let search = cx.background().spawn(async move {
6059                                search
6060                                    .await
6061                                    .map_err(|err| anyhow!("search request failed: {:?}", err))
6062                            });
6063                            if rng.lock().gen_bool(0.3) {
6064                                log::info!("Guest {}: detaching search request", guest_id);
6065                                cx.update(|cx| search.detach_and_log_err(cx));
6066                            } else {
6067                                client.buffers.extend(search.await?.into_keys());
6068                            }
6069                        }
6070                        _ => {
6071                            buffer.update(cx, |buffer, cx| {
6072                                log::info!(
6073                                    "Guest {}: updating buffer {} ({:?})",
6074                                    guest_id,
6075                                    buffer.remote_id(),
6076                                    buffer.file().unwrap().full_path(cx)
6077                                );
6078                                buffer.randomly_edit(&mut *rng.lock(), 5, cx)
6079                            });
6080                        }
6081                    }
6082                    cx.background().simulate_random_delay().await;
6083                }
6084                Ok(())
6085            }
6086
6087            match simulate_guest_internal(
6088                &mut self,
6089                guest_id,
6090                project.clone(),
6091                op_start_signal,
6092                rng,
6093                &mut cx,
6094            )
6095            .await
6096            {
6097                Ok(()) => log::info!("guest {} done", guest_id),
6098                Err(err) => {
6099                    if host_disconnected.load(SeqCst) {
6100                        log::error!("guest {} simulation error - {:?}", guest_id, err);
6101                    } else {
6102                        panic!("guest {} simulation error - {:?}", guest_id, err);
6103                    }
6104                }
6105            }
6106
6107            self.project = Some(project);
6108            (self, cx)
6109        }
6110    }
6111
6112    impl Drop for TestClient {
6113        fn drop(&mut self) {
6114            self.client.tear_down();
6115        }
6116    }
6117
6118    impl Executor for Arc<gpui::executor::Background> {
6119        type Timer = gpui::executor::Timer;
6120
6121        fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F) {
6122            self.spawn(future).detach();
6123        }
6124
6125        fn timer(&self, duration: Duration) -> Self::Timer {
6126            self.as_ref().timer(duration)
6127        }
6128    }
6129
6130    fn channel_messages(channel: &Channel) -> Vec<(String, String, bool)> {
6131        channel
6132            .messages()
6133            .cursor::<()>()
6134            .map(|m| {
6135                (
6136                    m.sender.github_login.clone(),
6137                    m.body.clone(),
6138                    m.is_pending(),
6139                )
6140            })
6141            .collect()
6142    }
6143
6144    struct EmptyView;
6145
6146    impl gpui::Entity for EmptyView {
6147        type Event = ();
6148    }
6149
6150    impl gpui::View for EmptyView {
6151        fn ui_name() -> &'static str {
6152            "empty view"
6153        }
6154
6155        fn render(&mut self, _: &mut gpui::RenderContext<Self>) -> gpui::ElementBox {
6156            gpui::Element::boxed(gpui::elements::Empty)
6157        }
6158    }
6159}