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