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