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