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 parking_lot::{RwLock, RwLockReadGuard, RwLockWriteGuard};
  15use rpc::{
  16    proto::{self, AnyTypedEnvelope, EntityMessage, EnvelopedMessage, RequestMessage},
  17    Connection, ConnectionId, Peer, TypedEnvelope,
  18};
  19use sha1::{Digest as _, Sha1};
  20use std::{
  21    any::TypeId,
  22    future::Future,
  23    sync::Arc,
  24    time::{Duration, Instant},
  25};
  26use store::{Store, Worktree};
  27use surf::StatusCode;
  28use tide::log;
  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!("rpc message received. connection:{}, type:{}", connection_id, type_name);
 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!("rpc message error. connection:{}, type:{}, error:{:?}", connection_id, type_name, err);
 229                                    } else {
 230                                        log::info!("rpc message handled. connection:{}, type:{}, duration:{:?}", connection_id, type_name, start_time.elapsed());
 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!("rpc connection closed {:?}", addr);
 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 postage::barrier;
1098    use project::{
1099        fs::{FakeFs, Fs as _},
1100        search::SearchQuery,
1101        worktree::WorktreeHandle,
1102        DiagnosticSummary, Project, ProjectPath, WorktreeId,
1103    };
1104    use rand::prelude::*;
1105    use rpc::PeerId;
1106    use serde_json::json;
1107    use settings::Settings;
1108    use sqlx::types::time::OffsetDateTime;
1109    use std::{
1110        cell::Cell,
1111        env,
1112        ops::Deref,
1113        path::{Path, PathBuf},
1114        rc::Rc,
1115        sync::{
1116            atomic::{AtomicBool, Ordering::SeqCst},
1117            Arc,
1118        },
1119        time::Duration,
1120    };
1121    use util::TryFutureExt;
1122    use workspace::{Item, SplitDirection, 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(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(&ToggleCodeActions(false), cx);
3612        });
3613        editor_b
3614            .condition(&cx_b, |editor, _| editor.context_menu_visible())
3615            .await;
3616
3617        fake_language_server.remove_request_handler::<lsp::request::CodeActionRequest>();
3618
3619        // Confirming the code action will trigger a resolve request.
3620        let confirm_action = workspace_b
3621            .update(cx_b, |workspace, cx| {
3622                Editor::confirm_code_action(workspace, &ConfirmCodeAction(Some(0)), cx)
3623            })
3624            .unwrap();
3625        fake_language_server.handle_request::<lsp::request::CodeActionResolveRequest, _, _>(
3626            |_, _| async move {
3627                Ok(lsp::CodeAction {
3628                    title: "Inline into all callers".to_string(),
3629                    edit: Some(lsp::WorkspaceEdit {
3630                        changes: Some(
3631                            [
3632                                (
3633                                    lsp::Url::from_file_path("/a/main.rs").unwrap(),
3634                                    vec![lsp::TextEdit::new(
3635                                        lsp::Range::new(
3636                                            lsp::Position::new(1, 22),
3637                                            lsp::Position::new(1, 34),
3638                                        ),
3639                                        "4".to_string(),
3640                                    )],
3641                                ),
3642                                (
3643                                    lsp::Url::from_file_path("/a/other.rs").unwrap(),
3644                                    vec![lsp::TextEdit::new(
3645                                        lsp::Range::new(
3646                                            lsp::Position::new(0, 0),
3647                                            lsp::Position::new(0, 27),
3648                                        ),
3649                                        "".to_string(),
3650                                    )],
3651                                ),
3652                            ]
3653                            .into_iter()
3654                            .collect(),
3655                        ),
3656                        ..Default::default()
3657                    }),
3658                    ..Default::default()
3659                })
3660            },
3661        );
3662
3663        // After the action is confirmed, an editor containing both modified files is opened.
3664        confirm_action.await.unwrap();
3665        let code_action_editor = workspace_b.read_with(cx_b, |workspace, cx| {
3666            workspace
3667                .active_item(cx)
3668                .unwrap()
3669                .downcast::<Editor>()
3670                .unwrap()
3671        });
3672        code_action_editor.update(cx_b, |editor, cx| {
3673            assert_eq!(editor.text(cx), "\nmod other;\nfn main() { let foo = 4; }");
3674            editor.undo(&Undo, cx);
3675            assert_eq!(
3676                editor.text(cx),
3677                "pub fn foo() -> usize { 4 }\nmod other;\nfn main() { let foo = other::foo(); }"
3678            );
3679            editor.redo(&Redo, cx);
3680            assert_eq!(editor.text(cx), "\nmod other;\nfn main() { let foo = 4; }");
3681        });
3682    }
3683
3684    #[gpui::test(iterations = 10)]
3685    async fn test_collaborating_with_renames(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3686        cx_a.foreground().forbid_parking();
3687        let lang_registry = Arc::new(LanguageRegistry::test());
3688        let fs = FakeFs::new(cx_a.background());
3689        cx_b.update(|cx| editor::init(cx));
3690
3691        // Set up a fake language server.
3692        let mut language = Language::new(
3693            LanguageConfig {
3694                name: "Rust".into(),
3695                path_suffixes: vec!["rs".to_string()],
3696                ..Default::default()
3697            },
3698            Some(tree_sitter_rust::language()),
3699        );
3700        let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3701        lang_registry.add(Arc::new(language));
3702
3703        // Connect to a server as 2 clients.
3704        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3705        let client_a = server.create_client(cx_a, "user_a").await;
3706        let client_b = server.create_client(cx_b, "user_b").await;
3707
3708        // Share a project as client A
3709        fs.insert_tree(
3710            "/dir",
3711            json!({
3712                ".zed.toml": r#"collaborators = ["user_b"]"#,
3713                "one.rs": "const ONE: usize = 1;",
3714                "two.rs": "const TWO: usize = one::ONE + one::ONE;"
3715            }),
3716        )
3717        .await;
3718        let project_a = cx_a.update(|cx| {
3719            Project::local(
3720                client_a.clone(),
3721                client_a.user_store.clone(),
3722                lang_registry.clone(),
3723                fs.clone(),
3724                cx,
3725            )
3726        });
3727        let (worktree_a, _) = project_a
3728            .update(cx_a, |p, cx| {
3729                p.find_or_create_local_worktree("/dir", true, cx)
3730            })
3731            .await
3732            .unwrap();
3733        worktree_a
3734            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3735            .await;
3736        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
3737        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3738        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
3739
3740        // Join the worktree as client B.
3741        let project_b = Project::remote(
3742            project_id,
3743            client_b.clone(),
3744            client_b.user_store.clone(),
3745            lang_registry.clone(),
3746            fs.clone(),
3747            &mut cx_b.to_async(),
3748        )
3749        .await
3750        .unwrap();
3751        let mut params = cx_b.update(WorkspaceParams::test);
3752        params.languages = lang_registry.clone();
3753        params.client = client_b.client.clone();
3754        params.user_store = client_b.user_store.clone();
3755        params.project = project_b;
3756
3757        let (_window_b, workspace_b) = cx_b.add_window(|cx| Workspace::new(&params, cx));
3758        let editor_b = workspace_b
3759            .update(cx_b, |workspace, cx| {
3760                workspace.open_path((worktree_id, "one.rs"), cx)
3761            })
3762            .await
3763            .unwrap()
3764            .downcast::<Editor>()
3765            .unwrap();
3766        let fake_language_server = fake_language_servers.next().await.unwrap();
3767
3768        // Move cursor to a location that can be renamed.
3769        let prepare_rename = editor_b.update(cx_b, |editor, cx| {
3770            editor.select_ranges([7..7], None, cx);
3771            editor.rename(&Rename, cx).unwrap()
3772        });
3773
3774        fake_language_server
3775            .handle_request::<lsp::request::PrepareRenameRequest, _, _>(|params, _| async move {
3776                assert_eq!(params.text_document.uri.as_str(), "file:///dir/one.rs");
3777                assert_eq!(params.position, lsp::Position::new(0, 7));
3778                Ok(Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
3779                    lsp::Position::new(0, 6),
3780                    lsp::Position::new(0, 9),
3781                ))))
3782            })
3783            .next()
3784            .await
3785            .unwrap();
3786        prepare_rename.await.unwrap();
3787        editor_b.update(cx_b, |editor, cx| {
3788            let rename = editor.pending_rename().unwrap();
3789            let buffer = editor.buffer().read(cx).snapshot(cx);
3790            assert_eq!(
3791                rename.range.start.to_offset(&buffer)..rename.range.end.to_offset(&buffer),
3792                6..9
3793            );
3794            rename.editor.update(cx, |rename_editor, cx| {
3795                rename_editor.buffer().update(cx, |rename_buffer, cx| {
3796                    rename_buffer.edit([0..3], "THREE", cx);
3797                });
3798            });
3799        });
3800
3801        let confirm_rename = workspace_b.update(cx_b, |workspace, cx| {
3802            Editor::confirm_rename(workspace, &ConfirmRename, cx).unwrap()
3803        });
3804        fake_language_server
3805            .handle_request::<lsp::request::Rename, _, _>(|params, _| async move {
3806                assert_eq!(
3807                    params.text_document_position.text_document.uri.as_str(),
3808                    "file:///dir/one.rs"
3809                );
3810                assert_eq!(
3811                    params.text_document_position.position,
3812                    lsp::Position::new(0, 6)
3813                );
3814                assert_eq!(params.new_name, "THREE");
3815                Ok(Some(lsp::WorkspaceEdit {
3816                    changes: Some(
3817                        [
3818                            (
3819                                lsp::Url::from_file_path("/dir/one.rs").unwrap(),
3820                                vec![lsp::TextEdit::new(
3821                                    lsp::Range::new(
3822                                        lsp::Position::new(0, 6),
3823                                        lsp::Position::new(0, 9),
3824                                    ),
3825                                    "THREE".to_string(),
3826                                )],
3827                            ),
3828                            (
3829                                lsp::Url::from_file_path("/dir/two.rs").unwrap(),
3830                                vec![
3831                                    lsp::TextEdit::new(
3832                                        lsp::Range::new(
3833                                            lsp::Position::new(0, 24),
3834                                            lsp::Position::new(0, 27),
3835                                        ),
3836                                        "THREE".to_string(),
3837                                    ),
3838                                    lsp::TextEdit::new(
3839                                        lsp::Range::new(
3840                                            lsp::Position::new(0, 35),
3841                                            lsp::Position::new(0, 38),
3842                                        ),
3843                                        "THREE".to_string(),
3844                                    ),
3845                                ],
3846                            ),
3847                        ]
3848                        .into_iter()
3849                        .collect(),
3850                    ),
3851                    ..Default::default()
3852                }))
3853            })
3854            .next()
3855            .await
3856            .unwrap();
3857        confirm_rename.await.unwrap();
3858
3859        let rename_editor = workspace_b.read_with(cx_b, |workspace, cx| {
3860            workspace
3861                .active_item(cx)
3862                .unwrap()
3863                .downcast::<Editor>()
3864                .unwrap()
3865        });
3866        rename_editor.update(cx_b, |editor, cx| {
3867            assert_eq!(
3868                editor.text(cx),
3869                "const TWO: usize = one::THREE + one::THREE;\nconst THREE: usize = 1;"
3870            );
3871            editor.undo(&Undo, cx);
3872            assert_eq!(
3873                editor.text(cx),
3874                "const TWO: usize = one::ONE + one::ONE;\nconst ONE: usize = 1;"
3875            );
3876            editor.redo(&Redo, cx);
3877            assert_eq!(
3878                editor.text(cx),
3879                "const TWO: usize = one::THREE + one::THREE;\nconst THREE: usize = 1;"
3880            );
3881        });
3882
3883        // Ensure temporary rename edits cannot be undone/redone.
3884        editor_b.update(cx_b, |editor, cx| {
3885            editor.undo(&Undo, cx);
3886            assert_eq!(editor.text(cx), "const ONE: usize = 1;");
3887            editor.undo(&Undo, cx);
3888            assert_eq!(editor.text(cx), "const ONE: usize = 1;");
3889            editor.redo(&Redo, cx);
3890            assert_eq!(editor.text(cx), "const THREE: usize = 1;");
3891        })
3892    }
3893
3894    #[gpui::test(iterations = 10)]
3895    async fn test_basic_chat(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3896        cx_a.foreground().forbid_parking();
3897
3898        // Connect to a server as 2 clients.
3899        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3900        let client_a = server.create_client(cx_a, "user_a").await;
3901        let client_b = server.create_client(cx_b, "user_b").await;
3902
3903        // Create an org that includes these 2 users.
3904        let db = &server.app_state.db;
3905        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
3906        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
3907            .await
3908            .unwrap();
3909        db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
3910            .await
3911            .unwrap();
3912
3913        // Create a channel that includes all the users.
3914        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
3915        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
3916            .await
3917            .unwrap();
3918        db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
3919            .await
3920            .unwrap();
3921        db.create_channel_message(
3922            channel_id,
3923            client_b.current_user_id(&cx_b),
3924            "hello A, it's B.",
3925            OffsetDateTime::now_utc(),
3926            1,
3927        )
3928        .await
3929        .unwrap();
3930
3931        let channels_a = cx_a
3932            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
3933        channels_a
3934            .condition(cx_a, |list, _| list.available_channels().is_some())
3935            .await;
3936        channels_a.read_with(cx_a, |list, _| {
3937            assert_eq!(
3938                list.available_channels().unwrap(),
3939                &[ChannelDetails {
3940                    id: channel_id.to_proto(),
3941                    name: "test-channel".to_string()
3942                }]
3943            )
3944        });
3945        let channel_a = channels_a.update(cx_a, |this, cx| {
3946            this.get_channel(channel_id.to_proto(), cx).unwrap()
3947        });
3948        channel_a.read_with(cx_a, |channel, _| assert!(channel.messages().is_empty()));
3949        channel_a
3950            .condition(&cx_a, |channel, _| {
3951                channel_messages(channel)
3952                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
3953            })
3954            .await;
3955
3956        let channels_b = cx_b
3957            .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
3958        channels_b
3959            .condition(cx_b, |list, _| list.available_channels().is_some())
3960            .await;
3961        channels_b.read_with(cx_b, |list, _| {
3962            assert_eq!(
3963                list.available_channels().unwrap(),
3964                &[ChannelDetails {
3965                    id: channel_id.to_proto(),
3966                    name: "test-channel".to_string()
3967                }]
3968            )
3969        });
3970
3971        let channel_b = channels_b.update(cx_b, |this, cx| {
3972            this.get_channel(channel_id.to_proto(), cx).unwrap()
3973        });
3974        channel_b.read_with(cx_b, |channel, _| assert!(channel.messages().is_empty()));
3975        channel_b
3976            .condition(&cx_b, |channel, _| {
3977                channel_messages(channel)
3978                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
3979            })
3980            .await;
3981
3982        channel_a
3983            .update(cx_a, |channel, cx| {
3984                channel
3985                    .send_message("oh, hi B.".to_string(), cx)
3986                    .unwrap()
3987                    .detach();
3988                let task = channel.send_message("sup".to_string(), cx).unwrap();
3989                assert_eq!(
3990                    channel_messages(channel),
3991                    &[
3992                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
3993                        ("user_a".to_string(), "oh, hi B.".to_string(), true),
3994                        ("user_a".to_string(), "sup".to_string(), true)
3995                    ]
3996                );
3997                task
3998            })
3999            .await
4000            .unwrap();
4001
4002        channel_b
4003            .condition(&cx_b, |channel, _| {
4004                channel_messages(channel)
4005                    == [
4006                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4007                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4008                        ("user_a".to_string(), "sup".to_string(), false),
4009                    ]
4010            })
4011            .await;
4012
4013        assert_eq!(
4014            server
4015                .state()
4016                .await
4017                .channel(channel_id)
4018                .unwrap()
4019                .connection_ids
4020                .len(),
4021            2
4022        );
4023        cx_b.update(|_| drop(channel_b));
4024        server
4025            .condition(|state| state.channel(channel_id).unwrap().connection_ids.len() == 1)
4026            .await;
4027
4028        cx_a.update(|_| drop(channel_a));
4029        server
4030            .condition(|state| state.channel(channel_id).is_none())
4031            .await;
4032    }
4033
4034    #[gpui::test(iterations = 10)]
4035    async fn test_chat_message_validation(cx_a: &mut TestAppContext) {
4036        cx_a.foreground().forbid_parking();
4037
4038        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4039        let client_a = server.create_client(cx_a, "user_a").await;
4040
4041        let db = &server.app_state.db;
4042        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
4043        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
4044        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
4045            .await
4046            .unwrap();
4047        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
4048            .await
4049            .unwrap();
4050
4051        let channels_a = cx_a
4052            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
4053        channels_a
4054            .condition(cx_a, |list, _| list.available_channels().is_some())
4055            .await;
4056        let channel_a = channels_a.update(cx_a, |this, cx| {
4057            this.get_channel(channel_id.to_proto(), cx).unwrap()
4058        });
4059
4060        // Messages aren't allowed to be too long.
4061        channel_a
4062            .update(cx_a, |channel, cx| {
4063                let long_body = "this is long.\n".repeat(1024);
4064                channel.send_message(long_body, cx).unwrap()
4065            })
4066            .await
4067            .unwrap_err();
4068
4069        // Messages aren't allowed to be blank.
4070        channel_a.update(cx_a, |channel, cx| {
4071            channel.send_message(String::new(), cx).unwrap_err()
4072        });
4073
4074        // Leading and trailing whitespace are trimmed.
4075        channel_a
4076            .update(cx_a, |channel, cx| {
4077                channel
4078                    .send_message("\n surrounded by whitespace  \n".to_string(), cx)
4079                    .unwrap()
4080            })
4081            .await
4082            .unwrap();
4083        assert_eq!(
4084            db.get_channel_messages(channel_id, 10, None)
4085                .await
4086                .unwrap()
4087                .iter()
4088                .map(|m| &m.body)
4089                .collect::<Vec<_>>(),
4090            &["surrounded by whitespace"]
4091        );
4092    }
4093
4094    #[gpui::test(iterations = 10)]
4095    async fn test_chat_reconnection(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4096        cx_a.foreground().forbid_parking();
4097
4098        // Connect to a server as 2 clients.
4099        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4100        let client_a = server.create_client(cx_a, "user_a").await;
4101        let client_b = server.create_client(cx_b, "user_b").await;
4102        let mut status_b = client_b.status();
4103
4104        // Create an org that includes these 2 users.
4105        let db = &server.app_state.db;
4106        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
4107        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
4108            .await
4109            .unwrap();
4110        db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
4111            .await
4112            .unwrap();
4113
4114        // Create a channel that includes all the users.
4115        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
4116        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
4117            .await
4118            .unwrap();
4119        db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
4120            .await
4121            .unwrap();
4122        db.create_channel_message(
4123            channel_id,
4124            client_b.current_user_id(&cx_b),
4125            "hello A, it's B.",
4126            OffsetDateTime::now_utc(),
4127            2,
4128        )
4129        .await
4130        .unwrap();
4131
4132        let channels_a = cx_a
4133            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
4134        channels_a
4135            .condition(cx_a, |list, _| list.available_channels().is_some())
4136            .await;
4137
4138        channels_a.read_with(cx_a, |list, _| {
4139            assert_eq!(
4140                list.available_channels().unwrap(),
4141                &[ChannelDetails {
4142                    id: channel_id.to_proto(),
4143                    name: "test-channel".to_string()
4144                }]
4145            )
4146        });
4147        let channel_a = channels_a.update(cx_a, |this, cx| {
4148            this.get_channel(channel_id.to_proto(), cx).unwrap()
4149        });
4150        channel_a.read_with(cx_a, |channel, _| assert!(channel.messages().is_empty()));
4151        channel_a
4152            .condition(&cx_a, |channel, _| {
4153                channel_messages(channel)
4154                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4155            })
4156            .await;
4157
4158        let channels_b = cx_b
4159            .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
4160        channels_b
4161            .condition(cx_b, |list, _| list.available_channels().is_some())
4162            .await;
4163        channels_b.read_with(cx_b, |list, _| {
4164            assert_eq!(
4165                list.available_channels().unwrap(),
4166                &[ChannelDetails {
4167                    id: channel_id.to_proto(),
4168                    name: "test-channel".to_string()
4169                }]
4170            )
4171        });
4172
4173        let channel_b = channels_b.update(cx_b, |this, cx| {
4174            this.get_channel(channel_id.to_proto(), cx).unwrap()
4175        });
4176        channel_b.read_with(cx_b, |channel, _| assert!(channel.messages().is_empty()));
4177        channel_b
4178            .condition(&cx_b, |channel, _| {
4179                channel_messages(channel)
4180                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4181            })
4182            .await;
4183
4184        // Disconnect client B, ensuring we can still access its cached channel data.
4185        server.forbid_connections();
4186        server.disconnect_client(client_b.current_user_id(&cx_b));
4187        cx_b.foreground().advance_clock(Duration::from_secs(3));
4188        while !matches!(
4189            status_b.next().await,
4190            Some(client::Status::ReconnectionError { .. })
4191        ) {}
4192
4193        channels_b.read_with(cx_b, |channels, _| {
4194            assert_eq!(
4195                channels.available_channels().unwrap(),
4196                [ChannelDetails {
4197                    id: channel_id.to_proto(),
4198                    name: "test-channel".to_string()
4199                }]
4200            )
4201        });
4202        channel_b.read_with(cx_b, |channel, _| {
4203            assert_eq!(
4204                channel_messages(channel),
4205                [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4206            )
4207        });
4208
4209        // Send a message from client B while it is disconnected.
4210        channel_b
4211            .update(cx_b, |channel, cx| {
4212                let task = channel
4213                    .send_message("can you see this?".to_string(), cx)
4214                    .unwrap();
4215                assert_eq!(
4216                    channel_messages(channel),
4217                    &[
4218                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4219                        ("user_b".to_string(), "can you see this?".to_string(), true)
4220                    ]
4221                );
4222                task
4223            })
4224            .await
4225            .unwrap_err();
4226
4227        // Send a message from client A while B is disconnected.
4228        channel_a
4229            .update(cx_a, |channel, cx| {
4230                channel
4231                    .send_message("oh, hi B.".to_string(), cx)
4232                    .unwrap()
4233                    .detach();
4234                let task = channel.send_message("sup".to_string(), cx).unwrap();
4235                assert_eq!(
4236                    channel_messages(channel),
4237                    &[
4238                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4239                        ("user_a".to_string(), "oh, hi B.".to_string(), true),
4240                        ("user_a".to_string(), "sup".to_string(), true)
4241                    ]
4242                );
4243                task
4244            })
4245            .await
4246            .unwrap();
4247
4248        // Give client B a chance to reconnect.
4249        server.allow_connections();
4250        cx_b.foreground().advance_clock(Duration::from_secs(10));
4251
4252        // Verify that B sees the new messages upon reconnection, as well as the message client B
4253        // sent while offline.
4254        channel_b
4255            .condition(&cx_b, |channel, _| {
4256                channel_messages(channel)
4257                    == [
4258                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4259                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4260                        ("user_a".to_string(), "sup".to_string(), false),
4261                        ("user_b".to_string(), "can you see this?".to_string(), false),
4262                    ]
4263            })
4264            .await;
4265
4266        // Ensure client A and B can communicate normally after reconnection.
4267        channel_a
4268            .update(cx_a, |channel, cx| {
4269                channel.send_message("you online?".to_string(), cx).unwrap()
4270            })
4271            .await
4272            .unwrap();
4273        channel_b
4274            .condition(&cx_b, |channel, _| {
4275                channel_messages(channel)
4276                    == [
4277                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4278                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4279                        ("user_a".to_string(), "sup".to_string(), false),
4280                        ("user_b".to_string(), "can you see this?".to_string(), false),
4281                        ("user_a".to_string(), "you online?".to_string(), false),
4282                    ]
4283            })
4284            .await;
4285
4286        channel_b
4287            .update(cx_b, |channel, cx| {
4288                channel.send_message("yep".to_string(), cx).unwrap()
4289            })
4290            .await
4291            .unwrap();
4292        channel_a
4293            .condition(&cx_a, |channel, _| {
4294                channel_messages(channel)
4295                    == [
4296                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4297                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4298                        ("user_a".to_string(), "sup".to_string(), false),
4299                        ("user_b".to_string(), "can you see this?".to_string(), false),
4300                        ("user_a".to_string(), "you online?".to_string(), false),
4301                        ("user_b".to_string(), "yep".to_string(), false),
4302                    ]
4303            })
4304            .await;
4305    }
4306
4307    #[gpui::test(iterations = 10)]
4308    async fn test_contacts(
4309        cx_a: &mut TestAppContext,
4310        cx_b: &mut TestAppContext,
4311        cx_c: &mut TestAppContext,
4312    ) {
4313        cx_a.foreground().forbid_parking();
4314        let lang_registry = Arc::new(LanguageRegistry::test());
4315        let fs = FakeFs::new(cx_a.background());
4316
4317        // Connect to a server as 3 clients.
4318        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4319        let client_a = server.create_client(cx_a, "user_a").await;
4320        let client_b = server.create_client(cx_b, "user_b").await;
4321        let client_c = server.create_client(cx_c, "user_c").await;
4322
4323        // Share a worktree as client A.
4324        fs.insert_tree(
4325            "/a",
4326            json!({
4327                ".zed.toml": r#"collaborators = ["user_b", "user_c"]"#,
4328            }),
4329        )
4330        .await;
4331
4332        let project_a = cx_a.update(|cx| {
4333            Project::local(
4334                client_a.clone(),
4335                client_a.user_store.clone(),
4336                lang_registry.clone(),
4337                fs.clone(),
4338                cx,
4339            )
4340        });
4341        let (worktree_a, _) = project_a
4342            .update(cx_a, |p, cx| {
4343                p.find_or_create_local_worktree("/a", true, cx)
4344            })
4345            .await
4346            .unwrap();
4347        worktree_a
4348            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
4349            .await;
4350
4351        client_a
4352            .user_store
4353            .condition(&cx_a, |user_store, _| {
4354                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4355            })
4356            .await;
4357        client_b
4358            .user_store
4359            .condition(&cx_b, |user_store, _| {
4360                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4361            })
4362            .await;
4363        client_c
4364            .user_store
4365            .condition(&cx_c, |user_store, _| {
4366                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4367            })
4368            .await;
4369
4370        let project_id = project_a
4371            .update(cx_a, |project, _| project.next_remote_id())
4372            .await;
4373        project_a
4374            .update(cx_a, |project, cx| project.share(cx))
4375            .await
4376            .unwrap();
4377
4378        let _project_b = Project::remote(
4379            project_id,
4380            client_b.clone(),
4381            client_b.user_store.clone(),
4382            lang_registry.clone(),
4383            fs.clone(),
4384            &mut cx_b.to_async(),
4385        )
4386        .await
4387        .unwrap();
4388
4389        client_a
4390            .user_store
4391            .condition(&cx_a, |user_store, _| {
4392                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4393            })
4394            .await;
4395        client_b
4396            .user_store
4397            .condition(&cx_b, |user_store, _| {
4398                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4399            })
4400            .await;
4401        client_c
4402            .user_store
4403            .condition(&cx_c, |user_store, _| {
4404                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4405            })
4406            .await;
4407
4408        project_a
4409            .condition(&cx_a, |project, _| {
4410                project.collaborators().contains_key(&client_b.peer_id)
4411            })
4412            .await;
4413
4414        cx_a.update(move |_| drop(project_a));
4415        client_a
4416            .user_store
4417            .condition(&cx_a, |user_store, _| contacts(user_store) == vec![])
4418            .await;
4419        client_b
4420            .user_store
4421            .condition(&cx_b, |user_store, _| contacts(user_store) == vec![])
4422            .await;
4423        client_c
4424            .user_store
4425            .condition(&cx_c, |user_store, _| contacts(user_store) == vec![])
4426            .await;
4427
4428        fn contacts(user_store: &UserStore) -> Vec<(&str, Vec<(&str, Vec<&str>)>)> {
4429            user_store
4430                .contacts()
4431                .iter()
4432                .map(|contact| {
4433                    let worktrees = contact
4434                        .projects
4435                        .iter()
4436                        .map(|p| {
4437                            (
4438                                p.worktree_root_names[0].as_str(),
4439                                p.guests.iter().map(|p| p.github_login.as_str()).collect(),
4440                            )
4441                        })
4442                        .collect();
4443                    (contact.user.github_login.as_str(), worktrees)
4444                })
4445                .collect()
4446        }
4447    }
4448
4449    #[gpui::test(iterations = 10)]
4450    async fn test_following(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4451        cx_a.foreground().forbid_parking();
4452        let fs = FakeFs::new(cx_a.background());
4453
4454        // 2 clients connect to a server.
4455        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4456        let mut client_a = server.create_client(cx_a, "user_a").await;
4457        let mut client_b = server.create_client(cx_b, "user_b").await;
4458        cx_a.update(editor::init);
4459        cx_b.update(editor::init);
4460
4461        // Client A shares a project.
4462        fs.insert_tree(
4463            "/a",
4464            json!({
4465                ".zed.toml": r#"collaborators = ["user_b"]"#,
4466                "1.txt": "one",
4467                "2.txt": "two",
4468                "3.txt": "three",
4469            }),
4470        )
4471        .await;
4472        let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
4473        project_a
4474            .update(cx_a, |project, cx| project.share(cx))
4475            .await
4476            .unwrap();
4477
4478        // Client B joins the project.
4479        let project_b = client_b
4480            .build_remote_project(
4481                project_a
4482                    .read_with(cx_a, |project, _| project.remote_id())
4483                    .unwrap(),
4484                cx_b,
4485            )
4486            .await;
4487
4488        // Client A opens some editors.
4489        let workspace_a = client_a.build_workspace(&project_a, cx_a);
4490        let pane_a = workspace_a.read_with(cx_a, |workspace, _| workspace.active_pane().clone());
4491        let editor_a1 = workspace_a
4492            .update(cx_a, |workspace, cx| {
4493                workspace.open_path((worktree_id, "1.txt"), cx)
4494            })
4495            .await
4496            .unwrap()
4497            .downcast::<Editor>()
4498            .unwrap();
4499        let editor_a2 = workspace_a
4500            .update(cx_a, |workspace, cx| {
4501                workspace.open_path((worktree_id, "2.txt"), cx)
4502            })
4503            .await
4504            .unwrap()
4505            .downcast::<Editor>()
4506            .unwrap();
4507
4508        // Client B opens an editor.
4509        let workspace_b = client_b.build_workspace(&project_b, cx_b);
4510        let editor_b1 = workspace_b
4511            .update(cx_b, |workspace, cx| {
4512                workspace.open_path((worktree_id, "1.txt"), cx)
4513            })
4514            .await
4515            .unwrap()
4516            .downcast::<Editor>()
4517            .unwrap();
4518
4519        let client_a_id = project_b.read_with(cx_b, |project, _| {
4520            project.collaborators().values().next().unwrap().peer_id
4521        });
4522        let client_b_id = project_a.read_with(cx_a, |project, _| {
4523            project.collaborators().values().next().unwrap().peer_id
4524        });
4525
4526        // When client B starts following client A, all visible view states are replicated to client B.
4527        editor_a1.update(cx_a, |editor, cx| editor.select_ranges([0..1], None, cx));
4528        editor_a2.update(cx_a, |editor, cx| editor.select_ranges([2..3], None, cx));
4529        workspace_b
4530            .update(cx_b, |workspace, cx| {
4531                workspace.toggle_follow(&client_a_id.into(), cx).unwrap()
4532            })
4533            .await
4534            .unwrap();
4535        let editor_b2 = workspace_b.read_with(cx_b, |workspace, cx| {
4536            workspace
4537                .active_item(cx)
4538                .unwrap()
4539                .downcast::<Editor>()
4540                .unwrap()
4541        });
4542        assert!(cx_b.read(|cx| editor_b2.is_focused(cx)));
4543        assert_eq!(
4544            editor_b2.read_with(cx_b, |editor, cx| editor.project_path(cx)),
4545            Some((worktree_id, "2.txt").into())
4546        );
4547        assert_eq!(
4548            editor_b2.read_with(cx_b, |editor, cx| editor.selected_ranges(cx)),
4549            vec![2..3]
4550        );
4551        assert_eq!(
4552            editor_b1.read_with(cx_b, |editor, cx| editor.selected_ranges(cx)),
4553            vec![0..1]
4554        );
4555
4556        // When client A activates a different editor, client B does so as well.
4557        workspace_a.update(cx_a, |workspace, cx| {
4558            workspace.activate_item(&editor_a1, cx)
4559        });
4560        workspace_b
4561            .condition(cx_b, |workspace, cx| {
4562                workspace.active_item(cx).unwrap().id() == editor_b1.id()
4563            })
4564            .await;
4565
4566        // When client A navigates back and forth, client B does so as well.
4567        workspace_a
4568            .update(cx_a, |workspace, cx| {
4569                workspace::Pane::go_back(workspace, None, cx)
4570            })
4571            .await;
4572        workspace_b
4573            .condition(cx_b, |workspace, cx| {
4574                workspace.active_item(cx).unwrap().id() == editor_b2.id()
4575            })
4576            .await;
4577
4578        workspace_a
4579            .update(cx_a, |workspace, cx| {
4580                workspace::Pane::go_forward(workspace, None, cx)
4581            })
4582            .await;
4583        workspace_b
4584            .condition(cx_b, |workspace, cx| {
4585                workspace.active_item(cx).unwrap().id() == editor_b1.id()
4586            })
4587            .await;
4588
4589        // Changes to client A's editor are reflected on client B.
4590        editor_a1.update(cx_a, |editor, cx| {
4591            editor.select_ranges([1..1, 2..2], None, cx);
4592        });
4593        editor_b1
4594            .condition(cx_b, |editor, cx| {
4595                editor.selected_ranges(cx) == vec![1..1, 2..2]
4596            })
4597            .await;
4598
4599        editor_a1.update(cx_a, |editor, cx| editor.set_text("TWO", cx));
4600        editor_b1
4601            .condition(cx_b, |editor, cx| editor.text(cx) == "TWO")
4602            .await;
4603
4604        editor_a1.update(cx_a, |editor, cx| {
4605            editor.select_ranges([3..3], None, cx);
4606            editor.set_scroll_position(vec2f(0., 100.), cx);
4607        });
4608        editor_b1
4609            .condition(cx_b, |editor, cx| editor.selected_ranges(cx) == vec![3..3])
4610            .await;
4611
4612        // After unfollowing, client B stops receiving updates from client A.
4613        workspace_b.update(cx_b, |workspace, cx| {
4614            workspace.unfollow(&workspace.active_pane().clone(), cx)
4615        });
4616        workspace_a.update(cx_a, |workspace, cx| {
4617            workspace.activate_item(&editor_a2, cx)
4618        });
4619        cx_a.foreground().run_until_parked();
4620        assert_eq!(
4621            workspace_b.read_with(cx_b, |workspace, cx| workspace
4622                .active_item(cx)
4623                .unwrap()
4624                .id()),
4625            editor_b1.id()
4626        );
4627
4628        // Client A starts following client B.
4629        workspace_a
4630            .update(cx_a, |workspace, cx| {
4631                workspace.toggle_follow(&client_b_id.into(), cx).unwrap()
4632            })
4633            .await
4634            .unwrap();
4635        assert_eq!(
4636            workspace_a.read_with(cx_a, |workspace, _| workspace.leader_for_pane(&pane_a)),
4637            Some(client_b_id)
4638        );
4639        assert_eq!(
4640            workspace_a.read_with(cx_a, |workspace, cx| workspace
4641                .active_item(cx)
4642                .unwrap()
4643                .id()),
4644            editor_a1.id()
4645        );
4646
4647        // Following interrupts when client B disconnects.
4648        client_b.disconnect(&cx_b.to_async()).unwrap();
4649        cx_a.foreground().run_until_parked();
4650        assert_eq!(
4651            workspace_a.read_with(cx_a, |workspace, _| workspace.leader_for_pane(&pane_a)),
4652            None
4653        );
4654    }
4655
4656    #[gpui::test(iterations = 10)]
4657    async fn test_peers_following_each_other(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4658        cx_a.foreground().forbid_parking();
4659        let fs = FakeFs::new(cx_a.background());
4660
4661        // 2 clients connect to a server.
4662        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4663        let mut client_a = server.create_client(cx_a, "user_a").await;
4664        let mut client_b = server.create_client(cx_b, "user_b").await;
4665        cx_a.update(editor::init);
4666        cx_b.update(editor::init);
4667
4668        // Client A shares a project.
4669        fs.insert_tree(
4670            "/a",
4671            json!({
4672                ".zed.toml": r#"collaborators = ["user_b"]"#,
4673                "1.txt": "one",
4674                "2.txt": "two",
4675                "3.txt": "three",
4676                "4.txt": "four",
4677            }),
4678        )
4679        .await;
4680        let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
4681        project_a
4682            .update(cx_a, |project, cx| project.share(cx))
4683            .await
4684            .unwrap();
4685
4686        // Client B joins the project.
4687        let project_b = client_b
4688            .build_remote_project(
4689                project_a
4690                    .read_with(cx_a, |project, _| project.remote_id())
4691                    .unwrap(),
4692                cx_b,
4693            )
4694            .await;
4695
4696        // Client A opens some editors.
4697        let workspace_a = client_a.build_workspace(&project_a, cx_a);
4698        let pane_a1 = workspace_a.read_with(cx_a, |workspace, _| workspace.active_pane().clone());
4699        let _editor_a1 = workspace_a
4700            .update(cx_a, |workspace, cx| {
4701                workspace.open_path((worktree_id, "1.txt"), cx)
4702            })
4703            .await
4704            .unwrap()
4705            .downcast::<Editor>()
4706            .unwrap();
4707
4708        // Client B opens an editor.
4709        let workspace_b = client_b.build_workspace(&project_b, cx_b);
4710        let pane_b1 = workspace_b.read_with(cx_b, |workspace, _| workspace.active_pane().clone());
4711        let _editor_b1 = workspace_b
4712            .update(cx_b, |workspace, cx| {
4713                workspace.open_path((worktree_id, "2.txt"), cx)
4714            })
4715            .await
4716            .unwrap()
4717            .downcast::<Editor>()
4718            .unwrap();
4719
4720        // Clients A and B follow each other in split panes
4721        workspace_a
4722            .update(cx_a, |workspace, cx| {
4723                workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
4724                assert_ne!(*workspace.active_pane(), pane_a1);
4725                let leader_id = *project_a.read(cx).collaborators().keys().next().unwrap();
4726                workspace
4727                    .toggle_follow(&workspace::ToggleFollow(leader_id), cx)
4728                    .unwrap()
4729            })
4730            .await
4731            .unwrap();
4732        workspace_b
4733            .update(cx_b, |workspace, cx| {
4734                workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
4735                assert_ne!(*workspace.active_pane(), pane_b1);
4736                let leader_id = *project_b.read(cx).collaborators().keys().next().unwrap();
4737                workspace
4738                    .toggle_follow(&workspace::ToggleFollow(leader_id), cx)
4739                    .unwrap()
4740            })
4741            .await
4742            .unwrap();
4743
4744        workspace_a
4745            .update(cx_a, |workspace, cx| {
4746                workspace.activate_next_pane(cx);
4747                assert_eq!(*workspace.active_pane(), pane_a1);
4748                workspace.open_path((worktree_id, "3.txt"), cx)
4749            })
4750            .await
4751            .unwrap();
4752        workspace_b
4753            .update(cx_b, |workspace, cx| {
4754                workspace.activate_next_pane(cx);
4755                assert_eq!(*workspace.active_pane(), pane_b1);
4756                workspace.open_path((worktree_id, "4.txt"), cx)
4757            })
4758            .await
4759            .unwrap();
4760        cx_a.foreground().run_until_parked();
4761
4762        // Ensure leader updates don't change the active pane of followers
4763        workspace_a.read_with(cx_a, |workspace, _| {
4764            assert_eq!(*workspace.active_pane(), pane_a1);
4765        });
4766        workspace_b.read_with(cx_b, |workspace, _| {
4767            assert_eq!(*workspace.active_pane(), pane_b1);
4768        });
4769
4770        // Ensure peers following each other doesn't cause an infinite loop.
4771        assert_eq!(
4772            workspace_a.read_with(cx_a, |workspace, cx| workspace
4773                .active_item(cx)
4774                .unwrap()
4775                .project_path(cx)),
4776            Some((worktree_id, "3.txt").into())
4777        );
4778        workspace_a.update(cx_a, |workspace, cx| {
4779            assert_eq!(
4780                workspace.active_item(cx).unwrap().project_path(cx),
4781                Some((worktree_id, "3.txt").into())
4782            );
4783            workspace.activate_next_pane(cx);
4784            assert_eq!(
4785                workspace.active_item(cx).unwrap().project_path(cx),
4786                Some((worktree_id, "4.txt").into())
4787            );
4788        });
4789        workspace_b.update(cx_b, |workspace, cx| {
4790            assert_eq!(
4791                workspace.active_item(cx).unwrap().project_path(cx),
4792                Some((worktree_id, "4.txt").into())
4793            );
4794            workspace.activate_next_pane(cx);
4795            assert_eq!(
4796                workspace.active_item(cx).unwrap().project_path(cx),
4797                Some((worktree_id, "3.txt").into())
4798            );
4799        });
4800    }
4801
4802    #[gpui::test(iterations = 10)]
4803    async fn test_auto_unfollowing(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4804        cx_a.foreground().forbid_parking();
4805        let fs = FakeFs::new(cx_a.background());
4806
4807        // 2 clients connect to a server.
4808        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4809        let mut client_a = server.create_client(cx_a, "user_a").await;
4810        let mut client_b = server.create_client(cx_b, "user_b").await;
4811        cx_a.update(editor::init);
4812        cx_b.update(editor::init);
4813
4814        // Client A shares a project.
4815        fs.insert_tree(
4816            "/a",
4817            json!({
4818                ".zed.toml": r#"collaborators = ["user_b"]"#,
4819                "1.txt": "one",
4820                "2.txt": "two",
4821                "3.txt": "three",
4822            }),
4823        )
4824        .await;
4825        let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
4826        project_a
4827            .update(cx_a, |project, cx| project.share(cx))
4828            .await
4829            .unwrap();
4830
4831        // Client B joins the project.
4832        let project_b = client_b
4833            .build_remote_project(
4834                project_a
4835                    .read_with(cx_a, |project, _| project.remote_id())
4836                    .unwrap(),
4837                cx_b,
4838            )
4839            .await;
4840
4841        // Client A opens some editors.
4842        let workspace_a = client_a.build_workspace(&project_a, cx_a);
4843        let _editor_a1 = workspace_a
4844            .update(cx_a, |workspace, cx| {
4845                workspace.open_path((worktree_id, "1.txt"), cx)
4846            })
4847            .await
4848            .unwrap()
4849            .downcast::<Editor>()
4850            .unwrap();
4851
4852        // Client B starts following client A.
4853        let workspace_b = client_b.build_workspace(&project_b, cx_b);
4854        let pane_b = workspace_b.read_with(cx_b, |workspace, _| workspace.active_pane().clone());
4855        let leader_id = project_b.read_with(cx_b, |project, _| {
4856            project.collaborators().values().next().unwrap().peer_id
4857        });
4858        workspace_b
4859            .update(cx_b, |workspace, cx| {
4860                workspace.toggle_follow(&leader_id.into(), cx).unwrap()
4861            })
4862            .await
4863            .unwrap();
4864        assert_eq!(
4865            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4866            Some(leader_id)
4867        );
4868        let editor_b2 = workspace_b.read_with(cx_b, |workspace, cx| {
4869            workspace
4870                .active_item(cx)
4871                .unwrap()
4872                .downcast::<Editor>()
4873                .unwrap()
4874        });
4875
4876        // When client B moves, it automatically stops following client A.
4877        editor_b2.update(cx_b, |editor, cx| editor.move_right(&editor::MoveRight, cx));
4878        assert_eq!(
4879            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4880            None
4881        );
4882
4883        workspace_b
4884            .update(cx_b, |workspace, cx| {
4885                workspace.toggle_follow(&leader_id.into(), cx).unwrap()
4886            })
4887            .await
4888            .unwrap();
4889        assert_eq!(
4890            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4891            Some(leader_id)
4892        );
4893
4894        // When client B edits, it automatically stops following client A.
4895        editor_b2.update(cx_b, |editor, cx| editor.insert("X", cx));
4896        assert_eq!(
4897            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4898            None
4899        );
4900
4901        workspace_b
4902            .update(cx_b, |workspace, cx| {
4903                workspace.toggle_follow(&leader_id.into(), cx).unwrap()
4904            })
4905            .await
4906            .unwrap();
4907        assert_eq!(
4908            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4909            Some(leader_id)
4910        );
4911
4912        // When client B scrolls, it automatically stops following client A.
4913        editor_b2.update(cx_b, |editor, cx| {
4914            editor.set_scroll_position(vec2f(0., 3.), cx)
4915        });
4916        assert_eq!(
4917            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4918            None
4919        );
4920
4921        workspace_b
4922            .update(cx_b, |workspace, cx| {
4923                workspace.toggle_follow(&leader_id.into(), cx).unwrap()
4924            })
4925            .await
4926            .unwrap();
4927        assert_eq!(
4928            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4929            Some(leader_id)
4930        );
4931
4932        // When client B activates a different pane, it continues following client A in the original pane.
4933        workspace_b.update(cx_b, |workspace, cx| {
4934            workspace.split_pane(pane_b.clone(), SplitDirection::Right, cx)
4935        });
4936        assert_eq!(
4937            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4938            Some(leader_id)
4939        );
4940
4941        workspace_b.update(cx_b, |workspace, cx| workspace.activate_next_pane(cx));
4942        assert_eq!(
4943            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4944            Some(leader_id)
4945        );
4946
4947        // When client B activates a different item in the original pane, it automatically stops following client A.
4948        workspace_b
4949            .update(cx_b, |workspace, cx| {
4950                workspace.open_path((worktree_id, "2.txt"), cx)
4951            })
4952            .await
4953            .unwrap();
4954        assert_eq!(
4955            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4956            None
4957        );
4958    }
4959
4960    #[gpui::test(iterations = 100)]
4961    async fn test_random_collaboration(cx: &mut TestAppContext, rng: StdRng) {
4962        cx.foreground().forbid_parking();
4963        let max_peers = env::var("MAX_PEERS")
4964            .map(|i| i.parse().expect("invalid `MAX_PEERS` variable"))
4965            .unwrap_or(5);
4966        let max_operations = env::var("OPERATIONS")
4967            .map(|i| i.parse().expect("invalid `OPERATIONS` variable"))
4968            .unwrap_or(10);
4969
4970        let rng = Arc::new(Mutex::new(rng));
4971
4972        let guest_lang_registry = Arc::new(LanguageRegistry::test());
4973        let host_language_registry = Arc::new(LanguageRegistry::test());
4974
4975        let fs = FakeFs::new(cx.background());
4976        fs.insert_tree(
4977            "/_collab",
4978            json!({
4979                ".zed.toml": r#"collaborators = ["guest-1", "guest-2", "guest-3", "guest-4", "guest-5"]"#
4980            }),
4981        )
4982        .await;
4983
4984        let operations = Rc::new(Cell::new(0));
4985        let mut server = TestServer::start(cx.foreground(), cx.background()).await;
4986        let mut clients = Vec::new();
4987        let mut user_ids = Vec::new();
4988        let files = Arc::new(Mutex::new(Vec::new()));
4989
4990        let mut next_entity_id = 100000;
4991        let mut host_cx = TestAppContext::new(
4992            cx.foreground_platform(),
4993            cx.platform(),
4994            cx.foreground(),
4995            cx.background(),
4996            cx.font_cache(),
4997            cx.leak_detector(),
4998            next_entity_id,
4999        );
5000        let host = server.create_client(&mut host_cx, "host").await;
5001        let host_project = host_cx.update(|cx| {
5002            Project::local(
5003                host.client.clone(),
5004                host.user_store.clone(),
5005                host_language_registry.clone(),
5006                fs.clone(),
5007                cx,
5008            )
5009        });
5010        let host_project_id = host_project
5011            .update(&mut host_cx, |p, _| p.next_remote_id())
5012            .await;
5013
5014        let (collab_worktree, _) = host_project
5015            .update(&mut host_cx, |project, cx| {
5016                project.find_or_create_local_worktree("/_collab", true, cx)
5017            })
5018            .await
5019            .unwrap();
5020        collab_worktree
5021            .read_with(&host_cx, |tree, _| tree.as_local().unwrap().scan_complete())
5022            .await;
5023        host_project
5024            .update(&mut host_cx, |project, cx| project.share(cx))
5025            .await
5026            .unwrap();
5027
5028        // Set up fake language servers.
5029        let mut language = Language::new(
5030            LanguageConfig {
5031                name: "Rust".into(),
5032                path_suffixes: vec!["rs".to_string()],
5033                ..Default::default()
5034            },
5035            None,
5036        );
5037        let _fake_servers = language.set_fake_lsp_adapter(FakeLspAdapter {
5038            name: "the-fake-language-server",
5039            capabilities: lsp::LanguageServer::full_capabilities(),
5040            initializer: Some(Box::new({
5041                let rng = rng.clone();
5042                let files = files.clone();
5043                let project = host_project.downgrade();
5044                move |fake_server: &mut FakeLanguageServer| {
5045                    fake_server.handle_request::<lsp::request::Completion, _, _>(
5046                        |_, _| async move {
5047                            Ok(Some(lsp::CompletionResponse::Array(vec![
5048                                lsp::CompletionItem {
5049                                    text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
5050                                        range: lsp::Range::new(
5051                                            lsp::Position::new(0, 0),
5052                                            lsp::Position::new(0, 0),
5053                                        ),
5054                                        new_text: "the-new-text".to_string(),
5055                                    })),
5056                                    ..Default::default()
5057                                },
5058                            ])))
5059                        },
5060                    );
5061
5062                    fake_server.handle_request::<lsp::request::CodeActionRequest, _, _>(
5063                        |_, _| async move {
5064                            Ok(Some(vec![lsp::CodeActionOrCommand::CodeAction(
5065                                lsp::CodeAction {
5066                                    title: "the-code-action".to_string(),
5067                                    ..Default::default()
5068                                },
5069                            )]))
5070                        },
5071                    );
5072
5073                    fake_server.handle_request::<lsp::request::PrepareRenameRequest, _, _>(
5074                        |params, _| async move {
5075                            Ok(Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
5076                                params.position,
5077                                params.position,
5078                            ))))
5079                        },
5080                    );
5081
5082                    fake_server.handle_request::<lsp::request::GotoDefinition, _, _>({
5083                        let files = files.clone();
5084                        let rng = rng.clone();
5085                        move |_, _| {
5086                            let files = files.clone();
5087                            let rng = rng.clone();
5088                            async move {
5089                                let files = files.lock();
5090                                let mut rng = rng.lock();
5091                                let count = rng.gen_range::<usize, _>(1..3);
5092                                let files = (0..count)
5093                                    .map(|_| files.choose(&mut *rng).unwrap())
5094                                    .collect::<Vec<_>>();
5095                                log::info!("LSP: Returning definitions in files {:?}", &files);
5096                                Ok(Some(lsp::GotoDefinitionResponse::Array(
5097                                    files
5098                                        .into_iter()
5099                                        .map(|file| lsp::Location {
5100                                            uri: lsp::Url::from_file_path(file).unwrap(),
5101                                            range: Default::default(),
5102                                        })
5103                                        .collect(),
5104                                )))
5105                            }
5106                        }
5107                    });
5108
5109                    fake_server.handle_request::<lsp::request::DocumentHighlightRequest, _, _>({
5110                        let rng = rng.clone();
5111                        let project = project.clone();
5112                        move |params, mut cx| {
5113                            let highlights = if let Some(project) = project.upgrade(&cx) {
5114                                project.update(&mut cx, |project, cx| {
5115                                    let path = params
5116                                        .text_document_position_params
5117                                        .text_document
5118                                        .uri
5119                                        .to_file_path()
5120                                        .unwrap();
5121                                    let (worktree, relative_path) =
5122                                        project.find_local_worktree(&path, cx)?;
5123                                    let project_path =
5124                                        ProjectPath::from((worktree.read(cx).id(), relative_path));
5125                                    let buffer =
5126                                        project.get_open_buffer(&project_path, cx)?.read(cx);
5127
5128                                    let mut highlights = Vec::new();
5129                                    let highlight_count = rng.lock().gen_range(1..=5);
5130                                    let mut prev_end = 0;
5131                                    for _ in 0..highlight_count {
5132                                        let range =
5133                                            buffer.random_byte_range(prev_end, &mut *rng.lock());
5134
5135                                        highlights.push(lsp::DocumentHighlight {
5136                                            range: range_to_lsp(range.to_point_utf16(buffer)),
5137                                            kind: Some(lsp::DocumentHighlightKind::READ),
5138                                        });
5139                                        prev_end = range.end;
5140                                    }
5141                                    Some(highlights)
5142                                })
5143                            } else {
5144                                None
5145                            };
5146                            async move { Ok(highlights) }
5147                        }
5148                    });
5149                }
5150            })),
5151            ..Default::default()
5152        });
5153        host_language_registry.add(Arc::new(language));
5154
5155        let host_disconnected = Rc::new(AtomicBool::new(false));
5156        user_ids.push(host.current_user_id(&host_cx));
5157        clients.push(cx.foreground().spawn(host.simulate_host(
5158            host_project,
5159            files,
5160            operations.clone(),
5161            max_operations,
5162            rng.clone(),
5163            host_cx,
5164        )));
5165
5166        while operations.get() < max_operations {
5167            cx.background().simulate_random_delay().await;
5168            if clients.len() >= max_peers {
5169                break;
5170            } else if rng.lock().gen_bool(0.05) {
5171                operations.set(operations.get() + 1);
5172
5173                let guest_id = clients.len();
5174                log::info!("Adding guest {}", guest_id);
5175                next_entity_id += 100000;
5176                let mut guest_cx = TestAppContext::new(
5177                    cx.foreground_platform(),
5178                    cx.platform(),
5179                    cx.foreground(),
5180                    cx.background(),
5181                    cx.font_cache(),
5182                    cx.leak_detector(),
5183                    next_entity_id,
5184                );
5185                let guest = server
5186                    .create_client(&mut guest_cx, &format!("guest-{}", guest_id))
5187                    .await;
5188                let guest_project = Project::remote(
5189                    host_project_id,
5190                    guest.client.clone(),
5191                    guest.user_store.clone(),
5192                    guest_lang_registry.clone(),
5193                    FakeFs::new(cx.background()),
5194                    &mut guest_cx.to_async(),
5195                )
5196                .await
5197                .unwrap();
5198                user_ids.push(guest.current_user_id(&guest_cx));
5199                clients.push(cx.foreground().spawn(guest.simulate_guest(
5200                    guest_id,
5201                    guest_project,
5202                    operations.clone(),
5203                    max_operations,
5204                    rng.clone(),
5205                    host_disconnected.clone(),
5206                    guest_cx,
5207                )));
5208
5209                log::info!("Guest {} added", guest_id);
5210            } else if rng.lock().gen_bool(0.05) {
5211                host_disconnected.store(true, SeqCst);
5212                server.disconnect_client(user_ids[0]);
5213                cx.foreground().advance_clock(RECEIVE_TIMEOUT);
5214                let mut clients = futures::future::join_all(clients).await;
5215                cx.foreground().run_until_parked();
5216
5217                let (host, mut host_cx) = clients.remove(0);
5218                host.project
5219                    .as_ref()
5220                    .unwrap()
5221                    .read_with(&host_cx, |project, _| assert!(!project.is_shared()));
5222                for (guest, mut guest_cx) in clients {
5223                    let contacts = server
5224                        .store
5225                        .read()
5226                        .contacts_for_user(guest.current_user_id(&guest_cx));
5227                    assert!(!contacts
5228                        .iter()
5229                        .flat_map(|contact| &contact.projects)
5230                        .any(|project| project.id == host_project_id));
5231                    guest
5232                        .project
5233                        .as_ref()
5234                        .unwrap()
5235                        .read_with(&guest_cx, |project, _| assert!(project.is_read_only()));
5236                    guest_cx.update(|_| drop(guest));
5237                }
5238                host_cx.update(|_| drop(host));
5239
5240                return;
5241            }
5242        }
5243
5244        let mut clients = futures::future::join_all(clients).await;
5245        cx.foreground().run_until_parked();
5246
5247        let (host_client, mut host_cx) = clients.remove(0);
5248        let host_project = host_client.project.as_ref().unwrap();
5249        let host_worktree_snapshots = host_project.read_with(&host_cx, |project, cx| {
5250            project
5251                .worktrees(cx)
5252                .map(|worktree| {
5253                    let snapshot = worktree.read(cx).snapshot();
5254                    (snapshot.id(), snapshot)
5255                })
5256                .collect::<BTreeMap<_, _>>()
5257        });
5258
5259        host_client
5260            .project
5261            .as_ref()
5262            .unwrap()
5263            .read_with(&host_cx, |project, cx| project.check_invariants(cx));
5264
5265        for (guest_client, mut guest_cx) in clients.into_iter() {
5266            let guest_id = guest_client.client.id();
5267            let worktree_snapshots =
5268                guest_client
5269                    .project
5270                    .as_ref()
5271                    .unwrap()
5272                    .read_with(&guest_cx, |project, cx| {
5273                        project
5274                            .worktrees(cx)
5275                            .map(|worktree| {
5276                                let worktree = worktree.read(cx);
5277                                (worktree.id(), worktree.snapshot())
5278                            })
5279                            .collect::<BTreeMap<_, _>>()
5280                    });
5281
5282            assert_eq!(
5283                worktree_snapshots.keys().collect::<Vec<_>>(),
5284                host_worktree_snapshots.keys().collect::<Vec<_>>(),
5285                "guest {} has different worktrees than the host",
5286                guest_id
5287            );
5288            for (id, host_snapshot) in &host_worktree_snapshots {
5289                let guest_snapshot = &worktree_snapshots[id];
5290                assert_eq!(
5291                    guest_snapshot.root_name(),
5292                    host_snapshot.root_name(),
5293                    "guest {} has different root name than the host for worktree {}",
5294                    guest_id,
5295                    id
5296                );
5297                assert_eq!(
5298                    guest_snapshot.entries(false).collect::<Vec<_>>(),
5299                    host_snapshot.entries(false).collect::<Vec<_>>(),
5300                    "guest {} has different snapshot than the host for worktree {}",
5301                    guest_id,
5302                    id
5303                );
5304            }
5305
5306            guest_client
5307                .project
5308                .as_ref()
5309                .unwrap()
5310                .read_with(&guest_cx, |project, cx| project.check_invariants(cx));
5311
5312            for guest_buffer in &guest_client.buffers {
5313                let buffer_id = guest_buffer.read_with(&guest_cx, |buffer, _| buffer.remote_id());
5314                let host_buffer = host_project.read_with(&host_cx, |project, cx| {
5315                    project.buffer_for_id(buffer_id, cx).expect(&format!(
5316                        "host does not have buffer for guest:{}, peer:{}, id:{}",
5317                        guest_id, guest_client.peer_id, buffer_id
5318                    ))
5319                });
5320                let path = host_buffer
5321                    .read_with(&host_cx, |buffer, cx| buffer.file().unwrap().full_path(cx));
5322
5323                assert_eq!(
5324                    guest_buffer.read_with(&guest_cx, |buffer, _| buffer.deferred_ops_len()),
5325                    0,
5326                    "guest {}, buffer {}, path {:?} has deferred operations",
5327                    guest_id,
5328                    buffer_id,
5329                    path,
5330                );
5331                assert_eq!(
5332                    guest_buffer.read_with(&guest_cx, |buffer, _| buffer.text()),
5333                    host_buffer.read_with(&host_cx, |buffer, _| buffer.text()),
5334                    "guest {}, buffer {}, path {:?}, differs from the host's buffer",
5335                    guest_id,
5336                    buffer_id,
5337                    path
5338                );
5339            }
5340
5341            guest_cx.update(|_| drop(guest_client));
5342        }
5343
5344        host_cx.update(|_| drop(host_client));
5345    }
5346
5347    struct TestServer {
5348        peer: Arc<Peer>,
5349        app_state: Arc<AppState>,
5350        server: Arc<Server>,
5351        foreground: Rc<executor::Foreground>,
5352        notifications: mpsc::UnboundedReceiver<()>,
5353        connection_killers: Arc<Mutex<HashMap<UserId, barrier::Sender>>>,
5354        forbid_connections: Arc<AtomicBool>,
5355        _test_db: TestDb,
5356    }
5357
5358    impl TestServer {
5359        async fn start(
5360            foreground: Rc<executor::Foreground>,
5361            background: Arc<executor::Background>,
5362        ) -> Self {
5363            let test_db = TestDb::fake(background);
5364            let app_state = Self::build_app_state(&test_db).await;
5365            let peer = Peer::new();
5366            let notifications = mpsc::unbounded();
5367            let server = Server::new(app_state.clone(), peer.clone(), Some(notifications.0));
5368            Self {
5369                peer,
5370                app_state,
5371                server,
5372                foreground,
5373                notifications: notifications.1,
5374                connection_killers: Default::default(),
5375                forbid_connections: Default::default(),
5376                _test_db: test_db,
5377            }
5378        }
5379
5380        async fn create_client(&mut self, cx: &mut TestAppContext, name: &str) -> TestClient {
5381            cx.update(|cx| {
5382                let settings = Settings::test(cx);
5383                cx.set_global(settings);
5384            });
5385
5386            let http = FakeHttpClient::with_404_response();
5387            let user_id = self.app_state.db.create_user(name, false).await.unwrap();
5388            let client_name = name.to_string();
5389            let mut client = Client::new(http.clone());
5390            let server = self.server.clone();
5391            let connection_killers = self.connection_killers.clone();
5392            let forbid_connections = self.forbid_connections.clone();
5393            let (connection_id_tx, mut connection_id_rx) = mpsc::channel(16);
5394
5395            Arc::get_mut(&mut client)
5396                .unwrap()
5397                .override_authenticate(move |cx| {
5398                    cx.spawn(|_| async move {
5399                        let access_token = "the-token".to_string();
5400                        Ok(Credentials {
5401                            user_id: user_id.0 as u64,
5402                            access_token,
5403                        })
5404                    })
5405                })
5406                .override_establish_connection(move |credentials, cx| {
5407                    assert_eq!(credentials.user_id, user_id.0 as u64);
5408                    assert_eq!(credentials.access_token, "the-token");
5409
5410                    let server = server.clone();
5411                    let connection_killers = connection_killers.clone();
5412                    let forbid_connections = forbid_connections.clone();
5413                    let client_name = client_name.clone();
5414                    let connection_id_tx = connection_id_tx.clone();
5415                    cx.spawn(move |cx| async move {
5416                        if forbid_connections.load(SeqCst) {
5417                            Err(EstablishConnectionError::other(anyhow!(
5418                                "server is forbidding connections"
5419                            )))
5420                        } else {
5421                            let (client_conn, server_conn, kill_conn) =
5422                                Connection::in_memory(cx.background());
5423                            connection_killers.lock().insert(user_id, kill_conn);
5424                            cx.background()
5425                                .spawn(server.handle_connection(
5426                                    server_conn,
5427                                    client_name,
5428                                    user_id,
5429                                    Some(connection_id_tx),
5430                                    cx.background(),
5431                                ))
5432                                .detach();
5433                            Ok(client_conn)
5434                        }
5435                    })
5436                });
5437
5438            client
5439                .authenticate_and_connect(false, &cx.to_async())
5440                .await
5441                .unwrap();
5442
5443            Channel::init(&client);
5444            Project::init(&client);
5445            cx.update(|cx| {
5446                workspace::init(&client, cx);
5447            });
5448
5449            let peer_id = PeerId(connection_id_rx.next().await.unwrap().0);
5450            let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http, cx));
5451
5452            let client = TestClient {
5453                client,
5454                peer_id,
5455                user_store,
5456                language_registry: Arc::new(LanguageRegistry::test()),
5457                project: Default::default(),
5458                buffers: Default::default(),
5459            };
5460            client.wait_for_current_user(cx).await;
5461            client
5462        }
5463
5464        fn disconnect_client(&self, user_id: UserId) {
5465            self.connection_killers.lock().remove(&user_id);
5466        }
5467
5468        fn forbid_connections(&self) {
5469            self.forbid_connections.store(true, SeqCst);
5470        }
5471
5472        fn allow_connections(&self) {
5473            self.forbid_connections.store(false, SeqCst);
5474        }
5475
5476        async fn build_app_state(test_db: &TestDb) -> Arc<AppState> {
5477            let mut config = Config::default();
5478            config.session_secret = "a".repeat(32);
5479            config.database_url = test_db.url.clone();
5480            let github_client = github::AppClient::test();
5481            Arc::new(AppState {
5482                db: test_db.db().clone(),
5483                handlebars: Default::default(),
5484                auth_client: auth::build_client("", ""),
5485                repo_client: github::RepoClient::test(&github_client),
5486                github_client,
5487                config,
5488            })
5489        }
5490
5491        async fn state<'a>(&'a self) -> RwLockReadGuard<'a, Store> {
5492            self.server.store.read()
5493        }
5494
5495        async fn condition<F>(&mut self, mut predicate: F)
5496        where
5497            F: FnMut(&Store) -> bool,
5498        {
5499            async_std::future::timeout(Duration::from_millis(500), async {
5500                while !(predicate)(&*self.server.store.read()) {
5501                    self.foreground.start_waiting();
5502                    self.notifications.next().await;
5503                    self.foreground.finish_waiting();
5504                }
5505            })
5506            .await
5507            .expect("condition timed out");
5508        }
5509    }
5510
5511    impl Deref for TestServer {
5512        type Target = Server;
5513
5514        fn deref(&self) -> &Self::Target {
5515            &self.server
5516        }
5517    }
5518
5519    impl Drop for TestServer {
5520        fn drop(&mut self) {
5521            self.peer.reset();
5522        }
5523    }
5524
5525    struct TestClient {
5526        client: Arc<Client>,
5527        pub peer_id: PeerId,
5528        pub user_store: ModelHandle<UserStore>,
5529        language_registry: Arc<LanguageRegistry>,
5530        project: Option<ModelHandle<Project>>,
5531        buffers: HashSet<ModelHandle<language::Buffer>>,
5532    }
5533
5534    impl Deref for TestClient {
5535        type Target = Arc<Client>;
5536
5537        fn deref(&self) -> &Self::Target {
5538            &self.client
5539        }
5540    }
5541
5542    impl TestClient {
5543        pub fn current_user_id(&self, cx: &TestAppContext) -> UserId {
5544            UserId::from_proto(
5545                self.user_store
5546                    .read_with(cx, |user_store, _| user_store.current_user().unwrap().id),
5547            )
5548        }
5549
5550        async fn wait_for_current_user(&self, cx: &TestAppContext) {
5551            let mut authed_user = self
5552                .user_store
5553                .read_with(cx, |user_store, _| user_store.watch_current_user());
5554            while authed_user.next().await.unwrap().is_none() {}
5555        }
5556
5557        async fn build_local_project(
5558            &mut self,
5559            fs: Arc<FakeFs>,
5560            root_path: impl AsRef<Path>,
5561            cx: &mut TestAppContext,
5562        ) -> (ModelHandle<Project>, WorktreeId) {
5563            let project = cx.update(|cx| {
5564                Project::local(
5565                    self.client.clone(),
5566                    self.user_store.clone(),
5567                    self.language_registry.clone(),
5568                    fs,
5569                    cx,
5570                )
5571            });
5572            self.project = Some(project.clone());
5573            let (worktree, _) = project
5574                .update(cx, |p, cx| {
5575                    p.find_or_create_local_worktree(root_path, true, cx)
5576                })
5577                .await
5578                .unwrap();
5579            worktree
5580                .read_with(cx, |tree, _| tree.as_local().unwrap().scan_complete())
5581                .await;
5582            project
5583                .update(cx, |project, _| project.next_remote_id())
5584                .await;
5585            (project, worktree.read_with(cx, |tree, _| tree.id()))
5586        }
5587
5588        async fn build_remote_project(
5589            &mut self,
5590            project_id: u64,
5591            cx: &mut TestAppContext,
5592        ) -> ModelHandle<Project> {
5593            let project = Project::remote(
5594                project_id,
5595                self.client.clone(),
5596                self.user_store.clone(),
5597                self.language_registry.clone(),
5598                FakeFs::new(cx.background()),
5599                &mut cx.to_async(),
5600            )
5601            .await
5602            .unwrap();
5603            self.project = Some(project.clone());
5604            project
5605        }
5606
5607        fn build_workspace(
5608            &self,
5609            project: &ModelHandle<Project>,
5610            cx: &mut TestAppContext,
5611        ) -> ViewHandle<Workspace> {
5612            let (window_id, _) = cx.add_window(|_| EmptyView);
5613            cx.add_view(window_id, |cx| {
5614                let fs = project.read(cx).fs().clone();
5615                Workspace::new(
5616                    &WorkspaceParams {
5617                        fs,
5618                        project: project.clone(),
5619                        user_store: self.user_store.clone(),
5620                        languages: self.language_registry.clone(),
5621                        channel_list: cx.add_model(|cx| {
5622                            ChannelList::new(self.user_store.clone(), self.client.clone(), cx)
5623                        }),
5624                        client: self.client.clone(),
5625                    },
5626                    cx,
5627                )
5628            })
5629        }
5630
5631        async fn simulate_host(
5632            mut self,
5633            project: ModelHandle<Project>,
5634            files: Arc<Mutex<Vec<PathBuf>>>,
5635            operations: Rc<Cell<usize>>,
5636            max_operations: usize,
5637            rng: Arc<Mutex<StdRng>>,
5638            mut cx: TestAppContext,
5639        ) -> (Self, TestAppContext) {
5640            async fn simulate_host_internal(
5641                client: &mut TestClient,
5642                project: ModelHandle<Project>,
5643                files: Arc<Mutex<Vec<PathBuf>>>,
5644                operations: Rc<Cell<usize>>,
5645                max_operations: usize,
5646                rng: Arc<Mutex<StdRng>>,
5647                cx: &mut TestAppContext,
5648            ) -> anyhow::Result<()> {
5649                let fs = project.read_with(cx, |project, _| project.fs().clone());
5650                while operations.get() < max_operations {
5651                    operations.set(operations.get() + 1);
5652
5653                    let distribution = rng.lock().gen_range::<usize, _>(0..100);
5654                    match distribution {
5655                        0..=20 if !files.lock().is_empty() => {
5656                            let path = files.lock().choose(&mut *rng.lock()).unwrap().clone();
5657                            let mut path = path.as_path();
5658                            while let Some(parent_path) = path.parent() {
5659                                path = parent_path;
5660                                if rng.lock().gen() {
5661                                    break;
5662                                }
5663                            }
5664
5665                            log::info!("Host: find/create local worktree {:?}", path);
5666                            let find_or_create_worktree = project.update(cx, |project, cx| {
5667                                project.find_or_create_local_worktree(path, true, cx)
5668                            });
5669                            if rng.lock().gen() {
5670                                cx.background().spawn(find_or_create_worktree).detach();
5671                            } else {
5672                                find_or_create_worktree.await?;
5673                            }
5674                        }
5675                        10..=80 if !files.lock().is_empty() => {
5676                            let buffer = if client.buffers.is_empty() || rng.lock().gen() {
5677                                let file = files.lock().choose(&mut *rng.lock()).unwrap().clone();
5678                                let (worktree, path) = project
5679                                    .update(cx, |project, cx| {
5680                                        project.find_or_create_local_worktree(
5681                                            file.clone(),
5682                                            true,
5683                                            cx,
5684                                        )
5685                                    })
5686                                    .await?;
5687                                let project_path =
5688                                    worktree.read_with(cx, |worktree, _| (worktree.id(), path));
5689                                log::info!(
5690                                    "Host: opening path {:?}, worktree {}, relative_path {:?}",
5691                                    file,
5692                                    project_path.0,
5693                                    project_path.1
5694                                );
5695                                let buffer = project
5696                                    .update(cx, |project, cx| project.open_buffer(project_path, cx))
5697                                    .await
5698                                    .unwrap();
5699                                client.buffers.insert(buffer.clone());
5700                                buffer
5701                            } else {
5702                                client
5703                                    .buffers
5704                                    .iter()
5705                                    .choose(&mut *rng.lock())
5706                                    .unwrap()
5707                                    .clone()
5708                            };
5709
5710                            if rng.lock().gen_bool(0.1) {
5711                                cx.update(|cx| {
5712                                    log::info!(
5713                                        "Host: dropping buffer {:?}",
5714                                        buffer.read(cx).file().unwrap().full_path(cx)
5715                                    );
5716                                    client.buffers.remove(&buffer);
5717                                    drop(buffer);
5718                                });
5719                            } else {
5720                                buffer.update(cx, |buffer, cx| {
5721                                    log::info!(
5722                                        "Host: updating buffer {:?} ({})",
5723                                        buffer.file().unwrap().full_path(cx),
5724                                        buffer.remote_id()
5725                                    );
5726                                    buffer.randomly_edit(&mut *rng.lock(), 5, cx)
5727                                });
5728                            }
5729                        }
5730                        _ => loop {
5731                            let path_component_count = rng.lock().gen_range::<usize, _>(1..=5);
5732                            let mut path = PathBuf::new();
5733                            path.push("/");
5734                            for _ in 0..path_component_count {
5735                                let letter = rng.lock().gen_range(b'a'..=b'z');
5736                                path.push(std::str::from_utf8(&[letter]).unwrap());
5737                            }
5738                            path.set_extension("rs");
5739                            let parent_path = path.parent().unwrap();
5740
5741                            log::info!("Host: creating file {:?}", path,);
5742
5743                            if fs.create_dir(&parent_path).await.is_ok()
5744                                && fs.create_file(&path, Default::default()).await.is_ok()
5745                            {
5746                                files.lock().push(path);
5747                                break;
5748                            } else {
5749                                log::info!("Host: cannot create file");
5750                            }
5751                        },
5752                    }
5753
5754                    cx.background().simulate_random_delay().await;
5755                }
5756
5757                Ok(())
5758            }
5759
5760            simulate_host_internal(
5761                &mut self,
5762                project.clone(),
5763                files,
5764                operations,
5765                max_operations,
5766                rng,
5767                &mut cx,
5768            )
5769            .log_err()
5770            .await;
5771            log::info!("Host done");
5772            self.project = Some(project);
5773            (self, cx)
5774        }
5775
5776        pub async fn simulate_guest(
5777            mut self,
5778            guest_id: usize,
5779            project: ModelHandle<Project>,
5780            operations: Rc<Cell<usize>>,
5781            max_operations: usize,
5782            rng: Arc<Mutex<StdRng>>,
5783            host_disconnected: Rc<AtomicBool>,
5784            mut cx: TestAppContext,
5785        ) -> (Self, TestAppContext) {
5786            async fn simulate_guest_internal(
5787                client: &mut TestClient,
5788                guest_id: usize,
5789                project: ModelHandle<Project>,
5790                operations: Rc<Cell<usize>>,
5791                max_operations: usize,
5792                rng: Arc<Mutex<StdRng>>,
5793                cx: &mut TestAppContext,
5794            ) -> anyhow::Result<()> {
5795                while operations.get() < max_operations {
5796                    let buffer = if client.buffers.is_empty() || rng.lock().gen() {
5797                        let worktree = if let Some(worktree) =
5798                            project.read_with(cx, |project, cx| {
5799                                project
5800                                    .worktrees(&cx)
5801                                    .filter(|worktree| {
5802                                        let worktree = worktree.read(cx);
5803                                        worktree.is_visible()
5804                                            && worktree.entries(false).any(|e| e.is_file())
5805                                    })
5806                                    .choose(&mut *rng.lock())
5807                            }) {
5808                            worktree
5809                        } else {
5810                            cx.background().simulate_random_delay().await;
5811                            continue;
5812                        };
5813
5814                        operations.set(operations.get() + 1);
5815                        let (worktree_root_name, project_path) =
5816                            worktree.read_with(cx, |worktree, _| {
5817                                let entry = worktree
5818                                    .entries(false)
5819                                    .filter(|e| e.is_file())
5820                                    .choose(&mut *rng.lock())
5821                                    .unwrap();
5822                                (
5823                                    worktree.root_name().to_string(),
5824                                    (worktree.id(), entry.path.clone()),
5825                                )
5826                            });
5827                        log::info!(
5828                            "Guest {}: opening path {:?} in worktree {} ({})",
5829                            guest_id,
5830                            project_path.1,
5831                            project_path.0,
5832                            worktree_root_name,
5833                        );
5834                        let buffer = project
5835                            .update(cx, |project, cx| {
5836                                project.open_buffer(project_path.clone(), cx)
5837                            })
5838                            .await?;
5839                        log::info!(
5840                            "Guest {}: opened path {:?} in worktree {} ({}) with buffer id {}",
5841                            guest_id,
5842                            project_path.1,
5843                            project_path.0,
5844                            worktree_root_name,
5845                            buffer.read_with(cx, |buffer, _| buffer.remote_id())
5846                        );
5847                        client.buffers.insert(buffer.clone());
5848                        buffer
5849                    } else {
5850                        operations.set(operations.get() + 1);
5851
5852                        client
5853                            .buffers
5854                            .iter()
5855                            .choose(&mut *rng.lock())
5856                            .unwrap()
5857                            .clone()
5858                    };
5859
5860                    let choice = rng.lock().gen_range(0..100);
5861                    match choice {
5862                        0..=9 => {
5863                            cx.update(|cx| {
5864                                log::info!(
5865                                    "Guest {}: dropping buffer {:?}",
5866                                    guest_id,
5867                                    buffer.read(cx).file().unwrap().full_path(cx)
5868                                );
5869                                client.buffers.remove(&buffer);
5870                                drop(buffer);
5871                            });
5872                        }
5873                        10..=19 => {
5874                            let completions = project.update(cx, |project, cx| {
5875                                log::info!(
5876                                    "Guest {}: requesting completions for buffer {} ({:?})",
5877                                    guest_id,
5878                                    buffer.read(cx).remote_id(),
5879                                    buffer.read(cx).file().unwrap().full_path(cx)
5880                                );
5881                                let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
5882                                project.completions(&buffer, offset, cx)
5883                            });
5884                            let completions = cx.background().spawn(async move {
5885                                completions
5886                                    .await
5887                                    .map_err(|err| anyhow!("completions request failed: {:?}", err))
5888                            });
5889                            if rng.lock().gen_bool(0.3) {
5890                                log::info!("Guest {}: detaching completions request", guest_id);
5891                                cx.update(|cx| completions.detach_and_log_err(cx));
5892                            } else {
5893                                completions.await?;
5894                            }
5895                        }
5896                        20..=29 => {
5897                            let code_actions = project.update(cx, |project, cx| {
5898                                log::info!(
5899                                    "Guest {}: requesting code actions for buffer {} ({:?})",
5900                                    guest_id,
5901                                    buffer.read(cx).remote_id(),
5902                                    buffer.read(cx).file().unwrap().full_path(cx)
5903                                );
5904                                let range = buffer.read(cx).random_byte_range(0, &mut *rng.lock());
5905                                project.code_actions(&buffer, range, cx)
5906                            });
5907                            let code_actions = cx.background().spawn(async move {
5908                                code_actions.await.map_err(|err| {
5909                                    anyhow!("code actions request failed: {:?}", err)
5910                                })
5911                            });
5912                            if rng.lock().gen_bool(0.3) {
5913                                log::info!("Guest {}: detaching code actions request", guest_id);
5914                                cx.update(|cx| code_actions.detach_and_log_err(cx));
5915                            } else {
5916                                code_actions.await?;
5917                            }
5918                        }
5919                        30..=39 if buffer.read_with(cx, |buffer, _| buffer.is_dirty()) => {
5920                            let (requested_version, save) = buffer.update(cx, |buffer, cx| {
5921                                log::info!(
5922                                    "Guest {}: saving buffer {} ({:?})",
5923                                    guest_id,
5924                                    buffer.remote_id(),
5925                                    buffer.file().unwrap().full_path(cx)
5926                                );
5927                                (buffer.version(), buffer.save(cx))
5928                            });
5929                            let save = cx.background().spawn(async move {
5930                                let (saved_version, _) = save
5931                                    .await
5932                                    .map_err(|err| anyhow!("save request failed: {:?}", err))?;
5933                                assert!(saved_version.observed_all(&requested_version));
5934                                Ok::<_, anyhow::Error>(())
5935                            });
5936                            if rng.lock().gen_bool(0.3) {
5937                                log::info!("Guest {}: detaching save request", guest_id);
5938                                cx.update(|cx| save.detach_and_log_err(cx));
5939                            } else {
5940                                save.await?;
5941                            }
5942                        }
5943                        40..=44 => {
5944                            let prepare_rename = project.update(cx, |project, cx| {
5945                                log::info!(
5946                                    "Guest {}: preparing rename for buffer {} ({:?})",
5947                                    guest_id,
5948                                    buffer.read(cx).remote_id(),
5949                                    buffer.read(cx).file().unwrap().full_path(cx)
5950                                );
5951                                let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
5952                                project.prepare_rename(buffer, offset, cx)
5953                            });
5954                            let prepare_rename = cx.background().spawn(async move {
5955                                prepare_rename.await.map_err(|err| {
5956                                    anyhow!("prepare rename request failed: {:?}", err)
5957                                })
5958                            });
5959                            if rng.lock().gen_bool(0.3) {
5960                                log::info!("Guest {}: detaching prepare rename request", guest_id);
5961                                cx.update(|cx| prepare_rename.detach_and_log_err(cx));
5962                            } else {
5963                                prepare_rename.await?;
5964                            }
5965                        }
5966                        45..=49 => {
5967                            let definitions = project.update(cx, |project, cx| {
5968                                log::info!(
5969                                    "Guest {}: requesting definitions for buffer {} ({:?})",
5970                                    guest_id,
5971                                    buffer.read(cx).remote_id(),
5972                                    buffer.read(cx).file().unwrap().full_path(cx)
5973                                );
5974                                let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
5975                                project.definition(&buffer, offset, cx)
5976                            });
5977                            let definitions = cx.background().spawn(async move {
5978                                definitions
5979                                    .await
5980                                    .map_err(|err| anyhow!("definitions request failed: {:?}", err))
5981                            });
5982                            if rng.lock().gen_bool(0.3) {
5983                                log::info!("Guest {}: detaching definitions request", guest_id);
5984                                cx.update(|cx| definitions.detach_and_log_err(cx));
5985                            } else {
5986                                client
5987                                    .buffers
5988                                    .extend(definitions.await?.into_iter().map(|loc| loc.buffer));
5989                            }
5990                        }
5991                        50..=54 => {
5992                            let highlights = project.update(cx, |project, cx| {
5993                                log::info!(
5994                                    "Guest {}: requesting highlights for buffer {} ({:?})",
5995                                    guest_id,
5996                                    buffer.read(cx).remote_id(),
5997                                    buffer.read(cx).file().unwrap().full_path(cx)
5998                                );
5999                                let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
6000                                project.document_highlights(&buffer, offset, cx)
6001                            });
6002                            let highlights = cx.background().spawn(async move {
6003                                highlights
6004                                    .await
6005                                    .map_err(|err| anyhow!("highlights request failed: {:?}", err))
6006                            });
6007                            if rng.lock().gen_bool(0.3) {
6008                                log::info!("Guest {}: detaching highlights request", guest_id);
6009                                cx.update(|cx| highlights.detach_and_log_err(cx));
6010                            } else {
6011                                highlights.await?;
6012                            }
6013                        }
6014                        55..=59 => {
6015                            let search = project.update(cx, |project, cx| {
6016                                let query = rng.lock().gen_range('a'..='z');
6017                                log::info!("Guest {}: project-wide search {:?}", guest_id, query);
6018                                project.search(SearchQuery::text(query, false, false), cx)
6019                            });
6020                            let search = cx.background().spawn(async move {
6021                                search
6022                                    .await
6023                                    .map_err(|err| anyhow!("search request failed: {:?}", err))
6024                            });
6025                            if rng.lock().gen_bool(0.3) {
6026                                log::info!("Guest {}: detaching search request", guest_id);
6027                                cx.update(|cx| search.detach_and_log_err(cx));
6028                            } else {
6029                                client.buffers.extend(search.await?.into_keys());
6030                            }
6031                        }
6032                        _ => {
6033                            buffer.update(cx, |buffer, cx| {
6034                                log::info!(
6035                                    "Guest {}: updating buffer {} ({:?})",
6036                                    guest_id,
6037                                    buffer.remote_id(),
6038                                    buffer.file().unwrap().full_path(cx)
6039                                );
6040                                buffer.randomly_edit(&mut *rng.lock(), 5, cx)
6041                            });
6042                        }
6043                    }
6044                    cx.background().simulate_random_delay().await;
6045                }
6046                Ok(())
6047            }
6048
6049            match simulate_guest_internal(
6050                &mut self,
6051                guest_id,
6052                project.clone(),
6053                operations,
6054                max_operations,
6055                rng,
6056                &mut cx,
6057            )
6058            .await
6059            {
6060                Ok(()) => log::info!("guest {} done", guest_id),
6061                Err(err) => {
6062                    if host_disconnected.load(SeqCst) {
6063                        log::error!("guest {} simulation error - {:?}", guest_id, err);
6064                    } else {
6065                        panic!("guest {} simulation error - {:?}", guest_id, err);
6066                    }
6067                }
6068            }
6069
6070            self.project = Some(project);
6071            (self, cx)
6072        }
6073    }
6074
6075    impl Drop for TestClient {
6076        fn drop(&mut self) {
6077            self.client.tear_down();
6078        }
6079    }
6080
6081    impl Executor for Arc<gpui::executor::Background> {
6082        type Timer = gpui::executor::Timer;
6083
6084        fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F) {
6085            self.spawn(future).detach();
6086        }
6087
6088        fn timer(&self, duration: Duration) -> Self::Timer {
6089            self.as_ref().timer(duration)
6090        }
6091    }
6092
6093    fn channel_messages(channel: &Channel) -> Vec<(String, String, bool)> {
6094        channel
6095            .messages()
6096            .cursor::<()>()
6097            .map(|m| {
6098                (
6099                    m.sender.github_login.clone(),
6100                    m.body.clone(),
6101                    m.is_pending(),
6102                )
6103            })
6104            .collect()
6105    }
6106
6107    struct EmptyView;
6108
6109    impl gpui::Entity for EmptyView {
6110        type Event = ();
6111    }
6112
6113    impl gpui::View for EmptyView {
6114        fn ui_name() -> &'static str {
6115            "empty view"
6116        }
6117
6118        fn render(&mut self, _: &mut gpui::RenderContext<Self>) -> gpui::ElementBox {
6119            gpui::Element::boxed(gpui::elements::Empty)
6120        }
6121    }
6122}