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