1mod store;
   2
   3use super::{
   4    auth::process_auth_header,
   5    db::{ChannelId, MessageId, UserId},
   6    AppState,
   7};
   8use anyhow::anyhow;
   9use async_std::task;
  10use async_tungstenite::{tungstenite::protocol::Role, WebSocketStream};
  11use collections::{HashMap, HashSet};
  12use futures::{channel::mpsc, future::BoxFuture, FutureExt, SinkExt, StreamExt};
  13use parking_lot::{RwLock, RwLockReadGuard, RwLockWriteGuard};
  14use rpc::{
  15    proto::{self, AnyTypedEnvelope, EnvelopedMessage, RequestMessage},
  16    Connection, ConnectionId, Peer, TypedEnvelope,
  17};
  18use sha1::{Digest as _, Sha1};
  19use std::{any::TypeId, future::Future, path::PathBuf, sync::Arc, time::Instant};
  20use store::{Store, Worktree};
  21use surf::StatusCode;
  22use tide::log;
  23use tide::{
  24    http::headers::{HeaderName, CONNECTION, UPGRADE},
  25    Request, Response,
  26};
  27use time::OffsetDateTime;
  28
  29type MessageHandler = Box<
  30    dyn Send
  31        + Sync
  32        + Fn(Arc<Server>, Box<dyn AnyTypedEnvelope>) -> BoxFuture<'static, tide::Result<()>>,
  33>;
  34
  35pub struct Server {
  36    peer: Arc<Peer>,
  37    store: RwLock<Store>,
  38    app_state: Arc<AppState>,
  39    handlers: HashMap<TypeId, MessageHandler>,
  40    notifications: Option<mpsc::UnboundedSender<()>>,
  41}
  42
  43pub trait Executor {
  44    fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F);
  45}
  46
  47pub struct RealExecutor;
  48
  49const MESSAGE_COUNT_PER_PAGE: usize = 100;
  50const MAX_MESSAGE_LEN: usize = 1024;
  51
  52impl Server {
  53    pub fn new(
  54        app_state: Arc<AppState>,
  55        peer: Arc<Peer>,
  56        notifications: Option<mpsc::UnboundedSender<()>>,
  57    ) -> Arc<Self> {
  58        let mut server = Self {
  59            peer,
  60            app_state,
  61            store: Default::default(),
  62            handlers: Default::default(),
  63            notifications,
  64        };
  65
  66        server
  67            .add_request_handler(Server::ping)
  68            .add_request_handler(Server::register_project)
  69            .add_message_handler(Server::unregister_project)
  70            .add_request_handler(Server::share_project)
  71            .add_message_handler(Server::unshare_project)
  72            .add_request_handler(Server::join_project)
  73            .add_message_handler(Server::leave_project)
  74            .add_request_handler(Server::register_worktree)
  75            .add_message_handler(Server::unregister_worktree)
  76            .add_request_handler(Server::share_worktree)
  77            .add_request_handler(Server::update_worktree)
  78            .add_message_handler(Server::update_diagnostic_summary)
  79            .add_message_handler(Server::disk_based_diagnostics_updating)
  80            .add_message_handler(Server::disk_based_diagnostics_updated)
  81            .add_request_handler(Server::get_definition)
  82            .add_request_handler(Server::get_references)
  83            .add_request_handler(Server::get_document_highlights)
  84            .add_request_handler(Server::get_project_symbols)
  85            .add_request_handler(Server::open_buffer_for_symbol)
  86            .add_request_handler(Server::open_buffer)
  87            .add_message_handler(Server::close_buffer)
  88            .add_request_handler(Server::update_buffer)
  89            .add_message_handler(Server::update_buffer_file)
  90            .add_message_handler(Server::buffer_reloaded)
  91            .add_message_handler(Server::buffer_saved)
  92            .add_request_handler(Server::save_buffer)
  93            .add_request_handler(Server::format_buffers)
  94            .add_request_handler(Server::get_completions)
  95            .add_request_handler(Server::apply_additional_edits_for_completion)
  96            .add_request_handler(Server::get_code_actions)
  97            .add_request_handler(Server::apply_code_action)
  98            .add_request_handler(Server::prepare_rename)
  99            .add_request_handler(Server::perform_rename)
 100            .add_request_handler(Server::get_channels)
 101            .add_request_handler(Server::get_users)
 102            .add_request_handler(Server::join_channel)
 103            .add_message_handler(Server::leave_channel)
 104            .add_request_handler(Server::send_channel_message)
 105            .add_request_handler(Server::get_channel_messages);
 106
 107        Arc::new(server)
 108    }
 109
 110    fn add_message_handler<F, Fut, M>(&mut self, handler: F) -> &mut Self
 111    where
 112        F: 'static + Send + Sync + Fn(Arc<Self>, TypedEnvelope<M>) -> Fut,
 113        Fut: 'static + Send + Future<Output = tide::Result<()>>,
 114        M: EnvelopedMessage,
 115    {
 116        let prev_handler = self.handlers.insert(
 117            TypeId::of::<M>(),
 118            Box::new(move |server, envelope| {
 119                let envelope = envelope.into_any().downcast::<TypedEnvelope<M>>().unwrap();
 120                (handler)(server, *envelope).boxed()
 121            }),
 122        );
 123        if prev_handler.is_some() {
 124            panic!("registered a handler for the same message twice");
 125        }
 126        self
 127    }
 128
 129    fn add_request_handler<F, Fut, M>(&mut self, handler: F) -> &mut Self
 130    where
 131        F: 'static + Send + Sync + Fn(Arc<Self>, TypedEnvelope<M>) -> Fut,
 132        Fut: 'static + Send + Future<Output = tide::Result<M::Response>>,
 133        M: RequestMessage,
 134    {
 135        self.add_message_handler(move |server, envelope| {
 136            let receipt = envelope.receipt();
 137            let response = (handler)(server.clone(), envelope);
 138            async move {
 139                match response.await {
 140                    Ok(response) => {
 141                        server.peer.respond(receipt, response)?;
 142                        Ok(())
 143                    }
 144                    Err(error) => {
 145                        server.peer.respond_with_error(
 146                            receipt,
 147                            proto::Error {
 148                                message: error.to_string(),
 149                            },
 150                        )?;
 151                        Err(error)
 152                    }
 153                }
 154            }
 155        })
 156    }
 157
 158    pub fn handle_connection<E: Executor>(
 159        self: &Arc<Self>,
 160        connection: Connection,
 161        addr: String,
 162        user_id: UserId,
 163        mut send_connection_id: Option<mpsc::Sender<ConnectionId>>,
 164        executor: E,
 165    ) -> impl Future<Output = ()> {
 166        let mut this = self.clone();
 167        async move {
 168            let (connection_id, handle_io, mut incoming_rx) =
 169                this.peer.add_connection(connection).await;
 170
 171            if let Some(send_connection_id) = send_connection_id.as_mut() {
 172                let _ = send_connection_id.send(connection_id).await;
 173            }
 174
 175            this.state_mut().add_connection(connection_id, user_id);
 176            if let Err(err) = this.update_contacts_for_users(&[user_id]) {
 177                log::error!("error updating contacts for {:?}: {}", user_id, err);
 178            }
 179
 180            let handle_io = handle_io.fuse();
 181            futures::pin_mut!(handle_io);
 182            loop {
 183                let next_message = incoming_rx.next().fuse();
 184                futures::pin_mut!(next_message);
 185                futures::select_biased! {
 186                    result = handle_io => {
 187                        if let Err(err) = result {
 188                            log::error!("error handling rpc connection {:?} - {:?}", addr, err);
 189                        }
 190                        break;
 191                    }
 192                    message = next_message => {
 193                        if let Some(message) = message {
 194                            let start_time = Instant::now();
 195                            let type_name = message.payload_type_name();
 196                            log::info!("rpc message received. connection:{}, type:{}", connection_id, type_name);
 197                            if let Some(handler) = this.handlers.get(&message.payload_type_id()) {
 198                                let notifications = this.notifications.clone();
 199                                let is_background = message.is_background();
 200                                let handle_message = (handler)(this.clone(), message);
 201                                let handle_message = async move {
 202                                    if let Err(err) = handle_message.await {
 203                                        log::error!("rpc message error. connection:{}, type:{}, error:{:?}", connection_id, type_name, err);
 204                                    } else {
 205                                        log::info!("rpc message handled. connection:{}, type:{}, duration:{:?}", connection_id, type_name, start_time.elapsed());
 206                                    }
 207                                    if let Some(mut notifications) = notifications {
 208                                        let _ = notifications.send(()).await;
 209                                    }
 210                                };
 211                                if is_background {
 212                                    executor.spawn_detached(handle_message);
 213                                } else {
 214                                    handle_message.await;
 215                                }
 216                            } else {
 217                                log::warn!("unhandled message: {}", type_name);
 218                            }
 219                        } else {
 220                            log::info!("rpc connection closed {:?}", addr);
 221                            break;
 222                        }
 223                    }
 224                }
 225            }
 226
 227            if let Err(err) = this.sign_out(connection_id).await {
 228                log::error!("error signing out connection {:?} - {:?}", addr, err);
 229            }
 230        }
 231    }
 232
 233    async fn sign_out(self: &mut Arc<Self>, connection_id: ConnectionId) -> tide::Result<()> {
 234        self.peer.disconnect(connection_id);
 235        let removed_connection = self.state_mut().remove_connection(connection_id)?;
 236
 237        for (project_id, project) in removed_connection.hosted_projects {
 238            if let Some(share) = project.share {
 239                broadcast(
 240                    connection_id,
 241                    share.guests.keys().copied().collect(),
 242                    |conn_id| {
 243                        self.peer
 244                            .send(conn_id, proto::UnshareProject { project_id })
 245                    },
 246                )?;
 247            }
 248        }
 249
 250        for (project_id, peer_ids) in removed_connection.guest_project_ids {
 251            broadcast(connection_id, peer_ids, |conn_id| {
 252                self.peer.send(
 253                    conn_id,
 254                    proto::RemoveProjectCollaborator {
 255                        project_id,
 256                        peer_id: connection_id.0,
 257                    },
 258                )
 259            })?;
 260        }
 261
 262        self.update_contacts_for_users(removed_connection.contact_ids.iter())?;
 263        Ok(())
 264    }
 265
 266    async fn ping(self: Arc<Server>, _: TypedEnvelope<proto::Ping>) -> tide::Result<proto::Ack> {
 267        Ok(proto::Ack {})
 268    }
 269
 270    async fn register_project(
 271        mut self: Arc<Server>,
 272        request: TypedEnvelope<proto::RegisterProject>,
 273    ) -> tide::Result<proto::RegisterProjectResponse> {
 274        let project_id = {
 275            let mut state = self.state_mut();
 276            let user_id = state.user_id_for_connection(request.sender_id)?;
 277            state.register_project(request.sender_id, user_id)
 278        };
 279        Ok(proto::RegisterProjectResponse { project_id })
 280    }
 281
 282    async fn unregister_project(
 283        mut self: Arc<Server>,
 284        request: TypedEnvelope<proto::UnregisterProject>,
 285    ) -> tide::Result<()> {
 286        let project = self
 287            .state_mut()
 288            .unregister_project(request.payload.project_id, request.sender_id)?;
 289        self.update_contacts_for_users(project.authorized_user_ids().iter())?;
 290        Ok(())
 291    }
 292
 293    async fn share_project(
 294        mut self: Arc<Server>,
 295        request: TypedEnvelope<proto::ShareProject>,
 296    ) -> tide::Result<proto::Ack> {
 297        self.state_mut()
 298            .share_project(request.payload.project_id, request.sender_id);
 299        Ok(proto::Ack {})
 300    }
 301
 302    async fn unshare_project(
 303        mut self: Arc<Server>,
 304        request: TypedEnvelope<proto::UnshareProject>,
 305    ) -> tide::Result<()> {
 306        let project_id = request.payload.project_id;
 307        let project = self
 308            .state_mut()
 309            .unshare_project(project_id, request.sender_id)?;
 310
 311        broadcast(request.sender_id, project.connection_ids, |conn_id| {
 312            self.peer
 313                .send(conn_id, proto::UnshareProject { project_id })
 314        })?;
 315        self.update_contacts_for_users(&project.authorized_user_ids)?;
 316        Ok(())
 317    }
 318
 319    async fn join_project(
 320        mut self: Arc<Server>,
 321        request: TypedEnvelope<proto::JoinProject>,
 322    ) -> tide::Result<proto::JoinProjectResponse> {
 323        let project_id = request.payload.project_id;
 324
 325        let user_id = self.state().user_id_for_connection(request.sender_id)?;
 326        let (response, connection_ids, contact_user_ids) = self
 327            .state_mut()
 328            .join_project(request.sender_id, user_id, project_id)
 329            .and_then(|joined| {
 330                let share = joined.project.share()?;
 331                let peer_count = share.guests.len();
 332                let mut collaborators = Vec::with_capacity(peer_count);
 333                collaborators.push(proto::Collaborator {
 334                    peer_id: joined.project.host_connection_id.0,
 335                    replica_id: 0,
 336                    user_id: joined.project.host_user_id.to_proto(),
 337                });
 338                let worktrees = joined
 339                    .project
 340                    .worktrees
 341                    .iter()
 342                    .filter_map(|(id, worktree)| {
 343                        worktree.share.as_ref().map(|share| proto::Worktree {
 344                            id: *id,
 345                            root_name: worktree.root_name.clone(),
 346                            entries: share.entries.values().cloned().collect(),
 347                            diagnostic_summaries: share
 348                                .diagnostic_summaries
 349                                .values()
 350                                .cloned()
 351                                .collect(),
 352                            weak: worktree.weak,
 353                            next_update_id: share.next_update_id as u64,
 354                        })
 355                    })
 356                    .collect();
 357                for (peer_conn_id, (peer_replica_id, peer_user_id)) in &share.guests {
 358                    if *peer_conn_id != request.sender_id {
 359                        collaborators.push(proto::Collaborator {
 360                            peer_id: peer_conn_id.0,
 361                            replica_id: *peer_replica_id as u32,
 362                            user_id: peer_user_id.to_proto(),
 363                        });
 364                    }
 365                }
 366                let response = proto::JoinProjectResponse {
 367                    worktrees,
 368                    replica_id: joined.replica_id as u32,
 369                    collaborators,
 370                };
 371                let connection_ids = joined.project.connection_ids();
 372                let contact_user_ids = joined.project.authorized_user_ids();
 373                Ok((response, connection_ids, contact_user_ids))
 374            })?;
 375
 376        broadcast(request.sender_id, connection_ids, |conn_id| {
 377            self.peer.send(
 378                conn_id,
 379                proto::AddProjectCollaborator {
 380                    project_id,
 381                    collaborator: Some(proto::Collaborator {
 382                        peer_id: request.sender_id.0,
 383                        replica_id: response.replica_id,
 384                        user_id: user_id.to_proto(),
 385                    }),
 386                },
 387            )
 388        })?;
 389        self.update_contacts_for_users(&contact_user_ids)?;
 390        Ok(response)
 391    }
 392
 393    async fn leave_project(
 394        mut self: Arc<Server>,
 395        request: TypedEnvelope<proto::LeaveProject>,
 396    ) -> tide::Result<()> {
 397        let sender_id = request.sender_id;
 398        let project_id = request.payload.project_id;
 399        let worktree = self.state_mut().leave_project(sender_id, project_id)?;
 400
 401        broadcast(sender_id, worktree.connection_ids, |conn_id| {
 402            self.peer.send(
 403                conn_id,
 404                proto::RemoveProjectCollaborator {
 405                    project_id,
 406                    peer_id: sender_id.0,
 407                },
 408            )
 409        })?;
 410        self.update_contacts_for_users(&worktree.authorized_user_ids)?;
 411
 412        Ok(())
 413    }
 414
 415    async fn register_worktree(
 416        mut self: Arc<Server>,
 417        request: TypedEnvelope<proto::RegisterWorktree>,
 418    ) -> tide::Result<proto::Ack> {
 419        let host_user_id = self.state().user_id_for_connection(request.sender_id)?;
 420
 421        let mut contact_user_ids = HashSet::default();
 422        contact_user_ids.insert(host_user_id);
 423        for github_login in request.payload.authorized_logins {
 424            let contact_user_id = self.app_state.db.create_user(&github_login, false).await?;
 425            contact_user_ids.insert(contact_user_id);
 426        }
 427
 428        let contact_user_ids = contact_user_ids.into_iter().collect::<Vec<_>>();
 429        self.state_mut().register_worktree(
 430            request.payload.project_id,
 431            request.payload.worktree_id,
 432            request.sender_id,
 433            Worktree {
 434                authorized_user_ids: contact_user_ids.clone(),
 435                root_name: request.payload.root_name,
 436                share: None,
 437                weak: false,
 438            },
 439        )?;
 440        self.update_contacts_for_users(&contact_user_ids)?;
 441        Ok(proto::Ack {})
 442    }
 443
 444    async fn unregister_worktree(
 445        mut self: Arc<Server>,
 446        request: TypedEnvelope<proto::UnregisterWorktree>,
 447    ) -> tide::Result<()> {
 448        let project_id = request.payload.project_id;
 449        let worktree_id = request.payload.worktree_id;
 450        let (worktree, guest_connection_ids) =
 451            self.state_mut()
 452                .unregister_worktree(project_id, worktree_id, request.sender_id)?;
 453        broadcast(request.sender_id, guest_connection_ids, |conn_id| {
 454            self.peer.send(
 455                conn_id,
 456                proto::UnregisterWorktree {
 457                    project_id,
 458                    worktree_id,
 459                },
 460            )
 461        })?;
 462        self.update_contacts_for_users(&worktree.authorized_user_ids)?;
 463        Ok(())
 464    }
 465
 466    async fn share_worktree(
 467        mut self: Arc<Server>,
 468        mut request: TypedEnvelope<proto::ShareWorktree>,
 469    ) -> tide::Result<proto::Ack> {
 470        let worktree = request
 471            .payload
 472            .worktree
 473            .as_mut()
 474            .ok_or_else(|| anyhow!("missing worktree"))?;
 475        let entries = worktree
 476            .entries
 477            .iter()
 478            .map(|entry| (entry.id, entry.clone()))
 479            .collect();
 480        let diagnostic_summaries = worktree
 481            .diagnostic_summaries
 482            .iter()
 483            .map(|summary| (PathBuf::from(summary.path.clone()), summary.clone()))
 484            .collect();
 485
 486        let shared_worktree = self.state_mut().share_worktree(
 487            request.payload.project_id,
 488            worktree.id,
 489            request.sender_id,
 490            entries,
 491            diagnostic_summaries,
 492            worktree.next_update_id,
 493        )?;
 494
 495        broadcast(
 496            request.sender_id,
 497            shared_worktree.connection_ids,
 498            |connection_id| {
 499                self.peer
 500                    .forward_send(request.sender_id, connection_id, request.payload.clone())
 501            },
 502        )?;
 503        self.update_contacts_for_users(&shared_worktree.authorized_user_ids)?;
 504
 505        Ok(proto::Ack {})
 506    }
 507
 508    async fn update_worktree(
 509        mut self: Arc<Server>,
 510        request: TypedEnvelope<proto::UpdateWorktree>,
 511    ) -> tide::Result<proto::Ack> {
 512        let connection_ids = self.state_mut().update_worktree(
 513            request.sender_id,
 514            request.payload.project_id,
 515            request.payload.worktree_id,
 516            request.payload.id,
 517            &request.payload.removed_entries,
 518            &request.payload.updated_entries,
 519        )?;
 520
 521        broadcast(request.sender_id, connection_ids, |connection_id| {
 522            self.peer
 523                .forward_send(request.sender_id, connection_id, request.payload.clone())
 524        })?;
 525
 526        Ok(proto::Ack {})
 527    }
 528
 529    async fn update_diagnostic_summary(
 530        mut self: Arc<Server>,
 531        request: TypedEnvelope<proto::UpdateDiagnosticSummary>,
 532    ) -> tide::Result<()> {
 533        let summary = request
 534            .payload
 535            .summary
 536            .clone()
 537            .ok_or_else(|| anyhow!("invalid summary"))?;
 538        let receiver_ids = self.state_mut().update_diagnostic_summary(
 539            request.payload.project_id,
 540            request.payload.worktree_id,
 541            request.sender_id,
 542            summary,
 543        )?;
 544
 545        broadcast(request.sender_id, receiver_ids, |connection_id| {
 546            self.peer
 547                .forward_send(request.sender_id, connection_id, request.payload.clone())
 548        })?;
 549        Ok(())
 550    }
 551
 552    async fn disk_based_diagnostics_updating(
 553        self: Arc<Server>,
 554        request: TypedEnvelope<proto::DiskBasedDiagnosticsUpdating>,
 555    ) -> tide::Result<()> {
 556        let receiver_ids = self
 557            .state()
 558            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 559        broadcast(request.sender_id, receiver_ids, |connection_id| {
 560            self.peer
 561                .forward_send(request.sender_id, connection_id, request.payload.clone())
 562        })?;
 563        Ok(())
 564    }
 565
 566    async fn disk_based_diagnostics_updated(
 567        self: Arc<Server>,
 568        request: TypedEnvelope<proto::DiskBasedDiagnosticsUpdated>,
 569    ) -> tide::Result<()> {
 570        let receiver_ids = self
 571            .state()
 572            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 573        broadcast(request.sender_id, receiver_ids, |connection_id| {
 574            self.peer
 575                .forward_send(request.sender_id, connection_id, request.payload.clone())
 576        })?;
 577        Ok(())
 578    }
 579
 580    async fn get_definition(
 581        self: Arc<Server>,
 582        request: TypedEnvelope<proto::GetDefinition>,
 583    ) -> tide::Result<proto::GetDefinitionResponse> {
 584        let host_connection_id = self
 585            .state()
 586            .read_project(request.payload.project_id, request.sender_id)?
 587            .host_connection_id;
 588        Ok(self
 589            .peer
 590            .forward_request(request.sender_id, host_connection_id, request.payload)
 591            .await?)
 592    }
 593
 594    async fn get_references(
 595        self: Arc<Server>,
 596        request: TypedEnvelope<proto::GetReferences>,
 597    ) -> tide::Result<proto::GetReferencesResponse> {
 598        let host_connection_id = self
 599            .state()
 600            .read_project(request.payload.project_id, request.sender_id)?
 601            .host_connection_id;
 602        Ok(self
 603            .peer
 604            .forward_request(request.sender_id, host_connection_id, request.payload)
 605            .await?)
 606    }
 607
 608    async fn get_document_highlights(
 609        self: Arc<Server>,
 610        request: TypedEnvelope<proto::GetDocumentHighlights>,
 611    ) -> tide::Result<proto::GetDocumentHighlightsResponse> {
 612        let host_connection_id = self
 613            .state()
 614            .read_project(request.payload.project_id, request.sender_id)?
 615            .host_connection_id;
 616        Ok(self
 617            .peer
 618            .forward_request(request.sender_id, host_connection_id, request.payload)
 619            .await?)
 620    }
 621
 622    async fn get_project_symbols(
 623        self: Arc<Server>,
 624        request: TypedEnvelope<proto::GetProjectSymbols>,
 625    ) -> tide::Result<proto::GetProjectSymbolsResponse> {
 626        let host_connection_id = self
 627            .state()
 628            .read_project(request.payload.project_id, request.sender_id)?
 629            .host_connection_id;
 630        Ok(self
 631            .peer
 632            .forward_request(request.sender_id, host_connection_id, request.payload)
 633            .await?)
 634    }
 635
 636    async fn open_buffer_for_symbol(
 637        self: Arc<Server>,
 638        request: TypedEnvelope<proto::OpenBufferForSymbol>,
 639    ) -> tide::Result<proto::OpenBufferForSymbolResponse> {
 640        let host_connection_id = self
 641            .state()
 642            .read_project(request.payload.project_id, request.sender_id)?
 643            .host_connection_id;
 644        Ok(self
 645            .peer
 646            .forward_request(request.sender_id, host_connection_id, request.payload)
 647            .await?)
 648    }
 649
 650    async fn open_buffer(
 651        self: Arc<Server>,
 652        request: TypedEnvelope<proto::OpenBuffer>,
 653    ) -> tide::Result<proto::OpenBufferResponse> {
 654        let host_connection_id = self
 655            .state()
 656            .read_project(request.payload.project_id, request.sender_id)?
 657            .host_connection_id;
 658        Ok(self
 659            .peer
 660            .forward_request(request.sender_id, host_connection_id, request.payload)
 661            .await?)
 662    }
 663
 664    async fn close_buffer(
 665        self: Arc<Server>,
 666        request: TypedEnvelope<proto::CloseBuffer>,
 667    ) -> tide::Result<()> {
 668        let host_connection_id = self
 669            .state()
 670            .read_project(request.payload.project_id, request.sender_id)?
 671            .host_connection_id;
 672        self.peer
 673            .forward_send(request.sender_id, host_connection_id, request.payload)?;
 674        Ok(())
 675    }
 676
 677    async fn save_buffer(
 678        self: Arc<Server>,
 679        request: TypedEnvelope<proto::SaveBuffer>,
 680    ) -> tide::Result<proto::BufferSaved> {
 681        let host;
 682        let mut guests;
 683        {
 684            let state = self.state();
 685            let project = state.read_project(request.payload.project_id, request.sender_id)?;
 686            host = project.host_connection_id;
 687            guests = project.guest_connection_ids()
 688        }
 689
 690        let response = self
 691            .peer
 692            .forward_request(request.sender_id, host, request.payload.clone())
 693            .await?;
 694
 695        guests.retain(|guest_connection_id| *guest_connection_id != request.sender_id);
 696        broadcast(host, guests, |conn_id| {
 697            self.peer.forward_send(host, conn_id, response.clone())
 698        })?;
 699
 700        Ok(response)
 701    }
 702
 703    async fn format_buffers(
 704        self: Arc<Server>,
 705        request: TypedEnvelope<proto::FormatBuffers>,
 706    ) -> tide::Result<proto::FormatBuffersResponse> {
 707        let host = self
 708            .state()
 709            .read_project(request.payload.project_id, request.sender_id)?
 710            .host_connection_id;
 711        Ok(self
 712            .peer
 713            .forward_request(request.sender_id, host, request.payload.clone())
 714            .await?)
 715    }
 716
 717    async fn get_completions(
 718        self: Arc<Server>,
 719        request: TypedEnvelope<proto::GetCompletions>,
 720    ) -> tide::Result<proto::GetCompletionsResponse> {
 721        let host = self
 722            .state()
 723            .read_project(request.payload.project_id, request.sender_id)?
 724            .host_connection_id;
 725        Ok(self
 726            .peer
 727            .forward_request(request.sender_id, host, request.payload.clone())
 728            .await?)
 729    }
 730
 731    async fn apply_additional_edits_for_completion(
 732        self: Arc<Server>,
 733        request: TypedEnvelope<proto::ApplyCompletionAdditionalEdits>,
 734    ) -> tide::Result<proto::ApplyCompletionAdditionalEditsResponse> {
 735        let host = self
 736            .state()
 737            .read_project(request.payload.project_id, request.sender_id)?
 738            .host_connection_id;
 739        Ok(self
 740            .peer
 741            .forward_request(request.sender_id, host, request.payload.clone())
 742            .await?)
 743    }
 744
 745    async fn get_code_actions(
 746        self: Arc<Server>,
 747        request: TypedEnvelope<proto::GetCodeActions>,
 748    ) -> tide::Result<proto::GetCodeActionsResponse> {
 749        let host = self
 750            .state()
 751            .read_project(request.payload.project_id, request.sender_id)?
 752            .host_connection_id;
 753        Ok(self
 754            .peer
 755            .forward_request(request.sender_id, host, request.payload.clone())
 756            .await?)
 757    }
 758
 759    async fn apply_code_action(
 760        self: Arc<Server>,
 761        request: TypedEnvelope<proto::ApplyCodeAction>,
 762    ) -> tide::Result<proto::ApplyCodeActionResponse> {
 763        let host = self
 764            .state()
 765            .read_project(request.payload.project_id, request.sender_id)?
 766            .host_connection_id;
 767        Ok(self
 768            .peer
 769            .forward_request(request.sender_id, host, request.payload.clone())
 770            .await?)
 771    }
 772
 773    async fn prepare_rename(
 774        self: Arc<Server>,
 775        request: TypedEnvelope<proto::PrepareRename>,
 776    ) -> tide::Result<proto::PrepareRenameResponse> {
 777        let host = self
 778            .state()
 779            .read_project(request.payload.project_id, request.sender_id)?
 780            .host_connection_id;
 781        Ok(self
 782            .peer
 783            .forward_request(request.sender_id, host, request.payload.clone())
 784            .await?)
 785    }
 786
 787    async fn perform_rename(
 788        self: Arc<Server>,
 789        request: TypedEnvelope<proto::PerformRename>,
 790    ) -> tide::Result<proto::PerformRenameResponse> {
 791        let host = self
 792            .state()
 793            .read_project(request.payload.project_id, request.sender_id)?
 794            .host_connection_id;
 795        Ok(self
 796            .peer
 797            .forward_request(request.sender_id, host, request.payload.clone())
 798            .await?)
 799    }
 800
 801    async fn update_buffer(
 802        self: Arc<Server>,
 803        request: TypedEnvelope<proto::UpdateBuffer>,
 804    ) -> tide::Result<proto::Ack> {
 805        let receiver_ids = self
 806            .state()
 807            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 808        broadcast(request.sender_id, receiver_ids, |connection_id| {
 809            self.peer
 810                .forward_send(request.sender_id, connection_id, request.payload.clone())
 811        })?;
 812        Ok(proto::Ack {})
 813    }
 814
 815    async fn update_buffer_file(
 816        self: Arc<Server>,
 817        request: TypedEnvelope<proto::UpdateBufferFile>,
 818    ) -> tide::Result<()> {
 819        let receiver_ids = self
 820            .state()
 821            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 822        broadcast(request.sender_id, receiver_ids, |connection_id| {
 823            self.peer
 824                .forward_send(request.sender_id, connection_id, request.payload.clone())
 825        })?;
 826        Ok(())
 827    }
 828
 829    async fn buffer_reloaded(
 830        self: Arc<Server>,
 831        request: TypedEnvelope<proto::BufferReloaded>,
 832    ) -> tide::Result<()> {
 833        let receiver_ids = self
 834            .state()
 835            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 836        broadcast(request.sender_id, receiver_ids, |connection_id| {
 837            self.peer
 838                .forward_send(request.sender_id, connection_id, request.payload.clone())
 839        })?;
 840        Ok(())
 841    }
 842
 843    async fn buffer_saved(
 844        self: Arc<Server>,
 845        request: TypedEnvelope<proto::BufferSaved>,
 846    ) -> tide::Result<()> {
 847        let receiver_ids = self
 848            .state()
 849            .project_connection_ids(request.payload.project_id, request.sender_id)?;
 850        broadcast(request.sender_id, receiver_ids, |connection_id| {
 851            self.peer
 852                .forward_send(request.sender_id, connection_id, request.payload.clone())
 853        })?;
 854        Ok(())
 855    }
 856
 857    async fn get_channels(
 858        self: Arc<Server>,
 859        request: TypedEnvelope<proto::GetChannels>,
 860    ) -> tide::Result<proto::GetChannelsResponse> {
 861        let user_id = self.state().user_id_for_connection(request.sender_id)?;
 862        let channels = self.app_state.db.get_accessible_channels(user_id).await?;
 863        Ok(proto::GetChannelsResponse {
 864            channels: channels
 865                .into_iter()
 866                .map(|chan| proto::Channel {
 867                    id: chan.id.to_proto(),
 868                    name: chan.name,
 869                })
 870                .collect(),
 871        })
 872    }
 873
 874    async fn get_users(
 875        self: Arc<Server>,
 876        request: TypedEnvelope<proto::GetUsers>,
 877    ) -> tide::Result<proto::GetUsersResponse> {
 878        let user_ids = request
 879            .payload
 880            .user_ids
 881            .into_iter()
 882            .map(UserId::from_proto)
 883            .collect();
 884        let users = self
 885            .app_state
 886            .db
 887            .get_users_by_ids(user_ids)
 888            .await?
 889            .into_iter()
 890            .map(|user| proto::User {
 891                id: user.id.to_proto(),
 892                avatar_url: format!("https://github.com/{}.png?size=128", user.github_login),
 893                github_login: user.github_login,
 894            })
 895            .collect();
 896        Ok(proto::GetUsersResponse { users })
 897    }
 898
 899    fn update_contacts_for_users<'a>(
 900        self: &Arc<Server>,
 901        user_ids: impl IntoIterator<Item = &'a UserId>,
 902    ) -> anyhow::Result<()> {
 903        let mut result = Ok(());
 904        let state = self.state();
 905        for user_id in user_ids {
 906            let contacts = state.contacts_for_user(*user_id);
 907            for connection_id in state.connection_ids_for_user(*user_id) {
 908                if let Err(error) = self.peer.send(
 909                    connection_id,
 910                    proto::UpdateContacts {
 911                        contacts: contacts.clone(),
 912                    },
 913                ) {
 914                    result = Err(error);
 915                }
 916            }
 917        }
 918        result
 919    }
 920
 921    async fn join_channel(
 922        mut self: Arc<Self>,
 923        request: TypedEnvelope<proto::JoinChannel>,
 924    ) -> tide::Result<proto::JoinChannelResponse> {
 925        let user_id = self.state().user_id_for_connection(request.sender_id)?;
 926        let channel_id = ChannelId::from_proto(request.payload.channel_id);
 927        if !self
 928            .app_state
 929            .db
 930            .can_user_access_channel(user_id, channel_id)
 931            .await?
 932        {
 933            Err(anyhow!("access denied"))?;
 934        }
 935
 936        self.state_mut().join_channel(request.sender_id, channel_id);
 937        let messages = self
 938            .app_state
 939            .db
 940            .get_channel_messages(channel_id, MESSAGE_COUNT_PER_PAGE, None)
 941            .await?
 942            .into_iter()
 943            .map(|msg| proto::ChannelMessage {
 944                id: msg.id.to_proto(),
 945                body: msg.body,
 946                timestamp: msg.sent_at.unix_timestamp() as u64,
 947                sender_id: msg.sender_id.to_proto(),
 948                nonce: Some(msg.nonce.as_u128().into()),
 949            })
 950            .collect::<Vec<_>>();
 951        Ok(proto::JoinChannelResponse {
 952            done: messages.len() < MESSAGE_COUNT_PER_PAGE,
 953            messages,
 954        })
 955    }
 956
 957    async fn leave_channel(
 958        mut self: Arc<Self>,
 959        request: TypedEnvelope<proto::LeaveChannel>,
 960    ) -> tide::Result<()> {
 961        let user_id = self.state().user_id_for_connection(request.sender_id)?;
 962        let channel_id = ChannelId::from_proto(request.payload.channel_id);
 963        if !self
 964            .app_state
 965            .db
 966            .can_user_access_channel(user_id, channel_id)
 967            .await?
 968        {
 969            Err(anyhow!("access denied"))?;
 970        }
 971
 972        self.state_mut()
 973            .leave_channel(request.sender_id, channel_id);
 974
 975        Ok(())
 976    }
 977
 978    async fn send_channel_message(
 979        self: Arc<Self>,
 980        request: TypedEnvelope<proto::SendChannelMessage>,
 981    ) -> tide::Result<proto::SendChannelMessageResponse> {
 982        let channel_id = ChannelId::from_proto(request.payload.channel_id);
 983        let user_id;
 984        let connection_ids;
 985        {
 986            let state = self.state();
 987            user_id = state.user_id_for_connection(request.sender_id)?;
 988            connection_ids = state.channel_connection_ids(channel_id)?;
 989        }
 990
 991        // Validate the message body.
 992        let body = request.payload.body.trim().to_string();
 993        if body.len() > MAX_MESSAGE_LEN {
 994            return Err(anyhow!("message is too long"))?;
 995        }
 996        if body.is_empty() {
 997            return Err(anyhow!("message can't be blank"))?;
 998        }
 999
1000        let timestamp = OffsetDateTime::now_utc();
1001        let nonce = request
1002            .payload
1003            .nonce
1004            .ok_or_else(|| anyhow!("nonce can't be blank"))?;
1005
1006        let message_id = self
1007            .app_state
1008            .db
1009            .create_channel_message(channel_id, user_id, &body, timestamp, nonce.clone().into())
1010            .await?
1011            .to_proto();
1012        let message = proto::ChannelMessage {
1013            sender_id: user_id.to_proto(),
1014            id: message_id,
1015            body,
1016            timestamp: timestamp.unix_timestamp() as u64,
1017            nonce: Some(nonce),
1018        };
1019        broadcast(request.sender_id, connection_ids, |conn_id| {
1020            self.peer.send(
1021                conn_id,
1022                proto::ChannelMessageSent {
1023                    channel_id: channel_id.to_proto(),
1024                    message: Some(message.clone()),
1025                },
1026            )
1027        })?;
1028        Ok(proto::SendChannelMessageResponse {
1029            message: Some(message),
1030        })
1031    }
1032
1033    async fn get_channel_messages(
1034        self: Arc<Self>,
1035        request: TypedEnvelope<proto::GetChannelMessages>,
1036    ) -> tide::Result<proto::GetChannelMessagesResponse> {
1037        let user_id = self.state().user_id_for_connection(request.sender_id)?;
1038        let channel_id = ChannelId::from_proto(request.payload.channel_id);
1039        if !self
1040            .app_state
1041            .db
1042            .can_user_access_channel(user_id, channel_id)
1043            .await?
1044        {
1045            Err(anyhow!("access denied"))?;
1046        }
1047
1048        let messages = self
1049            .app_state
1050            .db
1051            .get_channel_messages(
1052                channel_id,
1053                MESSAGE_COUNT_PER_PAGE,
1054                Some(MessageId::from_proto(request.payload.before_message_id)),
1055            )
1056            .await?
1057            .into_iter()
1058            .map(|msg| proto::ChannelMessage {
1059                id: msg.id.to_proto(),
1060                body: msg.body,
1061                timestamp: msg.sent_at.unix_timestamp() as u64,
1062                sender_id: msg.sender_id.to_proto(),
1063                nonce: Some(msg.nonce.as_u128().into()),
1064            })
1065            .collect::<Vec<_>>();
1066
1067        Ok(proto::GetChannelMessagesResponse {
1068            done: messages.len() < MESSAGE_COUNT_PER_PAGE,
1069            messages,
1070        })
1071    }
1072
1073    fn state<'a>(self: &'a Arc<Self>) -> RwLockReadGuard<'a, Store> {
1074        self.store.read()
1075    }
1076
1077    fn state_mut<'a>(self: &'a mut Arc<Self>) -> RwLockWriteGuard<'a, Store> {
1078        self.store.write()
1079    }
1080}
1081
1082impl Executor for RealExecutor {
1083    fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F) {
1084        task::spawn(future);
1085    }
1086}
1087
1088fn broadcast<F>(
1089    sender_id: ConnectionId,
1090    receiver_ids: Vec<ConnectionId>,
1091    mut f: F,
1092) -> anyhow::Result<()>
1093where
1094    F: FnMut(ConnectionId) -> anyhow::Result<()>,
1095{
1096    let mut result = Ok(());
1097    for receiver_id in receiver_ids {
1098        if receiver_id != sender_id {
1099            if let Err(error) = f(receiver_id) {
1100                if result.is_ok() {
1101                    result = Err(error);
1102                }
1103            }
1104        }
1105    }
1106    result
1107}
1108
1109pub fn add_routes(app: &mut tide::Server<Arc<AppState>>, rpc: &Arc<Peer>) {
1110    let server = Server::new(app.state().clone(), rpc.clone(), None);
1111    app.at("/rpc").get(move |request: Request<Arc<AppState>>| {
1112        let server = server.clone();
1113        async move {
1114            const WEBSOCKET_GUID: &str = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
1115
1116            let connection_upgrade = header_contains_ignore_case(&request, CONNECTION, "upgrade");
1117            let upgrade_to_websocket = header_contains_ignore_case(&request, UPGRADE, "websocket");
1118            let upgrade_requested = connection_upgrade && upgrade_to_websocket;
1119            let client_protocol_version: Option<u32> = request
1120                .header("X-Zed-Protocol-Version")
1121                .and_then(|v| v.as_str().parse().ok());
1122
1123            if !upgrade_requested || client_protocol_version != Some(rpc::PROTOCOL_VERSION) {
1124                return Ok(Response::new(StatusCode::UpgradeRequired));
1125            }
1126
1127            let header = match request.header("Sec-Websocket-Key") {
1128                Some(h) => h.as_str(),
1129                None => return Err(anyhow!("expected sec-websocket-key"))?,
1130            };
1131
1132            let user_id = process_auth_header(&request).await?;
1133
1134            let mut response = Response::new(StatusCode::SwitchingProtocols);
1135            response.insert_header(UPGRADE, "websocket");
1136            response.insert_header(CONNECTION, "Upgrade");
1137            let hash = Sha1::new().chain(header).chain(WEBSOCKET_GUID).finalize();
1138            response.insert_header("Sec-Websocket-Accept", base64::encode(&hash[..]));
1139            response.insert_header("Sec-Websocket-Version", "13");
1140
1141            let http_res: &mut tide::http::Response = response.as_mut();
1142            let upgrade_receiver = http_res.recv_upgrade().await;
1143            let addr = request.remote().unwrap_or("unknown").to_string();
1144            task::spawn(async move {
1145                if let Some(stream) = upgrade_receiver.await {
1146                    server
1147                        .handle_connection(
1148                            Connection::new(
1149                                WebSocketStream::from_raw_socket(stream, Role::Server, None).await,
1150                            ),
1151                            addr,
1152                            user_id,
1153                            None,
1154                            RealExecutor,
1155                        )
1156                        .await;
1157                }
1158            });
1159
1160            Ok(response)
1161        }
1162    });
1163}
1164
1165fn header_contains_ignore_case<T>(
1166    request: &tide::Request<T>,
1167    header_name: HeaderName,
1168    value: &str,
1169) -> bool {
1170    request
1171        .header(header_name)
1172        .map(|h| {
1173            h.as_str()
1174                .split(',')
1175                .any(|s| s.trim().eq_ignore_ascii_case(value.trim()))
1176        })
1177        .unwrap_or(false)
1178}
1179
1180#[cfg(test)]
1181mod tests {
1182    use super::*;
1183    use crate::{
1184        auth,
1185        db::{tests::TestDb, UserId},
1186        github, AppState, Config,
1187    };
1188    use ::rpc::Peer;
1189    use collections::BTreeMap;
1190    use gpui::{executor, ModelHandle, TestAppContext};
1191    use parking_lot::Mutex;
1192    use postage::{sink::Sink, watch};
1193    use rand::prelude::*;
1194    use rpc::PeerId;
1195    use serde_json::json;
1196    use sqlx::types::time::OffsetDateTime;
1197    use std::{
1198        cell::Cell,
1199        env,
1200        ops::Deref,
1201        path::Path,
1202        rc::Rc,
1203        sync::{
1204            atomic::{AtomicBool, Ordering::SeqCst},
1205            Arc,
1206        },
1207        time::Duration,
1208    };
1209    use zed::{
1210        client::{
1211            self, test::FakeHttpClient, Channel, ChannelDetails, ChannelList, Client, Credentials,
1212            EstablishConnectionError, UserStore,
1213        },
1214        editor::{
1215            self, ConfirmCodeAction, ConfirmCompletion, ConfirmRename, Editor, EditorSettings,
1216            Input, MultiBuffer, Redo, Rename, ToOffset, ToggleCodeActions, Undo,
1217        },
1218        fs::{FakeFs, Fs as _},
1219        language::{
1220            tree_sitter_rust, AnchorRangeExt, Diagnostic, DiagnosticEntry, Language,
1221            LanguageConfig, LanguageRegistry, LanguageServerConfig, Point,
1222        },
1223        lsp,
1224        project::{DiagnosticSummary, Project, ProjectPath},
1225        workspace::{Workspace, WorkspaceParams},
1226    };
1227
1228    #[cfg(test)]
1229    #[ctor::ctor]
1230    fn init_logger() {
1231        if std::env::var("RUST_LOG").is_ok() {
1232            env_logger::init();
1233        }
1234    }
1235
1236    #[gpui::test(iterations = 10)]
1237    async fn test_share_project(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
1238        let (window_b, _) = cx_b.add_window(|_| EmptyView);
1239        let lang_registry = Arc::new(LanguageRegistry::new());
1240        let fs = FakeFs::new(cx_a.background());
1241        cx_a.foreground().forbid_parking();
1242
1243        // Connect to a server as 2 clients.
1244        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1245        let client_a = server.create_client(&mut cx_a, "user_a").await;
1246        let client_b = server.create_client(&mut cx_b, "user_b").await;
1247
1248        // Share a project as client A
1249        fs.insert_tree(
1250            "/a",
1251            json!({
1252                ".zed.toml": r#"collaborators = ["user_b"]"#,
1253                "a.txt": "a-contents",
1254                "b.txt": "b-contents",
1255            }),
1256        )
1257        .await;
1258        let project_a = cx_a.update(|cx| {
1259            Project::local(
1260                client_a.clone(),
1261                client_a.user_store.clone(),
1262                lang_registry.clone(),
1263                fs.clone(),
1264                cx,
1265            )
1266        });
1267        let (worktree_a, _) = project_a
1268            .update(&mut cx_a, |p, cx| {
1269                p.find_or_create_local_worktree("/a", false, cx)
1270            })
1271            .await
1272            .unwrap();
1273        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
1274        worktree_a
1275            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1276            .await;
1277        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
1278        project_a
1279            .update(&mut cx_a, |p, cx| p.share(cx))
1280            .await
1281            .unwrap();
1282
1283        // Join that project as client B
1284        let project_b = Project::remote(
1285            project_id,
1286            client_b.clone(),
1287            client_b.user_store.clone(),
1288            lang_registry.clone(),
1289            fs.clone(),
1290            &mut cx_b.to_async(),
1291        )
1292        .await
1293        .unwrap();
1294
1295        let replica_id_b = project_b.read_with(&cx_b, |project, _| {
1296            assert_eq!(
1297                project
1298                    .collaborators()
1299                    .get(&client_a.peer_id)
1300                    .unwrap()
1301                    .user
1302                    .github_login,
1303                "user_a"
1304            );
1305            project.replica_id()
1306        });
1307        project_a
1308            .condition(&cx_a, |tree, _| {
1309                tree.collaborators()
1310                    .get(&client_b.peer_id)
1311                    .map_or(false, |collaborator| {
1312                        collaborator.replica_id == replica_id_b
1313                            && collaborator.user.github_login == "user_b"
1314                    })
1315            })
1316            .await;
1317
1318        // Open the same file as client B and client A.
1319        let buffer_b = project_b
1320            .update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "b.txt"), cx))
1321            .await
1322            .unwrap();
1323        let buffer_b = cx_b.add_model(|cx| MultiBuffer::singleton(buffer_b, cx));
1324        buffer_b.read_with(&cx_b, |buf, cx| {
1325            assert_eq!(buf.read(cx).text(), "b-contents")
1326        });
1327        project_a.read_with(&cx_a, |project, cx| {
1328            assert!(project.has_open_buffer((worktree_id, "b.txt"), cx))
1329        });
1330        let buffer_a = project_a
1331            .update(&mut cx_a, |p, cx| p.open_buffer((worktree_id, "b.txt"), cx))
1332            .await
1333            .unwrap();
1334
1335        let editor_b = cx_b.add_view(window_b, |cx| {
1336            Editor::for_buffer(buffer_b, Arc::new(|cx| EditorSettings::test(cx)), None, cx)
1337        });
1338
1339        // TODO
1340        // // Create a selection set as client B and see that selection set as client A.
1341        // buffer_a
1342        //     .condition(&cx_a, |buffer, _| buffer.selection_sets().count() == 1)
1343        //     .await;
1344
1345        // Edit the buffer as client B and see that edit as client A.
1346        editor_b.update(&mut cx_b, |editor, cx| {
1347            editor.handle_input(&Input("ok, ".into()), cx)
1348        });
1349        buffer_a
1350            .condition(&cx_a, |buffer, _| buffer.text() == "ok, b-contents")
1351            .await;
1352
1353        // TODO
1354        // // Remove the selection set as client B, see those selections disappear as client A.
1355        cx_b.update(move |_| drop(editor_b));
1356        // buffer_a
1357        //     .condition(&cx_a, |buffer, _| buffer.selection_sets().count() == 0)
1358        //     .await;
1359
1360        // Close the buffer as client A, see that the buffer is closed.
1361        cx_a.update(move |_| drop(buffer_a));
1362        project_a
1363            .condition(&cx_a, |project, cx| {
1364                !project.has_open_buffer((worktree_id, "b.txt"), cx)
1365            })
1366            .await;
1367
1368        // Dropping the client B's project removes client B from client A's collaborators.
1369        cx_b.update(move |_| drop(project_b));
1370        project_a
1371            .condition(&cx_a, |project, _| project.collaborators().is_empty())
1372            .await;
1373    }
1374
1375    #[gpui::test(iterations = 10)]
1376    async fn test_unshare_project(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
1377        let lang_registry = Arc::new(LanguageRegistry::new());
1378        let fs = FakeFs::new(cx_a.background());
1379        cx_a.foreground().forbid_parking();
1380
1381        // Connect to a server as 2 clients.
1382        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1383        let client_a = server.create_client(&mut cx_a, "user_a").await;
1384        let client_b = server.create_client(&mut cx_b, "user_b").await;
1385
1386        // Share a project as client A
1387        fs.insert_tree(
1388            "/a",
1389            json!({
1390                ".zed.toml": r#"collaborators = ["user_b"]"#,
1391                "a.txt": "a-contents",
1392                "b.txt": "b-contents",
1393            }),
1394        )
1395        .await;
1396        let project_a = cx_a.update(|cx| {
1397            Project::local(
1398                client_a.clone(),
1399                client_a.user_store.clone(),
1400                lang_registry.clone(),
1401                fs.clone(),
1402                cx,
1403            )
1404        });
1405        let (worktree_a, _) = project_a
1406            .update(&mut cx_a, |p, cx| {
1407                p.find_or_create_local_worktree("/a", false, cx)
1408            })
1409            .await
1410            .unwrap();
1411        worktree_a
1412            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1413            .await;
1414        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
1415        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
1416        project_a
1417            .update(&mut cx_a, |p, cx| p.share(cx))
1418            .await
1419            .unwrap();
1420        assert!(worktree_a.read_with(&cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1421
1422        // Join that project as client B
1423        let project_b = Project::remote(
1424            project_id,
1425            client_b.clone(),
1426            client_b.user_store.clone(),
1427            lang_registry.clone(),
1428            fs.clone(),
1429            &mut cx_b.to_async(),
1430        )
1431        .await
1432        .unwrap();
1433        project_b
1434            .update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1435            .await
1436            .unwrap();
1437
1438        // Unshare the project as client A
1439        project_a
1440            .update(&mut cx_a, |project, cx| project.unshare(cx))
1441            .await
1442            .unwrap();
1443        project_b
1444            .condition(&mut cx_b, |project, _| project.is_read_only())
1445            .await;
1446        assert!(worktree_a.read_with(&cx_a, |tree, _| !tree.as_local().unwrap().is_shared()));
1447        drop(project_b);
1448
1449        // Share the project again and ensure guests can still join.
1450        project_a
1451            .update(&mut cx_a, |project, cx| project.share(cx))
1452            .await
1453            .unwrap();
1454        assert!(worktree_a.read_with(&cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1455
1456        let project_c = Project::remote(
1457            project_id,
1458            client_b.clone(),
1459            client_b.user_store.clone(),
1460            lang_registry.clone(),
1461            fs.clone(),
1462            &mut cx_b.to_async(),
1463        )
1464        .await
1465        .unwrap();
1466        project_c
1467            .update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1468            .await
1469            .unwrap();
1470    }
1471
1472    #[gpui::test(iterations = 10)]
1473    async fn test_propagate_saves_and_fs_changes(
1474        mut cx_a: TestAppContext,
1475        mut cx_b: TestAppContext,
1476        mut cx_c: TestAppContext,
1477    ) {
1478        let lang_registry = Arc::new(LanguageRegistry::new());
1479        let fs = FakeFs::new(cx_a.background());
1480        cx_a.foreground().forbid_parking();
1481
1482        // Connect to a server as 3 clients.
1483        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1484        let client_a = server.create_client(&mut cx_a, "user_a").await;
1485        let client_b = server.create_client(&mut cx_b, "user_b").await;
1486        let client_c = server.create_client(&mut cx_c, "user_c").await;
1487
1488        // Share a worktree as client A.
1489        fs.insert_tree(
1490            "/a",
1491            json!({
1492                ".zed.toml": r#"collaborators = ["user_b", "user_c"]"#,
1493                "file1": "",
1494                "file2": ""
1495            }),
1496        )
1497        .await;
1498        let project_a = cx_a.update(|cx| {
1499            Project::local(
1500                client_a.clone(),
1501                client_a.user_store.clone(),
1502                lang_registry.clone(),
1503                fs.clone(),
1504                cx,
1505            )
1506        });
1507        let (worktree_a, _) = project_a
1508            .update(&mut cx_a, |p, cx| {
1509                p.find_or_create_local_worktree("/a", false, cx)
1510            })
1511            .await
1512            .unwrap();
1513        worktree_a
1514            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1515            .await;
1516        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
1517        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
1518        project_a
1519            .update(&mut cx_a, |p, cx| p.share(cx))
1520            .await
1521            .unwrap();
1522
1523        // Join that worktree as clients B and C.
1524        let project_b = Project::remote(
1525            project_id,
1526            client_b.clone(),
1527            client_b.user_store.clone(),
1528            lang_registry.clone(),
1529            fs.clone(),
1530            &mut cx_b.to_async(),
1531        )
1532        .await
1533        .unwrap();
1534        let project_c = Project::remote(
1535            project_id,
1536            client_c.clone(),
1537            client_c.user_store.clone(),
1538            lang_registry.clone(),
1539            fs.clone(),
1540            &mut cx_c.to_async(),
1541        )
1542        .await
1543        .unwrap();
1544        let worktree_b = project_b.read_with(&cx_b, |p, cx| p.worktrees(cx).next().unwrap());
1545        let worktree_c = project_c.read_with(&cx_c, |p, cx| p.worktrees(cx).next().unwrap());
1546
1547        // Open and edit a buffer as both guests B and C.
1548        let buffer_b = project_b
1549            .update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
1550            .await
1551            .unwrap();
1552        let buffer_c = project_c
1553            .update(&mut cx_c, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
1554            .await
1555            .unwrap();
1556        buffer_b.update(&mut cx_b, |buf, cx| buf.edit([0..0], "i-am-b, ", cx));
1557        buffer_c.update(&mut cx_c, |buf, cx| buf.edit([0..0], "i-am-c, ", cx));
1558
1559        // Open and edit that buffer as the host.
1560        let buffer_a = project_a
1561            .update(&mut cx_a, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
1562            .await
1563            .unwrap();
1564
1565        buffer_a
1566            .condition(&mut cx_a, |buf, _| buf.text() == "i-am-c, i-am-b, ")
1567            .await;
1568        buffer_a.update(&mut cx_a, |buf, cx| {
1569            buf.edit([buf.len()..buf.len()], "i-am-a", cx)
1570        });
1571
1572        // Wait for edits to propagate
1573        buffer_a
1574            .condition(&mut cx_a, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
1575            .await;
1576        buffer_b
1577            .condition(&mut cx_b, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
1578            .await;
1579        buffer_c
1580            .condition(&mut cx_c, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
1581            .await;
1582
1583        // Edit the buffer as the host and concurrently save as guest B.
1584        let save_b = buffer_b.update(&mut cx_b, |buf, cx| buf.save(cx));
1585        buffer_a.update(&mut cx_a, |buf, cx| buf.edit([0..0], "hi-a, ", cx));
1586        save_b.await.unwrap();
1587        assert_eq!(
1588            fs.load("/a/file1".as_ref()).await.unwrap(),
1589            "hi-a, i-am-c, i-am-b, i-am-a"
1590        );
1591        buffer_a.read_with(&cx_a, |buf, _| assert!(!buf.is_dirty()));
1592        buffer_b.read_with(&cx_b, |buf, _| assert!(!buf.is_dirty()));
1593        buffer_c.condition(&cx_c, |buf, _| !buf.is_dirty()).await;
1594
1595        // Make changes on host's file system, see those changes on guest worktrees.
1596        fs.rename(
1597            "/a/file1".as_ref(),
1598            "/a/file1-renamed".as_ref(),
1599            Default::default(),
1600        )
1601        .await
1602        .unwrap();
1603
1604        fs.rename("/a/file2".as_ref(), "/a/file3".as_ref(), Default::default())
1605            .await
1606            .unwrap();
1607        fs.insert_file(Path::new("/a/file4"), "4".into()).await;
1608
1609        worktree_a
1610            .condition(&cx_a, |tree, _| {
1611                tree.paths()
1612                    .map(|p| p.to_string_lossy())
1613                    .collect::<Vec<_>>()
1614                    == [".zed.toml", "file1-renamed", "file3", "file4"]
1615            })
1616            .await;
1617        worktree_b
1618            .condition(&cx_b, |tree, _| {
1619                tree.paths()
1620                    .map(|p| p.to_string_lossy())
1621                    .collect::<Vec<_>>()
1622                    == [".zed.toml", "file1-renamed", "file3", "file4"]
1623            })
1624            .await;
1625        worktree_c
1626            .condition(&cx_c, |tree, _| {
1627                tree.paths()
1628                    .map(|p| p.to_string_lossy())
1629                    .collect::<Vec<_>>()
1630                    == [".zed.toml", "file1-renamed", "file3", "file4"]
1631            })
1632            .await;
1633
1634        // Ensure buffer files are updated as well.
1635        buffer_a
1636            .condition(&cx_a, |buf, _| {
1637                buf.file().unwrap().path().to_str() == Some("file1-renamed")
1638            })
1639            .await;
1640        buffer_b
1641            .condition(&cx_b, |buf, _| {
1642                buf.file().unwrap().path().to_str() == Some("file1-renamed")
1643            })
1644            .await;
1645        buffer_c
1646            .condition(&cx_c, |buf, _| {
1647                buf.file().unwrap().path().to_str() == Some("file1-renamed")
1648            })
1649            .await;
1650    }
1651
1652    #[gpui::test(iterations = 10)]
1653    async fn test_buffer_conflict_after_save(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
1654        cx_a.foreground().forbid_parking();
1655        let lang_registry = Arc::new(LanguageRegistry::new());
1656        let fs = FakeFs::new(cx_a.background());
1657
1658        // Connect to a server as 2 clients.
1659        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1660        let client_a = server.create_client(&mut cx_a, "user_a").await;
1661        let client_b = server.create_client(&mut cx_b, "user_b").await;
1662
1663        // Share a project as client A
1664        fs.insert_tree(
1665            "/dir",
1666            json!({
1667                ".zed.toml": r#"collaborators = ["user_b", "user_c"]"#,
1668                "a.txt": "a-contents",
1669            }),
1670        )
1671        .await;
1672
1673        let project_a = cx_a.update(|cx| {
1674            Project::local(
1675                client_a.clone(),
1676                client_a.user_store.clone(),
1677                lang_registry.clone(),
1678                fs.clone(),
1679                cx,
1680            )
1681        });
1682        let (worktree_a, _) = project_a
1683            .update(&mut cx_a, |p, cx| {
1684                p.find_or_create_local_worktree("/dir", false, cx)
1685            })
1686            .await
1687            .unwrap();
1688        worktree_a
1689            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1690            .await;
1691        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
1692        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
1693        project_a
1694            .update(&mut cx_a, |p, cx| p.share(cx))
1695            .await
1696            .unwrap();
1697
1698        // Join that project as client B
1699        let project_b = Project::remote(
1700            project_id,
1701            client_b.clone(),
1702            client_b.user_store.clone(),
1703            lang_registry.clone(),
1704            fs.clone(),
1705            &mut cx_b.to_async(),
1706        )
1707        .await
1708        .unwrap();
1709
1710        // Open a buffer as client B
1711        let buffer_b = project_b
1712            .update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1713            .await
1714            .unwrap();
1715
1716        buffer_b.update(&mut cx_b, |buf, cx| buf.edit([0..0], "world ", cx));
1717        buffer_b.read_with(&cx_b, |buf, _| {
1718            assert!(buf.is_dirty());
1719            assert!(!buf.has_conflict());
1720        });
1721
1722        buffer_b
1723            .update(&mut cx_b, |buf, cx| buf.save(cx))
1724            .await
1725            .unwrap();
1726        buffer_b
1727            .condition(&cx_b, |buffer_b, _| !buffer_b.is_dirty())
1728            .await;
1729        buffer_b.read_with(&cx_b, |buf, _| {
1730            assert!(!buf.has_conflict());
1731        });
1732
1733        buffer_b.update(&mut cx_b, |buf, cx| buf.edit([0..0], "hello ", cx));
1734        buffer_b.read_with(&cx_b, |buf, _| {
1735            assert!(buf.is_dirty());
1736            assert!(!buf.has_conflict());
1737        });
1738    }
1739
1740    #[gpui::test(iterations = 10)]
1741    async fn test_buffer_reloading(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
1742        cx_a.foreground().forbid_parking();
1743        let lang_registry = Arc::new(LanguageRegistry::new());
1744        let fs = FakeFs::new(cx_a.background());
1745
1746        // Connect to a server as 2 clients.
1747        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1748        let client_a = server.create_client(&mut cx_a, "user_a").await;
1749        let client_b = server.create_client(&mut cx_b, "user_b").await;
1750
1751        // Share a project as client A
1752        fs.insert_tree(
1753            "/dir",
1754            json!({
1755                ".zed.toml": r#"collaborators = ["user_b", "user_c"]"#,
1756                "a.txt": "a-contents",
1757            }),
1758        )
1759        .await;
1760
1761        let project_a = cx_a.update(|cx| {
1762            Project::local(
1763                client_a.clone(),
1764                client_a.user_store.clone(),
1765                lang_registry.clone(),
1766                fs.clone(),
1767                cx,
1768            )
1769        });
1770        let (worktree_a, _) = project_a
1771            .update(&mut cx_a, |p, cx| {
1772                p.find_or_create_local_worktree("/dir", false, cx)
1773            })
1774            .await
1775            .unwrap();
1776        worktree_a
1777            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1778            .await;
1779        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
1780        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
1781        project_a
1782            .update(&mut cx_a, |p, cx| p.share(cx))
1783            .await
1784            .unwrap();
1785
1786        // Join that project as client B
1787        let project_b = Project::remote(
1788            project_id,
1789            client_b.clone(),
1790            client_b.user_store.clone(),
1791            lang_registry.clone(),
1792            fs.clone(),
1793            &mut cx_b.to_async(),
1794        )
1795        .await
1796        .unwrap();
1797        let _worktree_b = project_b.update(&mut cx_b, |p, cx| p.worktrees(cx).next().unwrap());
1798
1799        // Open a buffer as client B
1800        let buffer_b = project_b
1801            .update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1802            .await
1803            .unwrap();
1804        buffer_b.read_with(&cx_b, |buf, _| {
1805            assert!(!buf.is_dirty());
1806            assert!(!buf.has_conflict());
1807        });
1808
1809        fs.save(Path::new("/dir/a.txt"), &"new contents".into())
1810            .await
1811            .unwrap();
1812        buffer_b
1813            .condition(&cx_b, |buf, _| {
1814                buf.text() == "new contents" && !buf.is_dirty()
1815            })
1816            .await;
1817        buffer_b.read_with(&cx_b, |buf, _| {
1818            assert!(!buf.has_conflict());
1819        });
1820    }
1821
1822    #[gpui::test(iterations = 10)]
1823    async fn test_editing_while_guest_opens_buffer(
1824        mut cx_a: TestAppContext,
1825        mut cx_b: TestAppContext,
1826    ) {
1827        cx_a.foreground().forbid_parking();
1828        let lang_registry = Arc::new(LanguageRegistry::new());
1829        let fs = FakeFs::new(cx_a.background());
1830
1831        // Connect to a server as 2 clients.
1832        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1833        let client_a = server.create_client(&mut cx_a, "user_a").await;
1834        let client_b = server.create_client(&mut cx_b, "user_b").await;
1835
1836        // Share a project as client A
1837        fs.insert_tree(
1838            "/dir",
1839            json!({
1840                ".zed.toml": r#"collaborators = ["user_b"]"#,
1841                "a.txt": "a-contents",
1842            }),
1843        )
1844        .await;
1845        let project_a = cx_a.update(|cx| {
1846            Project::local(
1847                client_a.clone(),
1848                client_a.user_store.clone(),
1849                lang_registry.clone(),
1850                fs.clone(),
1851                cx,
1852            )
1853        });
1854        let (worktree_a, _) = project_a
1855            .update(&mut cx_a, |p, cx| {
1856                p.find_or_create_local_worktree("/dir", false, cx)
1857            })
1858            .await
1859            .unwrap();
1860        worktree_a
1861            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1862            .await;
1863        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
1864        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
1865        project_a
1866            .update(&mut cx_a, |p, cx| p.share(cx))
1867            .await
1868            .unwrap();
1869
1870        // Join that project as client B
1871        let project_b = Project::remote(
1872            project_id,
1873            client_b.clone(),
1874            client_b.user_store.clone(),
1875            lang_registry.clone(),
1876            fs.clone(),
1877            &mut cx_b.to_async(),
1878        )
1879        .await
1880        .unwrap();
1881
1882        // Open a buffer as client A
1883        let buffer_a = project_a
1884            .update(&mut cx_a, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1885            .await
1886            .unwrap();
1887
1888        // Start opening the same buffer as client B
1889        let buffer_b = cx_b
1890            .background()
1891            .spawn(project_b.update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx)));
1892
1893        // Edit the buffer as client A while client B is still opening it.
1894        cx_b.background().simulate_random_delay().await;
1895        buffer_a.update(&mut cx_a, |buf, cx| buf.edit([0..0], "X", cx));
1896        cx_b.background().simulate_random_delay().await;
1897        buffer_a.update(&mut cx_a, |buf, cx| buf.edit([1..1], "Y", cx));
1898
1899        let text = buffer_a.read_with(&cx_a, |buf, _| buf.text());
1900        let buffer_b = buffer_b.await.unwrap();
1901        buffer_b.condition(&cx_b, |buf, _| buf.text() == text).await;
1902    }
1903
1904    #[gpui::test(iterations = 10)]
1905    async fn test_leaving_worktree_while_opening_buffer(
1906        mut cx_a: TestAppContext,
1907        mut cx_b: TestAppContext,
1908    ) {
1909        cx_a.foreground().forbid_parking();
1910        let lang_registry = Arc::new(LanguageRegistry::new());
1911        let fs = FakeFs::new(cx_a.background());
1912
1913        // Connect to a server as 2 clients.
1914        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1915        let client_a = server.create_client(&mut cx_a, "user_a").await;
1916        let client_b = server.create_client(&mut cx_b, "user_b").await;
1917
1918        // Share a project as client A
1919        fs.insert_tree(
1920            "/dir",
1921            json!({
1922                ".zed.toml": r#"collaborators = ["user_b"]"#,
1923                "a.txt": "a-contents",
1924            }),
1925        )
1926        .await;
1927        let project_a = cx_a.update(|cx| {
1928            Project::local(
1929                client_a.clone(),
1930                client_a.user_store.clone(),
1931                lang_registry.clone(),
1932                fs.clone(),
1933                cx,
1934            )
1935        });
1936        let (worktree_a, _) = project_a
1937            .update(&mut cx_a, |p, cx| {
1938                p.find_or_create_local_worktree("/dir", false, cx)
1939            })
1940            .await
1941            .unwrap();
1942        worktree_a
1943            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1944            .await;
1945        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
1946        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
1947        project_a
1948            .update(&mut cx_a, |p, cx| p.share(cx))
1949            .await
1950            .unwrap();
1951
1952        // Join that project as client B
1953        let project_b = Project::remote(
1954            project_id,
1955            client_b.clone(),
1956            client_b.user_store.clone(),
1957            lang_registry.clone(),
1958            fs.clone(),
1959            &mut cx_b.to_async(),
1960        )
1961        .await
1962        .unwrap();
1963
1964        // See that a guest has joined as client A.
1965        project_a
1966            .condition(&cx_a, |p, _| p.collaborators().len() == 1)
1967            .await;
1968
1969        // Begin opening a buffer as client B, but leave the project before the open completes.
1970        let buffer_b = cx_b
1971            .background()
1972            .spawn(project_b.update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx)));
1973        cx_b.update(|_| drop(project_b));
1974        drop(buffer_b);
1975
1976        // See that the guest has left.
1977        project_a
1978            .condition(&cx_a, |p, _| p.collaborators().len() == 0)
1979            .await;
1980    }
1981
1982    #[gpui::test(iterations = 10)]
1983    async fn test_peer_disconnection(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
1984        cx_a.foreground().forbid_parking();
1985        let lang_registry = Arc::new(LanguageRegistry::new());
1986        let fs = FakeFs::new(cx_a.background());
1987
1988        // Connect to a server as 2 clients.
1989        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1990        let client_a = server.create_client(&mut cx_a, "user_a").await;
1991        let client_b = server.create_client(&mut cx_b, "user_b").await;
1992
1993        // Share a project as client A
1994        fs.insert_tree(
1995            "/a",
1996            json!({
1997                ".zed.toml": r#"collaborators = ["user_b"]"#,
1998                "a.txt": "a-contents",
1999                "b.txt": "b-contents",
2000            }),
2001        )
2002        .await;
2003        let project_a = cx_a.update(|cx| {
2004            Project::local(
2005                client_a.clone(),
2006                client_a.user_store.clone(),
2007                lang_registry.clone(),
2008                fs.clone(),
2009                cx,
2010            )
2011        });
2012        let (worktree_a, _) = project_a
2013            .update(&mut cx_a, |p, cx| {
2014                p.find_or_create_local_worktree("/a", false, cx)
2015            })
2016            .await
2017            .unwrap();
2018        worktree_a
2019            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2020            .await;
2021        let project_id = project_a
2022            .update(&mut cx_a, |project, _| project.next_remote_id())
2023            .await;
2024        project_a
2025            .update(&mut cx_a, |project, cx| project.share(cx))
2026            .await
2027            .unwrap();
2028
2029        // Join that project as client B
2030        let _project_b = Project::remote(
2031            project_id,
2032            client_b.clone(),
2033            client_b.user_store.clone(),
2034            lang_registry.clone(),
2035            fs.clone(),
2036            &mut cx_b.to_async(),
2037        )
2038        .await
2039        .unwrap();
2040
2041        // See that a guest has joined as client A.
2042        project_a
2043            .condition(&cx_a, |p, _| p.collaborators().len() == 1)
2044            .await;
2045
2046        // Drop client B's connection and ensure client A observes client B leaving the worktree.
2047        client_b.disconnect(&cx_b.to_async()).unwrap();
2048        project_a
2049            .condition(&cx_a, |p, _| p.collaborators().len() == 0)
2050            .await;
2051    }
2052
2053    #[gpui::test(iterations = 10)]
2054    async fn test_collaborating_with_diagnostics(
2055        mut cx_a: TestAppContext,
2056        mut cx_b: TestAppContext,
2057    ) {
2058        cx_a.foreground().forbid_parking();
2059        let mut lang_registry = Arc::new(LanguageRegistry::new());
2060        let fs = FakeFs::new(cx_a.background());
2061
2062        // Set up a fake language server.
2063        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
2064        Arc::get_mut(&mut lang_registry)
2065            .unwrap()
2066            .add(Arc::new(Language::new(
2067                LanguageConfig {
2068                    name: "Rust".into(),
2069                    path_suffixes: vec!["rs".to_string()],
2070                    language_server: Some(language_server_config),
2071                    ..Default::default()
2072                },
2073                Some(tree_sitter_rust::language()),
2074            )));
2075
2076        // Connect to a server as 2 clients.
2077        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2078        let client_a = server.create_client(&mut cx_a, "user_a").await;
2079        let client_b = server.create_client(&mut cx_b, "user_b").await;
2080
2081        // Share a project as client A
2082        fs.insert_tree(
2083            "/a",
2084            json!({
2085                ".zed.toml": r#"collaborators = ["user_b"]"#,
2086                "a.rs": "let one = two",
2087                "other.rs": "",
2088            }),
2089        )
2090        .await;
2091        let project_a = cx_a.update(|cx| {
2092            Project::local(
2093                client_a.clone(),
2094                client_a.user_store.clone(),
2095                lang_registry.clone(),
2096                fs.clone(),
2097                cx,
2098            )
2099        });
2100        let (worktree_a, _) = project_a
2101            .update(&mut cx_a, |p, cx| {
2102                p.find_or_create_local_worktree("/a", false, cx)
2103            })
2104            .await
2105            .unwrap();
2106        worktree_a
2107            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2108            .await;
2109        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
2110        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
2111        project_a
2112            .update(&mut cx_a, |p, cx| p.share(cx))
2113            .await
2114            .unwrap();
2115
2116        // Cause the language server to start.
2117        let _ = cx_a
2118            .background()
2119            .spawn(project_a.update(&mut cx_a, |project, cx| {
2120                project.open_buffer(
2121                    ProjectPath {
2122                        worktree_id,
2123                        path: Path::new("other.rs").into(),
2124                    },
2125                    cx,
2126                )
2127            }))
2128            .await
2129            .unwrap();
2130
2131        // Simulate a language server reporting errors for a file.
2132        let mut fake_language_server = fake_language_servers.next().await.unwrap();
2133        fake_language_server
2134            .notify::<lsp::notification::PublishDiagnostics>(lsp::PublishDiagnosticsParams {
2135                uri: lsp::Url::from_file_path("/a/a.rs").unwrap(),
2136                version: None,
2137                diagnostics: vec![lsp::Diagnostic {
2138                    severity: Some(lsp::DiagnosticSeverity::ERROR),
2139                    range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 7)),
2140                    message: "message 1".to_string(),
2141                    ..Default::default()
2142                }],
2143            })
2144            .await;
2145
2146        // Wait for server to see the diagnostics update.
2147        server
2148            .condition(|store| {
2149                let worktree = store
2150                    .project(project_id)
2151                    .unwrap()
2152                    .worktrees
2153                    .get(&worktree_id.to_proto())
2154                    .unwrap();
2155
2156                !worktree
2157                    .share
2158                    .as_ref()
2159                    .unwrap()
2160                    .diagnostic_summaries
2161                    .is_empty()
2162            })
2163            .await;
2164
2165        // Join the worktree as client B.
2166        let project_b = Project::remote(
2167            project_id,
2168            client_b.clone(),
2169            client_b.user_store.clone(),
2170            lang_registry.clone(),
2171            fs.clone(),
2172            &mut cx_b.to_async(),
2173        )
2174        .await
2175        .unwrap();
2176
2177        project_b.read_with(&cx_b, |project, cx| {
2178            assert_eq!(
2179                project.diagnostic_summaries(cx).collect::<Vec<_>>(),
2180                &[(
2181                    ProjectPath {
2182                        worktree_id,
2183                        path: Arc::from(Path::new("a.rs")),
2184                    },
2185                    DiagnosticSummary {
2186                        error_count: 1,
2187                        warning_count: 0,
2188                        ..Default::default()
2189                    },
2190                )]
2191            )
2192        });
2193
2194        // Simulate a language server reporting more errors for a file.
2195        fake_language_server
2196            .notify::<lsp::notification::PublishDiagnostics>(lsp::PublishDiagnosticsParams {
2197                uri: lsp::Url::from_file_path("/a/a.rs").unwrap(),
2198                version: None,
2199                diagnostics: vec![
2200                    lsp::Diagnostic {
2201                        severity: Some(lsp::DiagnosticSeverity::ERROR),
2202                        range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 7)),
2203                        message: "message 1".to_string(),
2204                        ..Default::default()
2205                    },
2206                    lsp::Diagnostic {
2207                        severity: Some(lsp::DiagnosticSeverity::WARNING),
2208                        range: lsp::Range::new(
2209                            lsp::Position::new(0, 10),
2210                            lsp::Position::new(0, 13),
2211                        ),
2212                        message: "message 2".to_string(),
2213                        ..Default::default()
2214                    },
2215                ],
2216            })
2217            .await;
2218
2219        // Client b gets the updated summaries
2220        project_b
2221            .condition(&cx_b, |project, cx| {
2222                project.diagnostic_summaries(cx).collect::<Vec<_>>()
2223                    == &[(
2224                        ProjectPath {
2225                            worktree_id,
2226                            path: Arc::from(Path::new("a.rs")),
2227                        },
2228                        DiagnosticSummary {
2229                            error_count: 1,
2230                            warning_count: 1,
2231                            ..Default::default()
2232                        },
2233                    )]
2234            })
2235            .await;
2236
2237        // Open the file with the errors on client B. They should be present.
2238        let buffer_b = cx_b
2239            .background()
2240            .spawn(project_b.update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
2241            .await
2242            .unwrap();
2243
2244        buffer_b.read_with(&cx_b, |buffer, _| {
2245            assert_eq!(
2246                buffer
2247                    .snapshot()
2248                    .diagnostics_in_range::<_, Point>(0..buffer.len())
2249                    .map(|entry| entry)
2250                    .collect::<Vec<_>>(),
2251                &[
2252                    DiagnosticEntry {
2253                        range: Point::new(0, 4)..Point::new(0, 7),
2254                        diagnostic: Diagnostic {
2255                            group_id: 0,
2256                            message: "message 1".to_string(),
2257                            severity: lsp::DiagnosticSeverity::ERROR,
2258                            is_primary: true,
2259                            ..Default::default()
2260                        }
2261                    },
2262                    DiagnosticEntry {
2263                        range: Point::new(0, 10)..Point::new(0, 13),
2264                        diagnostic: Diagnostic {
2265                            group_id: 1,
2266                            severity: lsp::DiagnosticSeverity::WARNING,
2267                            message: "message 2".to_string(),
2268                            is_primary: true,
2269                            ..Default::default()
2270                        }
2271                    }
2272                ]
2273            );
2274        });
2275    }
2276
2277    #[gpui::test(iterations = 10)]
2278    async fn test_collaborating_with_completion(
2279        mut cx_a: TestAppContext,
2280        mut cx_b: TestAppContext,
2281    ) {
2282        cx_a.foreground().forbid_parking();
2283        let mut lang_registry = Arc::new(LanguageRegistry::new());
2284        let fs = FakeFs::new(cx_a.background());
2285
2286        // Set up a fake language server.
2287        let (mut language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
2288        language_server_config.set_fake_capabilities(lsp::ServerCapabilities {
2289            completion_provider: Some(lsp::CompletionOptions {
2290                trigger_characters: Some(vec![".".to_string()]),
2291                ..Default::default()
2292            }),
2293            ..Default::default()
2294        });
2295        Arc::get_mut(&mut lang_registry)
2296            .unwrap()
2297            .add(Arc::new(Language::new(
2298                LanguageConfig {
2299                    name: "Rust".into(),
2300                    path_suffixes: vec!["rs".to_string()],
2301                    language_server: Some(language_server_config),
2302                    ..Default::default()
2303                },
2304                Some(tree_sitter_rust::language()),
2305            )));
2306
2307        // Connect to a server as 2 clients.
2308        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2309        let client_a = server.create_client(&mut cx_a, "user_a").await;
2310        let client_b = server.create_client(&mut cx_b, "user_b").await;
2311
2312        // Share a project as client A
2313        fs.insert_tree(
2314            "/a",
2315            json!({
2316                ".zed.toml": r#"collaborators = ["user_b"]"#,
2317                "main.rs": "fn main() { a }",
2318                "other.rs": "",
2319            }),
2320        )
2321        .await;
2322        let project_a = cx_a.update(|cx| {
2323            Project::local(
2324                client_a.clone(),
2325                client_a.user_store.clone(),
2326                lang_registry.clone(),
2327                fs.clone(),
2328                cx,
2329            )
2330        });
2331        let (worktree_a, _) = project_a
2332            .update(&mut cx_a, |p, cx| {
2333                p.find_or_create_local_worktree("/a", false, cx)
2334            })
2335            .await
2336            .unwrap();
2337        worktree_a
2338            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2339            .await;
2340        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
2341        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
2342        project_a
2343            .update(&mut cx_a, |p, cx| p.share(cx))
2344            .await
2345            .unwrap();
2346
2347        // Join the worktree as client B.
2348        let project_b = Project::remote(
2349            project_id,
2350            client_b.clone(),
2351            client_b.user_store.clone(),
2352            lang_registry.clone(),
2353            fs.clone(),
2354            &mut cx_b.to_async(),
2355        )
2356        .await
2357        .unwrap();
2358
2359        // Open a file in an editor as the guest.
2360        let buffer_b = project_b
2361            .update(&mut cx_b, |p, cx| {
2362                p.open_buffer((worktree_id, "main.rs"), cx)
2363            })
2364            .await
2365            .unwrap();
2366        let (window_b, _) = cx_b.add_window(|_| EmptyView);
2367        let editor_b = cx_b.add_view(window_b, |cx| {
2368            Editor::for_buffer(
2369                cx.add_model(|cx| MultiBuffer::singleton(buffer_b.clone(), cx)),
2370                Arc::new(|cx| EditorSettings::test(cx)),
2371                Some(project_b.clone()),
2372                cx,
2373            )
2374        });
2375
2376        let mut fake_language_server = fake_language_servers.next().await.unwrap();
2377        buffer_b
2378            .condition(&cx_b, |buffer, _| !buffer.completion_triggers().is_empty())
2379            .await;
2380
2381        // Type a completion trigger character as the guest.
2382        editor_b.update(&mut cx_b, |editor, cx| {
2383            editor.select_ranges([13..13], None, cx);
2384            editor.handle_input(&Input(".".into()), cx);
2385            cx.focus(&editor_b);
2386        });
2387
2388        // Receive a completion request as the host's language server.
2389        // Return some completions from the host's language server.
2390        cx_a.foreground().start_waiting();
2391        fake_language_server
2392            .handle_request::<lsp::request::Completion, _>(|params| {
2393                assert_eq!(
2394                    params.text_document_position.text_document.uri,
2395                    lsp::Url::from_file_path("/a/main.rs").unwrap(),
2396                );
2397                assert_eq!(
2398                    params.text_document_position.position,
2399                    lsp::Position::new(0, 14),
2400                );
2401
2402                Some(lsp::CompletionResponse::Array(vec![
2403                    lsp::CompletionItem {
2404                        label: "first_method(…)".into(),
2405                        detail: Some("fn(&mut self, B) -> C".into()),
2406                        text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
2407                            new_text: "first_method($1)".to_string(),
2408                            range: lsp::Range::new(
2409                                lsp::Position::new(0, 14),
2410                                lsp::Position::new(0, 14),
2411                            ),
2412                        })),
2413                        insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
2414                        ..Default::default()
2415                    },
2416                    lsp::CompletionItem {
2417                        label: "second_method(…)".into(),
2418                        detail: Some("fn(&mut self, C) -> D<E>".into()),
2419                        text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
2420                            new_text: "second_method()".to_string(),
2421                            range: lsp::Range::new(
2422                                lsp::Position::new(0, 14),
2423                                lsp::Position::new(0, 14),
2424                            ),
2425                        })),
2426                        insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
2427                        ..Default::default()
2428                    },
2429                ]))
2430            })
2431            .next()
2432            .await
2433            .unwrap();
2434        cx_a.foreground().finish_waiting();
2435
2436        // Open the buffer on the host.
2437        let buffer_a = project_a
2438            .update(&mut cx_a, |p, cx| {
2439                p.open_buffer((worktree_id, "main.rs"), cx)
2440            })
2441            .await
2442            .unwrap();
2443        buffer_a
2444            .condition(&cx_a, |buffer, _| buffer.text() == "fn main() { a. }")
2445            .await;
2446
2447        // Confirm a completion on the guest.
2448        editor_b
2449            .condition(&cx_b, |editor, _| editor.context_menu_visible())
2450            .await;
2451        editor_b.update(&mut cx_b, |editor, cx| {
2452            editor.confirm_completion(&ConfirmCompletion(Some(0)), cx);
2453            assert_eq!(editor.text(cx), "fn main() { a.first_method() }");
2454        });
2455
2456        // Return a resolved completion from the host's language server.
2457        // The resolved completion has an additional text edit.
2458        fake_language_server.handle_request::<lsp::request::ResolveCompletionItem, _>(|params| {
2459            assert_eq!(params.label, "first_method(…)");
2460            lsp::CompletionItem {
2461                label: "first_method(…)".into(),
2462                detail: Some("fn(&mut self, B) -> C".into()),
2463                text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
2464                    new_text: "first_method($1)".to_string(),
2465                    range: lsp::Range::new(lsp::Position::new(0, 14), lsp::Position::new(0, 14)),
2466                })),
2467                additional_text_edits: Some(vec![lsp::TextEdit {
2468                    new_text: "use d::SomeTrait;\n".to_string(),
2469                    range: lsp::Range::new(lsp::Position::new(0, 0), lsp::Position::new(0, 0)),
2470                }]),
2471                insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
2472                ..Default::default()
2473            }
2474        });
2475
2476        // The additional edit is applied.
2477        buffer_a
2478            .condition(&cx_a, |buffer, _| {
2479                buffer.text() == "use d::SomeTrait;\nfn main() { a.first_method() }"
2480            })
2481            .await;
2482        buffer_b
2483            .condition(&cx_b, |buffer, _| {
2484                buffer.text() == "use d::SomeTrait;\nfn main() { a.first_method() }"
2485            })
2486            .await;
2487    }
2488
2489    #[gpui::test(iterations = 10)]
2490    async fn test_formatting_buffer(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
2491        cx_a.foreground().forbid_parking();
2492        let mut lang_registry = Arc::new(LanguageRegistry::new());
2493        let fs = FakeFs::new(cx_a.background());
2494
2495        // Set up a fake language server.
2496        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
2497        Arc::get_mut(&mut lang_registry)
2498            .unwrap()
2499            .add(Arc::new(Language::new(
2500                LanguageConfig {
2501                    name: "Rust".into(),
2502                    path_suffixes: vec!["rs".to_string()],
2503                    language_server: Some(language_server_config),
2504                    ..Default::default()
2505                },
2506                Some(tree_sitter_rust::language()),
2507            )));
2508
2509        // Connect to a server as 2 clients.
2510        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2511        let client_a = server.create_client(&mut cx_a, "user_a").await;
2512        let client_b = server.create_client(&mut cx_b, "user_b").await;
2513
2514        // Share a project as client A
2515        fs.insert_tree(
2516            "/a",
2517            json!({
2518                ".zed.toml": r#"collaborators = ["user_b"]"#,
2519                "a.rs": "let one = two",
2520            }),
2521        )
2522        .await;
2523        let project_a = cx_a.update(|cx| {
2524            Project::local(
2525                client_a.clone(),
2526                client_a.user_store.clone(),
2527                lang_registry.clone(),
2528                fs.clone(),
2529                cx,
2530            )
2531        });
2532        let (worktree_a, _) = project_a
2533            .update(&mut cx_a, |p, cx| {
2534                p.find_or_create_local_worktree("/a", false, cx)
2535            })
2536            .await
2537            .unwrap();
2538        worktree_a
2539            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2540            .await;
2541        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
2542        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
2543        project_a
2544            .update(&mut cx_a, |p, cx| p.share(cx))
2545            .await
2546            .unwrap();
2547
2548        // Join the worktree as client B.
2549        let project_b = Project::remote(
2550            project_id,
2551            client_b.clone(),
2552            client_b.user_store.clone(),
2553            lang_registry.clone(),
2554            fs.clone(),
2555            &mut cx_b.to_async(),
2556        )
2557        .await
2558        .unwrap();
2559
2560        let buffer_b = cx_b
2561            .background()
2562            .spawn(project_b.update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
2563            .await
2564            .unwrap();
2565
2566        let format = project_b.update(&mut cx_b, |project, cx| {
2567            project.format(HashSet::from_iter([buffer_b.clone()]), true, cx)
2568        });
2569
2570        let mut fake_language_server = fake_language_servers.next().await.unwrap();
2571        fake_language_server.handle_request::<lsp::request::Formatting, _>(|_| {
2572            Some(vec![
2573                lsp::TextEdit {
2574                    range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 4)),
2575                    new_text: "h".to_string(),
2576                },
2577                lsp::TextEdit {
2578                    range: lsp::Range::new(lsp::Position::new(0, 7), lsp::Position::new(0, 7)),
2579                    new_text: "y".to_string(),
2580                },
2581            ])
2582        });
2583
2584        format.await.unwrap();
2585        assert_eq!(
2586            buffer_b.read_with(&cx_b, |buffer, _| buffer.text()),
2587            "let honey = two"
2588        );
2589    }
2590
2591    #[gpui::test(iterations = 10)]
2592    async fn test_definition(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
2593        cx_a.foreground().forbid_parking();
2594        let mut lang_registry = Arc::new(LanguageRegistry::new());
2595        let fs = FakeFs::new(cx_a.background());
2596        fs.insert_tree(
2597            "/root-1",
2598            json!({
2599                ".zed.toml": r#"collaborators = ["user_b"]"#,
2600                "a.rs": "const ONE: usize = b::TWO + b::THREE;",
2601            }),
2602        )
2603        .await;
2604        fs.insert_tree(
2605            "/root-2",
2606            json!({
2607                "b.rs": "const TWO: usize = 2;\nconst THREE: usize = 3;",
2608            }),
2609        )
2610        .await;
2611
2612        // Set up a fake language server.
2613        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
2614        Arc::get_mut(&mut lang_registry)
2615            .unwrap()
2616            .add(Arc::new(Language::new(
2617                LanguageConfig {
2618                    name: "Rust".into(),
2619                    path_suffixes: vec!["rs".to_string()],
2620                    language_server: Some(language_server_config),
2621                    ..Default::default()
2622                },
2623                Some(tree_sitter_rust::language()),
2624            )));
2625
2626        // Connect to a server as 2 clients.
2627        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2628        let client_a = server.create_client(&mut cx_a, "user_a").await;
2629        let client_b = server.create_client(&mut cx_b, "user_b").await;
2630
2631        // Share a project as client A
2632        let project_a = cx_a.update(|cx| {
2633            Project::local(
2634                client_a.clone(),
2635                client_a.user_store.clone(),
2636                lang_registry.clone(),
2637                fs.clone(),
2638                cx,
2639            )
2640        });
2641        let (worktree_a, _) = project_a
2642            .update(&mut cx_a, |p, cx| {
2643                p.find_or_create_local_worktree("/root-1", false, cx)
2644            })
2645            .await
2646            .unwrap();
2647        worktree_a
2648            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2649            .await;
2650        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
2651        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
2652        project_a
2653            .update(&mut cx_a, |p, cx| p.share(cx))
2654            .await
2655            .unwrap();
2656
2657        // Join the worktree as client B.
2658        let project_b = Project::remote(
2659            project_id,
2660            client_b.clone(),
2661            client_b.user_store.clone(),
2662            lang_registry.clone(),
2663            fs.clone(),
2664            &mut cx_b.to_async(),
2665        )
2666        .await
2667        .unwrap();
2668
2669        // Open the file on client B.
2670        let buffer_b = cx_b
2671            .background()
2672            .spawn(project_b.update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
2673            .await
2674            .unwrap();
2675
2676        // Request the definition of a symbol as the guest.
2677        let definitions_1 = project_b.update(&mut cx_b, |p, cx| p.definition(&buffer_b, 23, cx));
2678
2679        let mut fake_language_server = fake_language_servers.next().await.unwrap();
2680        fake_language_server.handle_request::<lsp::request::GotoDefinition, _>(|_| {
2681            Some(lsp::GotoDefinitionResponse::Scalar(lsp::Location::new(
2682                lsp::Url::from_file_path("/root-2/b.rs").unwrap(),
2683                lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
2684            )))
2685        });
2686
2687        let definitions_1 = definitions_1.await.unwrap();
2688        cx_b.read(|cx| {
2689            assert_eq!(definitions_1.len(), 1);
2690            assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
2691            let target_buffer = definitions_1[0].buffer.read(cx);
2692            assert_eq!(
2693                target_buffer.text(),
2694                "const TWO: usize = 2;\nconst THREE: usize = 3;"
2695            );
2696            assert_eq!(
2697                definitions_1[0].range.to_point(target_buffer),
2698                Point::new(0, 6)..Point::new(0, 9)
2699            );
2700        });
2701
2702        // Try getting more definitions for the same buffer, ensuring the buffer gets reused from
2703        // the previous call to `definition`.
2704        let definitions_2 = project_b.update(&mut cx_b, |p, cx| p.definition(&buffer_b, 33, cx));
2705        fake_language_server.handle_request::<lsp::request::GotoDefinition, _>(|_| {
2706            Some(lsp::GotoDefinitionResponse::Scalar(lsp::Location::new(
2707                lsp::Url::from_file_path("/root-2/b.rs").unwrap(),
2708                lsp::Range::new(lsp::Position::new(1, 6), lsp::Position::new(1, 11)),
2709            )))
2710        });
2711
2712        let definitions_2 = definitions_2.await.unwrap();
2713        cx_b.read(|cx| {
2714            assert_eq!(definitions_2.len(), 1);
2715            assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
2716            let target_buffer = definitions_2[0].buffer.read(cx);
2717            assert_eq!(
2718                target_buffer.text(),
2719                "const TWO: usize = 2;\nconst THREE: usize = 3;"
2720            );
2721            assert_eq!(
2722                definitions_2[0].range.to_point(target_buffer),
2723                Point::new(1, 6)..Point::new(1, 11)
2724            );
2725        });
2726        assert_eq!(definitions_1[0].buffer, definitions_2[0].buffer);
2727
2728        cx_b.update(|_| {
2729            drop(definitions_1);
2730            drop(definitions_2);
2731        });
2732        project_b
2733            .condition(&cx_b, |proj, cx| proj.worktrees(cx).count() == 1)
2734            .await;
2735    }
2736
2737    #[gpui::test(iterations = 10)]
2738    async fn test_references(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
2739        cx_a.foreground().forbid_parking();
2740        let mut lang_registry = Arc::new(LanguageRegistry::new());
2741        let fs = FakeFs::new(cx_a.background());
2742        fs.insert_tree(
2743            "/root-1",
2744            json!({
2745                ".zed.toml": r#"collaborators = ["user_b"]"#,
2746                "one.rs": "const ONE: usize = 1;",
2747                "two.rs": "const TWO: usize = one::ONE + one::ONE;",
2748            }),
2749        )
2750        .await;
2751        fs.insert_tree(
2752            "/root-2",
2753            json!({
2754                "three.rs": "const THREE: usize = two::TWO + one::ONE;",
2755            }),
2756        )
2757        .await;
2758
2759        // Set up a fake language server.
2760        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
2761        Arc::get_mut(&mut lang_registry)
2762            .unwrap()
2763            .add(Arc::new(Language::new(
2764                LanguageConfig {
2765                    name: "Rust".into(),
2766                    path_suffixes: vec!["rs".to_string()],
2767                    language_server: Some(language_server_config),
2768                    ..Default::default()
2769                },
2770                Some(tree_sitter_rust::language()),
2771            )));
2772
2773        // Connect to a server as 2 clients.
2774        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2775        let client_a = server.create_client(&mut cx_a, "user_a").await;
2776        let client_b = server.create_client(&mut cx_b, "user_b").await;
2777
2778        // Share a project as client A
2779        let project_a = cx_a.update(|cx| {
2780            Project::local(
2781                client_a.clone(),
2782                client_a.user_store.clone(),
2783                lang_registry.clone(),
2784                fs.clone(),
2785                cx,
2786            )
2787        });
2788        let (worktree_a, _) = project_a
2789            .update(&mut cx_a, |p, cx| {
2790                p.find_or_create_local_worktree("/root-1", false, cx)
2791            })
2792            .await
2793            .unwrap();
2794        worktree_a
2795            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2796            .await;
2797        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
2798        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
2799        project_a
2800            .update(&mut cx_a, |p, cx| p.share(cx))
2801            .await
2802            .unwrap();
2803
2804        // Join the worktree as client B.
2805        let project_b = Project::remote(
2806            project_id,
2807            client_b.clone(),
2808            client_b.user_store.clone(),
2809            lang_registry.clone(),
2810            fs.clone(),
2811            &mut cx_b.to_async(),
2812        )
2813        .await
2814        .unwrap();
2815
2816        // Open the file on client B.
2817        let buffer_b = cx_b
2818            .background()
2819            .spawn(project_b.update(&mut cx_b, |p, cx| {
2820                p.open_buffer((worktree_id, "one.rs"), cx)
2821            }))
2822            .await
2823            .unwrap();
2824
2825        // Request references to a symbol as the guest.
2826        let references = project_b.update(&mut cx_b, |p, cx| p.references(&buffer_b, 7, cx));
2827
2828        let mut fake_language_server = fake_language_servers.next().await.unwrap();
2829        fake_language_server.handle_request::<lsp::request::References, _>(|params| {
2830            assert_eq!(
2831                params.text_document_position.text_document.uri.as_str(),
2832                "file:///root-1/one.rs"
2833            );
2834            Some(vec![
2835                lsp::Location {
2836                    uri: lsp::Url::from_file_path("/root-1/two.rs").unwrap(),
2837                    range: lsp::Range::new(lsp::Position::new(0, 24), lsp::Position::new(0, 27)),
2838                },
2839                lsp::Location {
2840                    uri: lsp::Url::from_file_path("/root-1/two.rs").unwrap(),
2841                    range: lsp::Range::new(lsp::Position::new(0, 35), lsp::Position::new(0, 38)),
2842                },
2843                lsp::Location {
2844                    uri: lsp::Url::from_file_path("/root-2/three.rs").unwrap(),
2845                    range: lsp::Range::new(lsp::Position::new(0, 37), lsp::Position::new(0, 40)),
2846                },
2847            ])
2848        });
2849
2850        let references = references.await.unwrap();
2851        cx_b.read(|cx| {
2852            assert_eq!(references.len(), 3);
2853            assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
2854
2855            let two_buffer = references[0].buffer.read(cx);
2856            let three_buffer = references[2].buffer.read(cx);
2857            assert_eq!(
2858                two_buffer.file().unwrap().path().as_ref(),
2859                Path::new("two.rs")
2860            );
2861            assert_eq!(references[1].buffer, references[0].buffer);
2862            assert_eq!(
2863                three_buffer.file().unwrap().full_path(cx),
2864                Path::new("three.rs")
2865            );
2866
2867            assert_eq!(references[0].range.to_offset(&two_buffer), 24..27);
2868            assert_eq!(references[1].range.to_offset(&two_buffer), 35..38);
2869            assert_eq!(references[2].range.to_offset(&three_buffer), 37..40);
2870        });
2871    }
2872
2873    #[gpui::test(iterations = 10)]
2874    async fn test_document_highlights(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
2875        cx_a.foreground().forbid_parking();
2876        let lang_registry = Arc::new(LanguageRegistry::new());
2877        let fs = FakeFs::new(cx_a.background());
2878        fs.insert_tree(
2879            "/root-1",
2880            json!({
2881                ".zed.toml": r#"collaborators = ["user_b"]"#,
2882                "main.rs": "fn double(number: i32) -> i32 { number + number }",
2883            }),
2884        )
2885        .await;
2886
2887        // Set up a fake language server.
2888        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
2889        lang_registry.add(Arc::new(Language::new(
2890            LanguageConfig {
2891                name: "Rust".into(),
2892                path_suffixes: vec!["rs".to_string()],
2893                language_server: Some(language_server_config),
2894                ..Default::default()
2895            },
2896            Some(tree_sitter_rust::language()),
2897        )));
2898
2899        // Connect to a server as 2 clients.
2900        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2901        let client_a = server.create_client(&mut cx_a, "user_a").await;
2902        let client_b = server.create_client(&mut cx_b, "user_b").await;
2903
2904        // Share a project as client A
2905        let project_a = cx_a.update(|cx| {
2906            Project::local(
2907                client_a.clone(),
2908                client_a.user_store.clone(),
2909                lang_registry.clone(),
2910                fs.clone(),
2911                cx,
2912            )
2913        });
2914        let (worktree_a, _) = project_a
2915            .update(&mut cx_a, |p, cx| {
2916                p.find_or_create_local_worktree("/root-1", false, cx)
2917            })
2918            .await
2919            .unwrap();
2920        worktree_a
2921            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2922            .await;
2923        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
2924        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
2925        project_a
2926            .update(&mut cx_a, |p, cx| p.share(cx))
2927            .await
2928            .unwrap();
2929
2930        // Join the worktree as client B.
2931        let project_b = Project::remote(
2932            project_id,
2933            client_b.clone(),
2934            client_b.user_store.clone(),
2935            lang_registry.clone(),
2936            fs.clone(),
2937            &mut cx_b.to_async(),
2938        )
2939        .await
2940        .unwrap();
2941
2942        // Open the file on client B.
2943        let buffer_b = cx_b
2944            .background()
2945            .spawn(project_b.update(&mut cx_b, |p, cx| {
2946                p.open_buffer((worktree_id, "main.rs"), cx)
2947            }))
2948            .await
2949            .unwrap();
2950
2951        // Request document highlights as the guest.
2952        let highlights =
2953            project_b.update(&mut cx_b, |p, cx| p.document_highlights(&buffer_b, 34, cx));
2954
2955        let mut fake_language_server = fake_language_servers.next().await.unwrap();
2956        fake_language_server.handle_request::<lsp::request::DocumentHighlightRequest, _>(
2957            |params| {
2958                assert_eq!(
2959                    params
2960                        .text_document_position_params
2961                        .text_document
2962                        .uri
2963                        .as_str(),
2964                    "file:///root-1/main.rs"
2965                );
2966                assert_eq!(
2967                    params.text_document_position_params.position,
2968                    lsp::Position::new(0, 34)
2969                );
2970                Some(vec![
2971                    lsp::DocumentHighlight {
2972                        kind: Some(lsp::DocumentHighlightKind::WRITE),
2973                        range: lsp::Range::new(
2974                            lsp::Position::new(0, 10),
2975                            lsp::Position::new(0, 16),
2976                        ),
2977                    },
2978                    lsp::DocumentHighlight {
2979                        kind: Some(lsp::DocumentHighlightKind::READ),
2980                        range: lsp::Range::new(
2981                            lsp::Position::new(0, 32),
2982                            lsp::Position::new(0, 38),
2983                        ),
2984                    },
2985                    lsp::DocumentHighlight {
2986                        kind: Some(lsp::DocumentHighlightKind::READ),
2987                        range: lsp::Range::new(
2988                            lsp::Position::new(0, 41),
2989                            lsp::Position::new(0, 47),
2990                        ),
2991                    },
2992                ])
2993            },
2994        );
2995
2996        let highlights = highlights.await.unwrap();
2997        buffer_b.read_with(&cx_b, |buffer, _| {
2998            let snapshot = buffer.snapshot();
2999
3000            let highlights = highlights
3001                .into_iter()
3002                .map(|highlight| (highlight.kind, highlight.range.to_offset(&snapshot)))
3003                .collect::<Vec<_>>();
3004            assert_eq!(
3005                highlights,
3006                &[
3007                    (lsp::DocumentHighlightKind::WRITE, 10..16),
3008                    (lsp::DocumentHighlightKind::READ, 32..38),
3009                    (lsp::DocumentHighlightKind::READ, 41..47)
3010                ]
3011            )
3012        });
3013    }
3014
3015    #[gpui::test(iterations = 10)]
3016    async fn test_project_symbols(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
3017        cx_a.foreground().forbid_parking();
3018        let mut lang_registry = Arc::new(LanguageRegistry::new());
3019        let fs = FakeFs::new(cx_a.background());
3020        fs.insert_tree(
3021            "/code",
3022            json!({
3023                "crate-1": {
3024                    ".zed.toml": r#"collaborators = ["user_b"]"#,
3025                    "one.rs": "const ONE: usize = 1;",
3026                },
3027                "crate-2": {
3028                    "two.rs": "const TWO: usize = 2; const THREE: usize = 3;",
3029                },
3030                "private": {
3031                    "passwords.txt": "the-password",
3032                }
3033            }),
3034        )
3035        .await;
3036
3037        // Set up a fake language server.
3038        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
3039        Arc::get_mut(&mut lang_registry)
3040            .unwrap()
3041            .add(Arc::new(Language::new(
3042                LanguageConfig {
3043                    name: "Rust".into(),
3044                    path_suffixes: vec!["rs".to_string()],
3045                    language_server: Some(language_server_config),
3046                    ..Default::default()
3047                },
3048                Some(tree_sitter_rust::language()),
3049            )));
3050
3051        // Connect to a server as 2 clients.
3052        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3053        let client_a = server.create_client(&mut cx_a, "user_a").await;
3054        let client_b = server.create_client(&mut cx_b, "user_b").await;
3055
3056        // Share a project as client A
3057        let project_a = cx_a.update(|cx| {
3058            Project::local(
3059                client_a.clone(),
3060                client_a.user_store.clone(),
3061                lang_registry.clone(),
3062                fs.clone(),
3063                cx,
3064            )
3065        });
3066        let (worktree_a, _) = project_a
3067            .update(&mut cx_a, |p, cx| {
3068                p.find_or_create_local_worktree("/code/crate-1", false, cx)
3069            })
3070            .await
3071            .unwrap();
3072        worktree_a
3073            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3074            .await;
3075        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
3076        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
3077        project_a
3078            .update(&mut cx_a, |p, cx| p.share(cx))
3079            .await
3080            .unwrap();
3081
3082        // Join the worktree as client B.
3083        let project_b = Project::remote(
3084            project_id,
3085            client_b.clone(),
3086            client_b.user_store.clone(),
3087            lang_registry.clone(),
3088            fs.clone(),
3089            &mut cx_b.to_async(),
3090        )
3091        .await
3092        .unwrap();
3093
3094        // Cause the language server to start.
3095        let _buffer = cx_b
3096            .background()
3097            .spawn(project_b.update(&mut cx_b, |p, cx| {
3098                p.open_buffer((worktree_id, "one.rs"), cx)
3099            }))
3100            .await
3101            .unwrap();
3102
3103        // Request the definition of a symbol as the guest.
3104        let symbols = project_b.update(&mut cx_b, |p, cx| p.symbols("two", cx));
3105        let mut fake_language_server = fake_language_servers.next().await.unwrap();
3106        fake_language_server.handle_request::<lsp::request::WorkspaceSymbol, _>(|_| {
3107            #[allow(deprecated)]
3108            Some(vec![lsp::SymbolInformation {
3109                name: "TWO".into(),
3110                location: lsp::Location {
3111                    uri: lsp::Url::from_file_path("/code/crate-2/two.rs").unwrap(),
3112                    range: lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
3113                },
3114                kind: lsp::SymbolKind::CONSTANT,
3115                tags: None,
3116                container_name: None,
3117                deprecated: None,
3118            }])
3119        });
3120
3121        let symbols = symbols.await.unwrap();
3122        assert_eq!(symbols.len(), 1);
3123        assert_eq!(symbols[0].name, "TWO");
3124
3125        // Open one of the returned symbols.
3126        let buffer_b_2 = project_b
3127            .update(&mut cx_b, |project, cx| {
3128                project.open_buffer_for_symbol(&symbols[0], cx)
3129            })
3130            .await
3131            .unwrap();
3132        buffer_b_2.read_with(&cx_b, |buffer, _| {
3133            assert_eq!(
3134                buffer.file().unwrap().path().as_ref(),
3135                Path::new("../crate-2/two.rs")
3136            );
3137        });
3138
3139        // Attempt to craft a symbol and violate host's privacy by opening an arbitrary file.
3140        let mut fake_symbol = symbols[0].clone();
3141        fake_symbol.path = Path::new("/code/secrets").into();
3142        let error = project_b
3143            .update(&mut cx_b, |project, cx| {
3144                project.open_buffer_for_symbol(&fake_symbol, cx)
3145            })
3146            .await
3147            .unwrap_err();
3148        assert!(error.to_string().contains("invalid symbol signature"));
3149    }
3150
3151    #[gpui::test(iterations = 10)]
3152    async fn test_open_buffer_while_getting_definition_pointing_to_it(
3153        mut cx_a: TestAppContext,
3154        mut cx_b: TestAppContext,
3155        mut rng: StdRng,
3156    ) {
3157        cx_a.foreground().forbid_parking();
3158        let mut lang_registry = Arc::new(LanguageRegistry::new());
3159        let fs = FakeFs::new(cx_a.background());
3160        fs.insert_tree(
3161            "/root",
3162            json!({
3163                ".zed.toml": r#"collaborators = ["user_b"]"#,
3164                "a.rs": "const ONE: usize = b::TWO;",
3165                "b.rs": "const TWO: usize = 2",
3166            }),
3167        )
3168        .await;
3169
3170        // Set up a fake language server.
3171        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
3172
3173        Arc::get_mut(&mut lang_registry)
3174            .unwrap()
3175            .add(Arc::new(Language::new(
3176                LanguageConfig {
3177                    name: "Rust".into(),
3178                    path_suffixes: vec!["rs".to_string()],
3179                    language_server: Some(language_server_config),
3180                    ..Default::default()
3181                },
3182                Some(tree_sitter_rust::language()),
3183            )));
3184
3185        // Connect to a server as 2 clients.
3186        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3187        let client_a = server.create_client(&mut cx_a, "user_a").await;
3188        let client_b = server.create_client(&mut cx_b, "user_b").await;
3189
3190        // Share a project as client A
3191        let project_a = cx_a.update(|cx| {
3192            Project::local(
3193                client_a.clone(),
3194                client_a.user_store.clone(),
3195                lang_registry.clone(),
3196                fs.clone(),
3197                cx,
3198            )
3199        });
3200
3201        let (worktree_a, _) = project_a
3202            .update(&mut cx_a, |p, cx| {
3203                p.find_or_create_local_worktree("/root", false, cx)
3204            })
3205            .await
3206            .unwrap();
3207        worktree_a
3208            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3209            .await;
3210        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
3211        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
3212        project_a
3213            .update(&mut cx_a, |p, cx| p.share(cx))
3214            .await
3215            .unwrap();
3216
3217        // Join the worktree as client B.
3218        let project_b = Project::remote(
3219            project_id,
3220            client_b.clone(),
3221            client_b.user_store.clone(),
3222            lang_registry.clone(),
3223            fs.clone(),
3224            &mut cx_b.to_async(),
3225        )
3226        .await
3227        .unwrap();
3228
3229        let buffer_b1 = cx_b
3230            .background()
3231            .spawn(project_b.update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
3232            .await
3233            .unwrap();
3234
3235        let definitions;
3236        let buffer_b2;
3237        if rng.gen() {
3238            definitions = project_b.update(&mut cx_b, |p, cx| p.definition(&buffer_b1, 23, cx));
3239            buffer_b2 =
3240                project_b.update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "b.rs"), cx));
3241        } else {
3242            buffer_b2 =
3243                project_b.update(&mut cx_b, |p, cx| p.open_buffer((worktree_id, "b.rs"), cx));
3244            definitions = project_b.update(&mut cx_b, |p, cx| p.definition(&buffer_b1, 23, cx));
3245        }
3246
3247        let mut fake_language_server = fake_language_servers.next().await.unwrap();
3248        fake_language_server.handle_request::<lsp::request::GotoDefinition, _>(|_| {
3249            Some(lsp::GotoDefinitionResponse::Scalar(lsp::Location::new(
3250                lsp::Url::from_file_path("/root/b.rs").unwrap(),
3251                lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
3252            )))
3253        });
3254
3255        let buffer_b2 = buffer_b2.await.unwrap();
3256        let definitions = definitions.await.unwrap();
3257        assert_eq!(definitions.len(), 1);
3258        assert_eq!(definitions[0].buffer, buffer_b2);
3259    }
3260
3261    #[gpui::test(iterations = 10)]
3262    async fn test_collaborating_with_code_actions(
3263        mut cx_a: TestAppContext,
3264        mut cx_b: TestAppContext,
3265    ) {
3266        cx_a.foreground().forbid_parking();
3267        let mut lang_registry = Arc::new(LanguageRegistry::new());
3268        let fs = FakeFs::new(cx_a.background());
3269        let mut path_openers_b = Vec::new();
3270        cx_b.update(|cx| editor::init(cx, &mut path_openers_b));
3271
3272        // Set up a fake language server.
3273        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
3274        Arc::get_mut(&mut lang_registry)
3275            .unwrap()
3276            .add(Arc::new(Language::new(
3277                LanguageConfig {
3278                    name: "Rust".into(),
3279                    path_suffixes: vec!["rs".to_string()],
3280                    language_server: Some(language_server_config),
3281                    ..Default::default()
3282                },
3283                Some(tree_sitter_rust::language()),
3284            )));
3285
3286        // Connect to a server as 2 clients.
3287        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3288        let client_a = server.create_client(&mut cx_a, "user_a").await;
3289        let client_b = server.create_client(&mut cx_b, "user_b").await;
3290
3291        // Share a project as client A
3292        fs.insert_tree(
3293            "/a",
3294            json!({
3295                ".zed.toml": r#"collaborators = ["user_b"]"#,
3296                "main.rs": "mod other;\nfn main() { let foo = other::foo(); }",
3297                "other.rs": "pub fn foo() -> usize { 4 }",
3298            }),
3299        )
3300        .await;
3301        let project_a = cx_a.update(|cx| {
3302            Project::local(
3303                client_a.clone(),
3304                client_a.user_store.clone(),
3305                lang_registry.clone(),
3306                fs.clone(),
3307                cx,
3308            )
3309        });
3310        let (worktree_a, _) = project_a
3311            .update(&mut cx_a, |p, cx| {
3312                p.find_or_create_local_worktree("/a", false, cx)
3313            })
3314            .await
3315            .unwrap();
3316        worktree_a
3317            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3318            .await;
3319        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
3320        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
3321        project_a
3322            .update(&mut cx_a, |p, cx| p.share(cx))
3323            .await
3324            .unwrap();
3325
3326        // Join the worktree as client B.
3327        let project_b = Project::remote(
3328            project_id,
3329            client_b.clone(),
3330            client_b.user_store.clone(),
3331            lang_registry.clone(),
3332            fs.clone(),
3333            &mut cx_b.to_async(),
3334        )
3335        .await
3336        .unwrap();
3337        let mut params = cx_b.update(WorkspaceParams::test);
3338        params.languages = lang_registry.clone();
3339        params.client = client_b.client.clone();
3340        params.user_store = client_b.user_store.clone();
3341        params.project = project_b;
3342        params.path_openers = path_openers_b.into();
3343
3344        let (_window_b, workspace_b) = cx_b.add_window(|cx| Workspace::new(&params, cx));
3345        let editor_b = workspace_b
3346            .update(&mut cx_b, |workspace, cx| {
3347                workspace.open_path((worktree_id, "main.rs").into(), cx)
3348            })
3349            .await
3350            .unwrap()
3351            .downcast::<Editor>()
3352            .unwrap();
3353
3354        let mut fake_language_server = fake_language_servers.next().await.unwrap();
3355        fake_language_server
3356            .handle_request::<lsp::request::CodeActionRequest, _>(|params| {
3357                assert_eq!(
3358                    params.text_document.uri,
3359                    lsp::Url::from_file_path("/a/main.rs").unwrap(),
3360                );
3361                assert_eq!(params.range.start, lsp::Position::new(0, 0));
3362                assert_eq!(params.range.end, lsp::Position::new(0, 0));
3363                None
3364            })
3365            .next()
3366            .await;
3367
3368        // Move cursor to a location that contains code actions.
3369        editor_b.update(&mut cx_b, |editor, cx| {
3370            editor.select_ranges([Point::new(1, 31)..Point::new(1, 31)], None, cx);
3371            cx.focus(&editor_b);
3372        });
3373
3374        fake_language_server
3375            .handle_request::<lsp::request::CodeActionRequest, _>(|params| {
3376                assert_eq!(
3377                    params.text_document.uri,
3378                    lsp::Url::from_file_path("/a/main.rs").unwrap(),
3379                );
3380                assert_eq!(params.range.start, lsp::Position::new(1, 31));
3381                assert_eq!(params.range.end, lsp::Position::new(1, 31));
3382
3383                Some(vec![lsp::CodeActionOrCommand::CodeAction(
3384                    lsp::CodeAction {
3385                        title: "Inline into all callers".to_string(),
3386                        edit: Some(lsp::WorkspaceEdit {
3387                            changes: Some(
3388                                [
3389                                    (
3390                                        lsp::Url::from_file_path("/a/main.rs").unwrap(),
3391                                        vec![lsp::TextEdit::new(
3392                                            lsp::Range::new(
3393                                                lsp::Position::new(1, 22),
3394                                                lsp::Position::new(1, 34),
3395                                            ),
3396                                            "4".to_string(),
3397                                        )],
3398                                    ),
3399                                    (
3400                                        lsp::Url::from_file_path("/a/other.rs").unwrap(),
3401                                        vec![lsp::TextEdit::new(
3402                                            lsp::Range::new(
3403                                                lsp::Position::new(0, 0),
3404                                                lsp::Position::new(0, 27),
3405                                            ),
3406                                            "".to_string(),
3407                                        )],
3408                                    ),
3409                                ]
3410                                .into_iter()
3411                                .collect(),
3412                            ),
3413                            ..Default::default()
3414                        }),
3415                        data: Some(json!({
3416                            "codeActionParams": {
3417                                "range": {
3418                                    "start": {"line": 1, "column": 31},
3419                                    "end": {"line": 1, "column": 31},
3420                                }
3421                            }
3422                        })),
3423                        ..Default::default()
3424                    },
3425                )])
3426            })
3427            .next()
3428            .await;
3429
3430        // Toggle code actions and wait for them to display.
3431        editor_b.update(&mut cx_b, |editor, cx| {
3432            editor.toggle_code_actions(&ToggleCodeActions(false), cx);
3433        });
3434        editor_b
3435            .condition(&cx_b, |editor, _| editor.context_menu_visible())
3436            .await;
3437
3438        fake_language_server.remove_request_handler::<lsp::request::CodeActionRequest>();
3439
3440        // Confirming the code action will trigger a resolve request.
3441        let confirm_action = workspace_b
3442            .update(&mut cx_b, |workspace, cx| {
3443                Editor::confirm_code_action(workspace, &ConfirmCodeAction(Some(0)), cx)
3444            })
3445            .unwrap();
3446        fake_language_server.handle_request::<lsp::request::CodeActionResolveRequest, _>(|_| {
3447            lsp::CodeAction {
3448                title: "Inline into all callers".to_string(),
3449                edit: Some(lsp::WorkspaceEdit {
3450                    changes: Some(
3451                        [
3452                            (
3453                                lsp::Url::from_file_path("/a/main.rs").unwrap(),
3454                                vec![lsp::TextEdit::new(
3455                                    lsp::Range::new(
3456                                        lsp::Position::new(1, 22),
3457                                        lsp::Position::new(1, 34),
3458                                    ),
3459                                    "4".to_string(),
3460                                )],
3461                            ),
3462                            (
3463                                lsp::Url::from_file_path("/a/other.rs").unwrap(),
3464                                vec![lsp::TextEdit::new(
3465                                    lsp::Range::new(
3466                                        lsp::Position::new(0, 0),
3467                                        lsp::Position::new(0, 27),
3468                                    ),
3469                                    "".to_string(),
3470                                )],
3471                            ),
3472                        ]
3473                        .into_iter()
3474                        .collect(),
3475                    ),
3476                    ..Default::default()
3477                }),
3478                ..Default::default()
3479            }
3480        });
3481
3482        // After the action is confirmed, an editor containing both modified files is opened.
3483        confirm_action.await.unwrap();
3484        let code_action_editor = workspace_b.read_with(&cx_b, |workspace, cx| {
3485            workspace
3486                .active_item(cx)
3487                .unwrap()
3488                .downcast::<Editor>()
3489                .unwrap()
3490        });
3491        code_action_editor.update(&mut cx_b, |editor, cx| {
3492            assert_eq!(editor.text(cx), "\nmod other;\nfn main() { let foo = 4; }");
3493            editor.undo(&Undo, cx);
3494            assert_eq!(
3495                editor.text(cx),
3496                "pub fn foo() -> usize { 4 }\nmod other;\nfn main() { let foo = other::foo(); }"
3497            );
3498            editor.redo(&Redo, cx);
3499            assert_eq!(editor.text(cx), "\nmod other;\nfn main() { let foo = 4; }");
3500        });
3501    }
3502
3503    #[gpui::test(iterations = 10)]
3504    async fn test_collaborating_with_renames(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
3505        cx_a.foreground().forbid_parking();
3506        let mut lang_registry = Arc::new(LanguageRegistry::new());
3507        let fs = FakeFs::new(cx_a.background());
3508        let mut path_openers_b = Vec::new();
3509        cx_b.update(|cx| editor::init(cx, &mut path_openers_b));
3510
3511        // Set up a fake language server.
3512        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
3513        Arc::get_mut(&mut lang_registry)
3514            .unwrap()
3515            .add(Arc::new(Language::new(
3516                LanguageConfig {
3517                    name: "Rust".into(),
3518                    path_suffixes: vec!["rs".to_string()],
3519                    language_server: Some(language_server_config),
3520                    ..Default::default()
3521                },
3522                Some(tree_sitter_rust::language()),
3523            )));
3524
3525        // Connect to a server as 2 clients.
3526        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3527        let client_a = server.create_client(&mut cx_a, "user_a").await;
3528        let client_b = server.create_client(&mut cx_b, "user_b").await;
3529
3530        // Share a project as client A
3531        fs.insert_tree(
3532            "/dir",
3533            json!({
3534                ".zed.toml": r#"collaborators = ["user_b"]"#,
3535                "one.rs": "const ONE: usize = 1;",
3536                "two.rs": "const TWO: usize = one::ONE + one::ONE;"
3537            }),
3538        )
3539        .await;
3540        let project_a = cx_a.update(|cx| {
3541            Project::local(
3542                client_a.clone(),
3543                client_a.user_store.clone(),
3544                lang_registry.clone(),
3545                fs.clone(),
3546                cx,
3547            )
3548        });
3549        let (worktree_a, _) = project_a
3550            .update(&mut cx_a, |p, cx| {
3551                p.find_or_create_local_worktree("/dir", false, cx)
3552            })
3553            .await
3554            .unwrap();
3555        worktree_a
3556            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3557            .await;
3558        let project_id = project_a.update(&mut cx_a, |p, _| p.next_remote_id()).await;
3559        let worktree_id = worktree_a.read_with(&cx_a, |tree, _| tree.id());
3560        project_a
3561            .update(&mut cx_a, |p, cx| p.share(cx))
3562            .await
3563            .unwrap();
3564
3565        // Join the worktree as client B.
3566        let project_b = Project::remote(
3567            project_id,
3568            client_b.clone(),
3569            client_b.user_store.clone(),
3570            lang_registry.clone(),
3571            fs.clone(),
3572            &mut cx_b.to_async(),
3573        )
3574        .await
3575        .unwrap();
3576        let mut params = cx_b.update(WorkspaceParams::test);
3577        params.languages = lang_registry.clone();
3578        params.client = client_b.client.clone();
3579        params.user_store = client_b.user_store.clone();
3580        params.project = project_b;
3581        params.path_openers = path_openers_b.into();
3582
3583        let (_window_b, workspace_b) = cx_b.add_window(|cx| Workspace::new(&params, cx));
3584        let editor_b = workspace_b
3585            .update(&mut cx_b, |workspace, cx| {
3586                workspace.open_path((worktree_id, "one.rs").into(), cx)
3587            })
3588            .await
3589            .unwrap()
3590            .downcast::<Editor>()
3591            .unwrap();
3592        let mut fake_language_server = fake_language_servers.next().await.unwrap();
3593
3594        // Move cursor to a location that can be renamed.
3595        let prepare_rename = editor_b.update(&mut cx_b, |editor, cx| {
3596            editor.select_ranges([7..7], None, cx);
3597            editor.rename(&Rename, cx).unwrap()
3598        });
3599
3600        fake_language_server
3601            .handle_request::<lsp::request::PrepareRenameRequest, _>(|params| {
3602                assert_eq!(params.text_document.uri.as_str(), "file:///dir/one.rs");
3603                assert_eq!(params.position, lsp::Position::new(0, 7));
3604                Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
3605                    lsp::Position::new(0, 6),
3606                    lsp::Position::new(0, 9),
3607                )))
3608            })
3609            .next()
3610            .await
3611            .unwrap();
3612        prepare_rename.await.unwrap();
3613        editor_b.update(&mut cx_b, |editor, cx| {
3614            let rename = editor.pending_rename().unwrap();
3615            let buffer = editor.buffer().read(cx).snapshot(cx);
3616            assert_eq!(
3617                rename.range.start.to_offset(&buffer)..rename.range.end.to_offset(&buffer),
3618                6..9
3619            );
3620            rename.editor.update(cx, |rename_editor, cx| {
3621                rename_editor.buffer().update(cx, |rename_buffer, cx| {
3622                    rename_buffer.edit([0..3], "THREE", cx);
3623                });
3624            });
3625        });
3626
3627        let confirm_rename = workspace_b.update(&mut cx_b, |workspace, cx| {
3628            Editor::confirm_rename(workspace, &ConfirmRename, cx).unwrap()
3629        });
3630        fake_language_server
3631            .handle_request::<lsp::request::Rename, _>(|params| {
3632                assert_eq!(
3633                    params.text_document_position.text_document.uri.as_str(),
3634                    "file:///dir/one.rs"
3635                );
3636                assert_eq!(
3637                    params.text_document_position.position,
3638                    lsp::Position::new(0, 6)
3639                );
3640                assert_eq!(params.new_name, "THREE");
3641                Some(lsp::WorkspaceEdit {
3642                    changes: Some(
3643                        [
3644                            (
3645                                lsp::Url::from_file_path("/dir/one.rs").unwrap(),
3646                                vec![lsp::TextEdit::new(
3647                                    lsp::Range::new(
3648                                        lsp::Position::new(0, 6),
3649                                        lsp::Position::new(0, 9),
3650                                    ),
3651                                    "THREE".to_string(),
3652                                )],
3653                            ),
3654                            (
3655                                lsp::Url::from_file_path("/dir/two.rs").unwrap(),
3656                                vec![
3657                                    lsp::TextEdit::new(
3658                                        lsp::Range::new(
3659                                            lsp::Position::new(0, 24),
3660                                            lsp::Position::new(0, 27),
3661                                        ),
3662                                        "THREE".to_string(),
3663                                    ),
3664                                    lsp::TextEdit::new(
3665                                        lsp::Range::new(
3666                                            lsp::Position::new(0, 35),
3667                                            lsp::Position::new(0, 38),
3668                                        ),
3669                                        "THREE".to_string(),
3670                                    ),
3671                                ],
3672                            ),
3673                        ]
3674                        .into_iter()
3675                        .collect(),
3676                    ),
3677                    ..Default::default()
3678                })
3679            })
3680            .next()
3681            .await
3682            .unwrap();
3683        confirm_rename.await.unwrap();
3684
3685        let rename_editor = workspace_b.read_with(&cx_b, |workspace, cx| {
3686            workspace
3687                .active_item(cx)
3688                .unwrap()
3689                .downcast::<Editor>()
3690                .unwrap()
3691        });
3692        rename_editor.update(&mut cx_b, |editor, cx| {
3693            assert_eq!(
3694                editor.text(cx),
3695                "const TWO: usize = one::THREE + one::THREE;\nconst THREE: usize = 1;"
3696            );
3697            editor.undo(&Undo, cx);
3698            assert_eq!(
3699                editor.text(cx),
3700                "const TWO: usize = one::ONE + one::ONE;\nconst ONE: usize = 1;"
3701            );
3702            editor.redo(&Redo, cx);
3703            assert_eq!(
3704                editor.text(cx),
3705                "const TWO: usize = one::THREE + one::THREE;\nconst THREE: usize = 1;"
3706            );
3707        });
3708
3709        // Ensure temporary rename edits cannot be undone/redone.
3710        editor_b.update(&mut cx_b, |editor, cx| {
3711            editor.undo(&Undo, cx);
3712            assert_eq!(editor.text(cx), "const ONE: usize = 1;");
3713            editor.undo(&Undo, cx);
3714            assert_eq!(editor.text(cx), "const ONE: usize = 1;");
3715            editor.redo(&Redo, cx);
3716            assert_eq!(editor.text(cx), "const THREE: usize = 1;");
3717        })
3718    }
3719
3720    #[gpui::test(iterations = 10)]
3721    async fn test_basic_chat(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
3722        cx_a.foreground().forbid_parking();
3723
3724        // Connect to a server as 2 clients.
3725        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3726        let client_a = server.create_client(&mut cx_a, "user_a").await;
3727        let client_b = server.create_client(&mut cx_b, "user_b").await;
3728
3729        // Create an org that includes these 2 users.
3730        let db = &server.app_state.db;
3731        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
3732        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
3733            .await
3734            .unwrap();
3735        db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
3736            .await
3737            .unwrap();
3738
3739        // Create a channel that includes all the users.
3740        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
3741        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
3742            .await
3743            .unwrap();
3744        db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
3745            .await
3746            .unwrap();
3747        db.create_channel_message(
3748            channel_id,
3749            client_b.current_user_id(&cx_b),
3750            "hello A, it's B.",
3751            OffsetDateTime::now_utc(),
3752            1,
3753        )
3754        .await
3755        .unwrap();
3756
3757        let channels_a = cx_a
3758            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
3759        channels_a
3760            .condition(&mut cx_a, |list, _| list.available_channels().is_some())
3761            .await;
3762        channels_a.read_with(&cx_a, |list, _| {
3763            assert_eq!(
3764                list.available_channels().unwrap(),
3765                &[ChannelDetails {
3766                    id: channel_id.to_proto(),
3767                    name: "test-channel".to_string()
3768                }]
3769            )
3770        });
3771        let channel_a = channels_a.update(&mut cx_a, |this, cx| {
3772            this.get_channel(channel_id.to_proto(), cx).unwrap()
3773        });
3774        channel_a.read_with(&cx_a, |channel, _| assert!(channel.messages().is_empty()));
3775        channel_a
3776            .condition(&cx_a, |channel, _| {
3777                channel_messages(channel)
3778                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
3779            })
3780            .await;
3781
3782        let channels_b = cx_b
3783            .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
3784        channels_b
3785            .condition(&mut cx_b, |list, _| list.available_channels().is_some())
3786            .await;
3787        channels_b.read_with(&cx_b, |list, _| {
3788            assert_eq!(
3789                list.available_channels().unwrap(),
3790                &[ChannelDetails {
3791                    id: channel_id.to_proto(),
3792                    name: "test-channel".to_string()
3793                }]
3794            )
3795        });
3796
3797        let channel_b = channels_b.update(&mut cx_b, |this, cx| {
3798            this.get_channel(channel_id.to_proto(), cx).unwrap()
3799        });
3800        channel_b.read_with(&cx_b, |channel, _| assert!(channel.messages().is_empty()));
3801        channel_b
3802            .condition(&cx_b, |channel, _| {
3803                channel_messages(channel)
3804                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
3805            })
3806            .await;
3807
3808        channel_a
3809            .update(&mut cx_a, |channel, cx| {
3810                channel
3811                    .send_message("oh, hi B.".to_string(), cx)
3812                    .unwrap()
3813                    .detach();
3814                let task = channel.send_message("sup".to_string(), cx).unwrap();
3815                assert_eq!(
3816                    channel_messages(channel),
3817                    &[
3818                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
3819                        ("user_a".to_string(), "oh, hi B.".to_string(), true),
3820                        ("user_a".to_string(), "sup".to_string(), true)
3821                    ]
3822                );
3823                task
3824            })
3825            .await
3826            .unwrap();
3827
3828        channel_b
3829            .condition(&cx_b, |channel, _| {
3830                channel_messages(channel)
3831                    == [
3832                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
3833                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
3834                        ("user_a".to_string(), "sup".to_string(), false),
3835                    ]
3836            })
3837            .await;
3838
3839        assert_eq!(
3840            server
3841                .state()
3842                .await
3843                .channel(channel_id)
3844                .unwrap()
3845                .connection_ids
3846                .len(),
3847            2
3848        );
3849        cx_b.update(|_| drop(channel_b));
3850        server
3851            .condition(|state| state.channel(channel_id).unwrap().connection_ids.len() == 1)
3852            .await;
3853
3854        cx_a.update(|_| drop(channel_a));
3855        server
3856            .condition(|state| state.channel(channel_id).is_none())
3857            .await;
3858    }
3859
3860    #[gpui::test(iterations = 10)]
3861    async fn test_chat_message_validation(mut cx_a: TestAppContext) {
3862        cx_a.foreground().forbid_parking();
3863
3864        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3865        let client_a = server.create_client(&mut cx_a, "user_a").await;
3866
3867        let db = &server.app_state.db;
3868        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
3869        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
3870        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
3871            .await
3872            .unwrap();
3873        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
3874            .await
3875            .unwrap();
3876
3877        let channels_a = cx_a
3878            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
3879        channels_a
3880            .condition(&mut cx_a, |list, _| list.available_channels().is_some())
3881            .await;
3882        let channel_a = channels_a.update(&mut cx_a, |this, cx| {
3883            this.get_channel(channel_id.to_proto(), cx).unwrap()
3884        });
3885
3886        // Messages aren't allowed to be too long.
3887        channel_a
3888            .update(&mut cx_a, |channel, cx| {
3889                let long_body = "this is long.\n".repeat(1024);
3890                channel.send_message(long_body, cx).unwrap()
3891            })
3892            .await
3893            .unwrap_err();
3894
3895        // Messages aren't allowed to be blank.
3896        channel_a.update(&mut cx_a, |channel, cx| {
3897            channel.send_message(String::new(), cx).unwrap_err()
3898        });
3899
3900        // Leading and trailing whitespace are trimmed.
3901        channel_a
3902            .update(&mut cx_a, |channel, cx| {
3903                channel
3904                    .send_message("\n surrounded by whitespace  \n".to_string(), cx)
3905                    .unwrap()
3906            })
3907            .await
3908            .unwrap();
3909        assert_eq!(
3910            db.get_channel_messages(channel_id, 10, None)
3911                .await
3912                .unwrap()
3913                .iter()
3914                .map(|m| &m.body)
3915                .collect::<Vec<_>>(),
3916            &["surrounded by whitespace"]
3917        );
3918    }
3919
3920    #[gpui::test(iterations = 10)]
3921    async fn test_chat_reconnection(mut cx_a: TestAppContext, mut cx_b: TestAppContext) {
3922        cx_a.foreground().forbid_parking();
3923
3924        // Connect to a server as 2 clients.
3925        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3926        let client_a = server.create_client(&mut cx_a, "user_a").await;
3927        let client_b = server.create_client(&mut cx_b, "user_b").await;
3928        let mut status_b = client_b.status();
3929
3930        // Create an org that includes these 2 users.
3931        let db = &server.app_state.db;
3932        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
3933        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
3934            .await
3935            .unwrap();
3936        db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
3937            .await
3938            .unwrap();
3939
3940        // Create a channel that includes all the users.
3941        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
3942        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
3943            .await
3944            .unwrap();
3945        db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
3946            .await
3947            .unwrap();
3948        db.create_channel_message(
3949            channel_id,
3950            client_b.current_user_id(&cx_b),
3951            "hello A, it's B.",
3952            OffsetDateTime::now_utc(),
3953            2,
3954        )
3955        .await
3956        .unwrap();
3957
3958        let channels_a = cx_a
3959            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
3960        channels_a
3961            .condition(&mut cx_a, |list, _| list.available_channels().is_some())
3962            .await;
3963
3964        channels_a.read_with(&cx_a, |list, _| {
3965            assert_eq!(
3966                list.available_channels().unwrap(),
3967                &[ChannelDetails {
3968                    id: channel_id.to_proto(),
3969                    name: "test-channel".to_string()
3970                }]
3971            )
3972        });
3973        let channel_a = channels_a.update(&mut cx_a, |this, cx| {
3974            this.get_channel(channel_id.to_proto(), cx).unwrap()
3975        });
3976        channel_a.read_with(&cx_a, |channel, _| assert!(channel.messages().is_empty()));
3977        channel_a
3978            .condition(&cx_a, |channel, _| {
3979                channel_messages(channel)
3980                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
3981            })
3982            .await;
3983
3984        let channels_b = cx_b
3985            .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
3986        channels_b
3987            .condition(&mut cx_b, |list, _| list.available_channels().is_some())
3988            .await;
3989        channels_b.read_with(&cx_b, |list, _| {
3990            assert_eq!(
3991                list.available_channels().unwrap(),
3992                &[ChannelDetails {
3993                    id: channel_id.to_proto(),
3994                    name: "test-channel".to_string()
3995                }]
3996            )
3997        });
3998
3999        let channel_b = channels_b.update(&mut cx_b, |this, cx| {
4000            this.get_channel(channel_id.to_proto(), cx).unwrap()
4001        });
4002        channel_b.read_with(&cx_b, |channel, _| assert!(channel.messages().is_empty()));
4003        channel_b
4004            .condition(&cx_b, |channel, _| {
4005                channel_messages(channel)
4006                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4007            })
4008            .await;
4009
4010        // Disconnect client B, ensuring we can still access its cached channel data.
4011        server.forbid_connections();
4012        server.disconnect_client(client_b.current_user_id(&cx_b));
4013        while !matches!(
4014            status_b.next().await,
4015            Some(client::Status::ReconnectionError { .. })
4016        ) {}
4017
4018        channels_b.read_with(&cx_b, |channels, _| {
4019            assert_eq!(
4020                channels.available_channels().unwrap(),
4021                [ChannelDetails {
4022                    id: channel_id.to_proto(),
4023                    name: "test-channel".to_string()
4024                }]
4025            )
4026        });
4027        channel_b.read_with(&cx_b, |channel, _| {
4028            assert_eq!(
4029                channel_messages(channel),
4030                [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4031            )
4032        });
4033
4034        // Send a message from client B while it is disconnected.
4035        channel_b
4036            .update(&mut cx_b, |channel, cx| {
4037                let task = channel
4038                    .send_message("can you see this?".to_string(), cx)
4039                    .unwrap();
4040                assert_eq!(
4041                    channel_messages(channel),
4042                    &[
4043                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4044                        ("user_b".to_string(), "can you see this?".to_string(), true)
4045                    ]
4046                );
4047                task
4048            })
4049            .await
4050            .unwrap_err();
4051
4052        // Send a message from client A while B is disconnected.
4053        channel_a
4054            .update(&mut cx_a, |channel, cx| {
4055                channel
4056                    .send_message("oh, hi B.".to_string(), cx)
4057                    .unwrap()
4058                    .detach();
4059                let task = channel.send_message("sup".to_string(), cx).unwrap();
4060                assert_eq!(
4061                    channel_messages(channel),
4062                    &[
4063                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4064                        ("user_a".to_string(), "oh, hi B.".to_string(), true),
4065                        ("user_a".to_string(), "sup".to_string(), true)
4066                    ]
4067                );
4068                task
4069            })
4070            .await
4071            .unwrap();
4072
4073        // Give client B a chance to reconnect.
4074        server.allow_connections();
4075        cx_b.foreground().advance_clock(Duration::from_secs(10));
4076
4077        // Verify that B sees the new messages upon reconnection, as well as the message client B
4078        // sent while offline.
4079        channel_b
4080            .condition(&cx_b, |channel, _| {
4081                channel_messages(channel)
4082                    == [
4083                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4084                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4085                        ("user_a".to_string(), "sup".to_string(), false),
4086                        ("user_b".to_string(), "can you see this?".to_string(), false),
4087                    ]
4088            })
4089            .await;
4090
4091        // Ensure client A and B can communicate normally after reconnection.
4092        channel_a
4093            .update(&mut cx_a, |channel, cx| {
4094                channel.send_message("you online?".to_string(), cx).unwrap()
4095            })
4096            .await
4097            .unwrap();
4098        channel_b
4099            .condition(&cx_b, |channel, _| {
4100                channel_messages(channel)
4101                    == [
4102                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4103                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4104                        ("user_a".to_string(), "sup".to_string(), false),
4105                        ("user_b".to_string(), "can you see this?".to_string(), false),
4106                        ("user_a".to_string(), "you online?".to_string(), false),
4107                    ]
4108            })
4109            .await;
4110
4111        channel_b
4112            .update(&mut cx_b, |channel, cx| {
4113                channel.send_message("yep".to_string(), cx).unwrap()
4114            })
4115            .await
4116            .unwrap();
4117        channel_a
4118            .condition(&cx_a, |channel, _| {
4119                channel_messages(channel)
4120                    == [
4121                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4122                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4123                        ("user_a".to_string(), "sup".to_string(), false),
4124                        ("user_b".to_string(), "can you see this?".to_string(), false),
4125                        ("user_a".to_string(), "you online?".to_string(), false),
4126                        ("user_b".to_string(), "yep".to_string(), false),
4127                    ]
4128            })
4129            .await;
4130    }
4131
4132    #[gpui::test(iterations = 10)]
4133    async fn test_contacts(
4134        mut cx_a: TestAppContext,
4135        mut cx_b: TestAppContext,
4136        mut cx_c: TestAppContext,
4137    ) {
4138        cx_a.foreground().forbid_parking();
4139        let lang_registry = Arc::new(LanguageRegistry::new());
4140        let fs = FakeFs::new(cx_a.background());
4141
4142        // Connect to a server as 3 clients.
4143        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4144        let client_a = server.create_client(&mut cx_a, "user_a").await;
4145        let client_b = server.create_client(&mut cx_b, "user_b").await;
4146        let client_c = server.create_client(&mut cx_c, "user_c").await;
4147
4148        // Share a worktree as client A.
4149        fs.insert_tree(
4150            "/a",
4151            json!({
4152                ".zed.toml": r#"collaborators = ["user_b", "user_c"]"#,
4153            }),
4154        )
4155        .await;
4156
4157        let project_a = cx_a.update(|cx| {
4158            Project::local(
4159                client_a.clone(),
4160                client_a.user_store.clone(),
4161                lang_registry.clone(),
4162                fs.clone(),
4163                cx,
4164            )
4165        });
4166        let (worktree_a, _) = project_a
4167            .update(&mut cx_a, |p, cx| {
4168                p.find_or_create_local_worktree("/a", false, cx)
4169            })
4170            .await
4171            .unwrap();
4172        worktree_a
4173            .read_with(&cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
4174            .await;
4175
4176        client_a
4177            .user_store
4178            .condition(&cx_a, |user_store, _| {
4179                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4180            })
4181            .await;
4182        client_b
4183            .user_store
4184            .condition(&cx_b, |user_store, _| {
4185                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4186            })
4187            .await;
4188        client_c
4189            .user_store
4190            .condition(&cx_c, |user_store, _| {
4191                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4192            })
4193            .await;
4194
4195        let project_id = project_a
4196            .update(&mut cx_a, |project, _| project.next_remote_id())
4197            .await;
4198        project_a
4199            .update(&mut cx_a, |project, cx| project.share(cx))
4200            .await
4201            .unwrap();
4202
4203        let _project_b = Project::remote(
4204            project_id,
4205            client_b.clone(),
4206            client_b.user_store.clone(),
4207            lang_registry.clone(),
4208            fs.clone(),
4209            &mut cx_b.to_async(),
4210        )
4211        .await
4212        .unwrap();
4213
4214        client_a
4215            .user_store
4216            .condition(&cx_a, |user_store, _| {
4217                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4218            })
4219            .await;
4220        client_b
4221            .user_store
4222            .condition(&cx_b, |user_store, _| {
4223                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4224            })
4225            .await;
4226        client_c
4227            .user_store
4228            .condition(&cx_c, |user_store, _| {
4229                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4230            })
4231            .await;
4232
4233        project_a
4234            .condition(&cx_a, |project, _| {
4235                project.collaborators().contains_key(&client_b.peer_id)
4236            })
4237            .await;
4238
4239        cx_a.update(move |_| drop(project_a));
4240        client_a
4241            .user_store
4242            .condition(&cx_a, |user_store, _| contacts(user_store) == vec![])
4243            .await;
4244        client_b
4245            .user_store
4246            .condition(&cx_b, |user_store, _| contacts(user_store) == vec![])
4247            .await;
4248        client_c
4249            .user_store
4250            .condition(&cx_c, |user_store, _| contacts(user_store) == vec![])
4251            .await;
4252
4253        fn contacts(user_store: &UserStore) -> Vec<(&str, Vec<(&str, Vec<&str>)>)> {
4254            user_store
4255                .contacts()
4256                .iter()
4257                .map(|contact| {
4258                    let worktrees = contact
4259                        .projects
4260                        .iter()
4261                        .map(|p| {
4262                            (
4263                                p.worktree_root_names[0].as_str(),
4264                                p.guests.iter().map(|p| p.github_login.as_str()).collect(),
4265                            )
4266                        })
4267                        .collect();
4268                    (contact.user.github_login.as_str(), worktrees)
4269                })
4270                .collect()
4271        }
4272    }
4273
4274    #[gpui::test(iterations = 100)]
4275    async fn test_random_collaboration(cx: TestAppContext, rng: StdRng) {
4276        cx.foreground().forbid_parking();
4277        let max_peers = env::var("MAX_PEERS")
4278            .map(|i| i.parse().expect("invalid `MAX_PEERS` variable"))
4279            .unwrap_or(5);
4280        let max_operations = env::var("OPERATIONS")
4281            .map(|i| i.parse().expect("invalid `OPERATIONS` variable"))
4282            .unwrap_or(10);
4283
4284        let rng = Arc::new(Mutex::new(rng));
4285
4286        let guest_lang_registry = Arc::new(LanguageRegistry::new());
4287        let (language_server_config, _fake_language_servers) = LanguageServerConfig::fake();
4288
4289        let fs = FakeFs::new(cx.background());
4290        fs.insert_tree(
4291            "/_collab",
4292            json!({
4293                ".zed.toml": r#"collaborators = ["guest-1", "guest-2", "guest-3", "guest-4", "guest-5"]"#
4294            }),
4295        )
4296        .await;
4297
4298        let operations = Rc::new(Cell::new(0));
4299        let mut server = TestServer::start(cx.foreground(), cx.background()).await;
4300        let mut clients = Vec::new();
4301
4302        let mut next_entity_id = 100000;
4303        let mut host_cx = TestAppContext::new(
4304            cx.foreground_platform(),
4305            cx.platform(),
4306            cx.foreground(),
4307            cx.background(),
4308            cx.font_cache(),
4309            next_entity_id,
4310        );
4311        let host = server.create_client(&mut host_cx, "host").await;
4312        let host_project = host_cx.update(|cx| {
4313            Project::local(
4314                host.client.clone(),
4315                host.user_store.clone(),
4316                Arc::new(LanguageRegistry::new()),
4317                fs.clone(),
4318                cx,
4319            )
4320        });
4321        let host_project_id = host_project
4322            .update(&mut host_cx, |p, _| p.next_remote_id())
4323            .await;
4324
4325        let (collab_worktree, _) = host_project
4326            .update(&mut host_cx, |project, cx| {
4327                project.find_or_create_local_worktree("/_collab", false, cx)
4328            })
4329            .await
4330            .unwrap();
4331        collab_worktree
4332            .read_with(&host_cx, |tree, _| tree.as_local().unwrap().scan_complete())
4333            .await;
4334        host_project
4335            .update(&mut host_cx, |project, cx| project.share(cx))
4336            .await
4337            .unwrap();
4338
4339        clients.push(cx.foreground().spawn(host.simulate_host(
4340            host_project.clone(),
4341            language_server_config,
4342            operations.clone(),
4343            max_operations,
4344            rng.clone(),
4345            host_cx.clone(),
4346        )));
4347
4348        while operations.get() < max_operations {
4349            cx.background().simulate_random_delay().await;
4350            if clients.len() < max_peers && rng.lock().gen_bool(0.05) {
4351                operations.set(operations.get() + 1);
4352
4353                let guest_id = clients.len();
4354                log::info!("Adding guest {}", guest_id);
4355                next_entity_id += 100000;
4356                let mut guest_cx = TestAppContext::new(
4357                    cx.foreground_platform(),
4358                    cx.platform(),
4359                    cx.foreground(),
4360                    cx.background(),
4361                    cx.font_cache(),
4362                    next_entity_id,
4363                );
4364                let guest = server
4365                    .create_client(&mut guest_cx, &format!("guest-{}", guest_id))
4366                    .await;
4367                let guest_project = Project::remote(
4368                    host_project_id,
4369                    guest.client.clone(),
4370                    guest.user_store.clone(),
4371                    guest_lang_registry.clone(),
4372                    fs.clone(),
4373                    &mut guest_cx.to_async(),
4374                )
4375                .await
4376                .unwrap();
4377                clients.push(cx.foreground().spawn(guest.simulate_guest(
4378                    guest_id,
4379                    guest_project,
4380                    operations.clone(),
4381                    max_operations,
4382                    rng.clone(),
4383                    guest_cx,
4384                )));
4385
4386                log::info!("Guest {} added", guest_id);
4387            }
4388        }
4389
4390        let clients = futures::future::join_all(clients).await;
4391        cx.foreground().run_until_parked();
4392
4393        let host_worktree_snapshots = host_project.read_with(&host_cx, |project, cx| {
4394            project
4395                .worktrees(cx)
4396                .map(|worktree| {
4397                    let snapshot = worktree.read(cx).snapshot();
4398                    (snapshot.id(), snapshot)
4399                })
4400                .collect::<BTreeMap<_, _>>()
4401        });
4402
4403        for (guest_client, guest_cx) in clients.iter().skip(1) {
4404            let guest_id = guest_client.client.id();
4405            let worktree_snapshots =
4406                guest_client
4407                    .project
4408                    .as_ref()
4409                    .unwrap()
4410                    .read_with(guest_cx, |project, cx| {
4411                        project
4412                            .worktrees(cx)
4413                            .map(|worktree| {
4414                                let worktree = worktree.read(cx);
4415                                assert!(
4416                                    !worktree.as_remote().unwrap().has_pending_updates(),
4417                                    "Guest {} worktree {:?} contains deferred updates",
4418                                    guest_id,
4419                                    worktree.id()
4420                                );
4421                                (worktree.id(), worktree.snapshot())
4422                            })
4423                            .collect::<BTreeMap<_, _>>()
4424                    });
4425
4426            assert_eq!(
4427                worktree_snapshots.keys().collect::<Vec<_>>(),
4428                host_worktree_snapshots.keys().collect::<Vec<_>>(),
4429                "guest {} has different worktrees than the host",
4430                guest_id
4431            );
4432            for (id, host_snapshot) in &host_worktree_snapshots {
4433                let guest_snapshot = &worktree_snapshots[id];
4434                assert_eq!(
4435                    guest_snapshot.root_name(),
4436                    host_snapshot.root_name(),
4437                    "guest {} has different root name than the host for worktree {}",
4438                    guest_id,
4439                    id
4440                );
4441                assert_eq!(
4442                    guest_snapshot.entries(false).collect::<Vec<_>>(),
4443                    host_snapshot.entries(false).collect::<Vec<_>>(),
4444                    "guest {} has different snapshot than the host for worktree {}",
4445                    guest_id,
4446                    id
4447                );
4448            }
4449
4450            guest_client
4451                .project
4452                .as_ref()
4453                .unwrap()
4454                .read_with(guest_cx, |project, _| {
4455                    assert!(
4456                        !project.has_buffered_operations(),
4457                        "guest {} has buffered operations ",
4458                        guest_id,
4459                    );
4460                });
4461
4462            for guest_buffer in &guest_client.buffers {
4463                let buffer_id = guest_buffer.read_with(guest_cx, |buffer, _| buffer.remote_id());
4464                let host_buffer = host_project.read_with(&host_cx, |project, _| {
4465                    project
4466                        .shared_buffer(guest_client.peer_id, buffer_id)
4467                        .expect(&format!(
4468                            "host doest not have buffer for guest:{}, peer:{}, id:{}",
4469                            guest_id, guest_client.peer_id, buffer_id
4470                        ))
4471                });
4472                assert_eq!(
4473                    guest_buffer.read_with(guest_cx, |buffer, _| buffer.text()),
4474                    host_buffer.read_with(&host_cx, |buffer, _| buffer.text()),
4475                    "guest {} buffer {} differs from the host's buffer",
4476                    guest_id,
4477                    buffer_id,
4478                );
4479            }
4480        }
4481    }
4482
4483    struct TestServer {
4484        peer: Arc<Peer>,
4485        app_state: Arc<AppState>,
4486        server: Arc<Server>,
4487        foreground: Rc<executor::Foreground>,
4488        notifications: mpsc::UnboundedReceiver<()>,
4489        connection_killers: Arc<Mutex<HashMap<UserId, watch::Sender<Option<()>>>>>,
4490        forbid_connections: Arc<AtomicBool>,
4491        _test_db: TestDb,
4492    }
4493
4494    impl TestServer {
4495        async fn start(
4496            foreground: Rc<executor::Foreground>,
4497            background: Arc<executor::Background>,
4498        ) -> Self {
4499            let test_db = TestDb::fake(background);
4500            let app_state = Self::build_app_state(&test_db).await;
4501            let peer = Peer::new();
4502            let notifications = mpsc::unbounded();
4503            let server = Server::new(app_state.clone(), peer.clone(), Some(notifications.0));
4504            Self {
4505                peer,
4506                app_state,
4507                server,
4508                foreground,
4509                notifications: notifications.1,
4510                connection_killers: Default::default(),
4511                forbid_connections: Default::default(),
4512                _test_db: test_db,
4513            }
4514        }
4515
4516        async fn create_client(&mut self, cx: &mut TestAppContext, name: &str) -> TestClient {
4517            let http = FakeHttpClient::with_404_response();
4518            let user_id = self.app_state.db.create_user(name, false).await.unwrap();
4519            let client_name = name.to_string();
4520            let mut client = Client::new(http.clone());
4521            let server = self.server.clone();
4522            let connection_killers = self.connection_killers.clone();
4523            let forbid_connections = self.forbid_connections.clone();
4524            let (connection_id_tx, mut connection_id_rx) = mpsc::channel(16);
4525
4526            Arc::get_mut(&mut client)
4527                .unwrap()
4528                .override_authenticate(move |cx| {
4529                    cx.spawn(|_| async move {
4530                        let access_token = "the-token".to_string();
4531                        Ok(Credentials {
4532                            user_id: user_id.0 as u64,
4533                            access_token,
4534                        })
4535                    })
4536                })
4537                .override_establish_connection(move |credentials, cx| {
4538                    assert_eq!(credentials.user_id, user_id.0 as u64);
4539                    assert_eq!(credentials.access_token, "the-token");
4540
4541                    let server = server.clone();
4542                    let connection_killers = connection_killers.clone();
4543                    let forbid_connections = forbid_connections.clone();
4544                    let client_name = client_name.clone();
4545                    let connection_id_tx = connection_id_tx.clone();
4546                    cx.spawn(move |cx| async move {
4547                        if forbid_connections.load(SeqCst) {
4548                            Err(EstablishConnectionError::other(anyhow!(
4549                                "server is forbidding connections"
4550                            )))
4551                        } else {
4552                            let (client_conn, server_conn, kill_conn) =
4553                                Connection::in_memory(cx.background());
4554                            connection_killers.lock().insert(user_id, kill_conn);
4555                            cx.background()
4556                                .spawn(server.handle_connection(
4557                                    server_conn,
4558                                    client_name,
4559                                    user_id,
4560                                    Some(connection_id_tx),
4561                                    cx.background(),
4562                                ))
4563                                .detach();
4564                            Ok(client_conn)
4565                        }
4566                    })
4567                });
4568
4569            client
4570                .authenticate_and_connect(&cx.to_async())
4571                .await
4572                .unwrap();
4573
4574            Channel::init(&client);
4575            Project::init(&client);
4576
4577            let peer_id = PeerId(connection_id_rx.next().await.unwrap().0);
4578            let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http, cx));
4579            let mut authed_user =
4580                user_store.read_with(cx, |user_store, _| user_store.watch_current_user());
4581            while authed_user.next().await.unwrap().is_none() {}
4582
4583            TestClient {
4584                client,
4585                peer_id,
4586                user_store,
4587                project: Default::default(),
4588                buffers: Default::default(),
4589            }
4590        }
4591
4592        fn disconnect_client(&self, user_id: UserId) {
4593            if let Some(mut kill_conn) = self.connection_killers.lock().remove(&user_id) {
4594                let _ = kill_conn.try_send(Some(()));
4595            }
4596        }
4597
4598        fn forbid_connections(&self) {
4599            self.forbid_connections.store(true, SeqCst);
4600        }
4601
4602        fn allow_connections(&self) {
4603            self.forbid_connections.store(false, SeqCst);
4604        }
4605
4606        async fn build_app_state(test_db: &TestDb) -> Arc<AppState> {
4607            let mut config = Config::default();
4608            config.session_secret = "a".repeat(32);
4609            config.database_url = test_db.url.clone();
4610            let github_client = github::AppClient::test();
4611            Arc::new(AppState {
4612                db: test_db.db().clone(),
4613                handlebars: Default::default(),
4614                auth_client: auth::build_client("", ""),
4615                repo_client: github::RepoClient::test(&github_client),
4616                github_client,
4617                config,
4618            })
4619        }
4620
4621        async fn state<'a>(&'a self) -> RwLockReadGuard<'a, Store> {
4622            self.server.store.read()
4623        }
4624
4625        async fn condition<F>(&mut self, mut predicate: F)
4626        where
4627            F: FnMut(&Store) -> bool,
4628        {
4629            async_std::future::timeout(Duration::from_millis(500), async {
4630                while !(predicate)(&*self.server.store.read()) {
4631                    self.foreground.start_waiting();
4632                    self.notifications.next().await;
4633                    self.foreground.finish_waiting();
4634                }
4635            })
4636            .await
4637            .expect("condition timed out");
4638        }
4639    }
4640
4641    impl Drop for TestServer {
4642        fn drop(&mut self) {
4643            self.peer.reset();
4644        }
4645    }
4646
4647    struct TestClient {
4648        client: Arc<Client>,
4649        pub peer_id: PeerId,
4650        pub user_store: ModelHandle<UserStore>,
4651        project: Option<ModelHandle<Project>>,
4652        buffers: HashSet<ModelHandle<zed::language::Buffer>>,
4653    }
4654
4655    impl Deref for TestClient {
4656        type Target = Arc<Client>;
4657
4658        fn deref(&self) -> &Self::Target {
4659            &self.client
4660        }
4661    }
4662
4663    impl TestClient {
4664        pub fn current_user_id(&self, cx: &TestAppContext) -> UserId {
4665            UserId::from_proto(
4666                self.user_store
4667                    .read_with(cx, |user_store, _| user_store.current_user().unwrap().id),
4668            )
4669        }
4670
4671        fn simulate_host(
4672            mut self,
4673            project: ModelHandle<Project>,
4674            mut language_server_config: LanguageServerConfig,
4675            operations: Rc<Cell<usize>>,
4676            max_operations: usize,
4677            rng: Arc<Mutex<StdRng>>,
4678            mut cx: TestAppContext,
4679        ) -> impl Future<Output = (Self, TestAppContext)> {
4680            let files: Arc<Mutex<Vec<PathBuf>>> = Default::default();
4681
4682            // Set up a fake language server.
4683            language_server_config.set_fake_initializer({
4684                let rng = rng.clone();
4685                let files = files.clone();
4686                move |fake_server| {
4687                    fake_server.handle_request::<lsp::request::Completion, _>(|_| {
4688                        Some(lsp::CompletionResponse::Array(vec![lsp::CompletionItem {
4689                            text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
4690                                range: lsp::Range::new(
4691                                    lsp::Position::new(0, 0),
4692                                    lsp::Position::new(0, 0),
4693                                ),
4694                                new_text: "the-new-text".to_string(),
4695                            })),
4696                            ..Default::default()
4697                        }]))
4698                    });
4699
4700                    fake_server.handle_request::<lsp::request::CodeActionRequest, _>(|_| {
4701                        Some(vec![lsp::CodeActionOrCommand::CodeAction(
4702                            lsp::CodeAction {
4703                                title: "the-code-action".to_string(),
4704                                ..Default::default()
4705                            },
4706                        )])
4707                    });
4708
4709                    fake_server.handle_request::<lsp::request::PrepareRenameRequest, _>(|params| {
4710                        Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
4711                            params.position,
4712                            params.position,
4713                        )))
4714                    });
4715
4716                    fake_server.handle_request::<lsp::request::GotoDefinition, _>({
4717                        let files = files.clone();
4718                        let rng = rng.clone();
4719                        move |_| {
4720                            let files = files.lock();
4721                            let mut rng = rng.lock();
4722                            let count = rng.gen_range::<usize, _>(1..3);
4723                            Some(lsp::GotoDefinitionResponse::Array(
4724                                (0..count)
4725                                    .map(|_| {
4726                                        let file = files.choose(&mut *rng).unwrap().as_path();
4727                                        lsp::Location {
4728                                            uri: lsp::Url::from_file_path(file).unwrap(),
4729                                            range: Default::default(),
4730                                        }
4731                                    })
4732                                    .collect(),
4733                            ))
4734                        }
4735                    });
4736                }
4737            });
4738
4739            project.update(&mut cx, |project, _| {
4740                project.languages().add(Arc::new(Language::new(
4741                    LanguageConfig {
4742                        name: "Rust".into(),
4743                        path_suffixes: vec!["rs".to_string()],
4744                        language_server: Some(language_server_config),
4745                        ..Default::default()
4746                    },
4747                    None,
4748                )));
4749            });
4750
4751            async move {
4752                let fs = project.read_with(&cx, |project, _| project.fs().clone());
4753                while operations.get() < max_operations {
4754                    operations.set(operations.get() + 1);
4755
4756                    let distribution = rng.lock().gen_range::<usize, _>(0..100);
4757                    match distribution {
4758                        0..=20 if !files.lock().is_empty() => {
4759                            let path = files.lock().choose(&mut *rng.lock()).unwrap().clone();
4760                            let mut path = path.as_path();
4761                            while let Some(parent_path) = path.parent() {
4762                                path = parent_path;
4763                                if rng.lock().gen() {
4764                                    break;
4765                                }
4766                            }
4767
4768                            log::info!("Host: find/create local worktree {:?}", path);
4769                            project
4770                                .update(&mut cx, |project, cx| {
4771                                    project.find_or_create_local_worktree(path, false, cx)
4772                                })
4773                                .await
4774                                .unwrap();
4775                        }
4776                        10..=80 if !files.lock().is_empty() => {
4777                            let buffer = if self.buffers.is_empty() || rng.lock().gen() {
4778                                let file = files.lock().choose(&mut *rng.lock()).unwrap().clone();
4779                                let (worktree, path) = project
4780                                    .update(&mut cx, |project, cx| {
4781                                        project.find_or_create_local_worktree(file, false, cx)
4782                                    })
4783                                    .await
4784                                    .unwrap();
4785                                let project_path =
4786                                    worktree.read_with(&cx, |worktree, _| (worktree.id(), path));
4787                                log::info!("Host: opening path {:?}", project_path);
4788                                let buffer = project
4789                                    .update(&mut cx, |project, cx| {
4790                                        project.open_buffer(project_path, cx)
4791                                    })
4792                                    .await
4793                                    .unwrap();
4794                                self.buffers.insert(buffer.clone());
4795                                buffer
4796                            } else {
4797                                self.buffers
4798                                    .iter()
4799                                    .choose(&mut *rng.lock())
4800                                    .unwrap()
4801                                    .clone()
4802                            };
4803
4804                            if rng.lock().gen_bool(0.1) {
4805                                cx.update(|cx| {
4806                                    log::info!(
4807                                        "Host: dropping buffer {:?}",
4808                                        buffer.read(cx).file().unwrap().full_path(cx)
4809                                    );
4810                                    self.buffers.remove(&buffer);
4811                                    drop(buffer);
4812                                });
4813                            } else {
4814                                buffer.update(&mut cx, |buffer, cx| {
4815                                    log::info!(
4816                                        "Host: updating buffer {:?}",
4817                                        buffer.file().unwrap().full_path(cx)
4818                                    );
4819                                    buffer.randomly_edit(&mut *rng.lock(), 5, cx)
4820                                });
4821                            }
4822                        }
4823                        _ => loop {
4824                            let path_component_count = rng.lock().gen_range::<usize, _>(1..=5);
4825                            let mut path = PathBuf::new();
4826                            path.push("/");
4827                            for _ in 0..path_component_count {
4828                                let letter = rng.lock().gen_range(b'a'..=b'z');
4829                                path.push(std::str::from_utf8(&[letter]).unwrap());
4830                            }
4831                            path.set_extension("rs");
4832                            let parent_path = path.parent().unwrap();
4833
4834                            log::info!("Host: creating file {:?}", path,);
4835
4836                            if fs.create_dir(&parent_path).await.is_ok()
4837                                && fs.create_file(&path, Default::default()).await.is_ok()
4838                            {
4839                                files.lock().push(path);
4840                                break;
4841                            } else {
4842                                log::info!("Host: cannot create file");
4843                            }
4844                        },
4845                    }
4846
4847                    cx.background().simulate_random_delay().await;
4848                }
4849
4850                self.project = Some(project);
4851                (self, cx)
4852            }
4853        }
4854
4855        pub async fn simulate_guest(
4856            mut self,
4857            guest_id: usize,
4858            project: ModelHandle<Project>,
4859            operations: Rc<Cell<usize>>,
4860            max_operations: usize,
4861            rng: Arc<Mutex<StdRng>>,
4862            mut cx: TestAppContext,
4863        ) -> (Self, TestAppContext) {
4864            while operations.get() < max_operations {
4865                let buffer = if self.buffers.is_empty() || rng.lock().gen() {
4866                    let worktree = if let Some(worktree) = project.read_with(&cx, |project, cx| {
4867                        project
4868                            .worktrees(&cx)
4869                            .filter(|worktree| {
4870                                worktree.read(cx).entries(false).any(|e| e.is_file())
4871                            })
4872                            .choose(&mut *rng.lock())
4873                    }) {
4874                        worktree
4875                    } else {
4876                        cx.background().simulate_random_delay().await;
4877                        continue;
4878                    };
4879
4880                    operations.set(operations.get() + 1);
4881                    let project_path = worktree.read_with(&cx, |worktree, _| {
4882                        let entry = worktree
4883                            .entries(false)
4884                            .filter(|e| e.is_file())
4885                            .choose(&mut *rng.lock())
4886                            .unwrap();
4887                        (worktree.id(), entry.path.clone())
4888                    });
4889                    log::info!("Guest {}: opening path {:?}", guest_id, project_path);
4890                    let buffer = project
4891                        .update(&mut cx, |project, cx| project.open_buffer(project_path, cx))
4892                        .await
4893                        .unwrap();
4894                    self.buffers.insert(buffer.clone());
4895                    buffer
4896                } else {
4897                    operations.set(operations.get() + 1);
4898
4899                    self.buffers
4900                        .iter()
4901                        .choose(&mut *rng.lock())
4902                        .unwrap()
4903                        .clone()
4904                };
4905
4906                let choice = rng.lock().gen_range(0..100);
4907                match choice {
4908                    0..=9 => {
4909                        cx.update(|cx| {
4910                            log::info!(
4911                                "Guest {}: dropping buffer {:?}",
4912                                guest_id,
4913                                buffer.read(cx).file().unwrap().full_path(cx)
4914                            );
4915                            self.buffers.remove(&buffer);
4916                            drop(buffer);
4917                        });
4918                    }
4919                    10..=19 => {
4920                        let completions = project.update(&mut cx, |project, cx| {
4921                            log::info!(
4922                                "Guest {}: requesting completions for buffer {:?}",
4923                                guest_id,
4924                                buffer.read(cx).file().unwrap().full_path(cx)
4925                            );
4926                            let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
4927                            project.completions(&buffer, offset, cx)
4928                        });
4929                        let completions = cx.background().spawn(async move {
4930                            completions.await.expect("completions request failed");
4931                        });
4932                        if rng.lock().gen_bool(0.3) {
4933                            log::info!("Guest {}: detaching completions request", guest_id);
4934                            completions.detach();
4935                        } else {
4936                            completions.await;
4937                        }
4938                    }
4939                    20..=29 => {
4940                        let code_actions = project.update(&mut cx, |project, cx| {
4941                            log::info!(
4942                                "Guest {}: requesting code actions for buffer {:?}",
4943                                guest_id,
4944                                buffer.read(cx).file().unwrap().full_path(cx)
4945                            );
4946                            let range = buffer.read(cx).random_byte_range(0, &mut *rng.lock());
4947                            project.code_actions(&buffer, range, cx)
4948                        });
4949                        let code_actions = cx.background().spawn(async move {
4950                            code_actions.await.expect("code actions request failed");
4951                        });
4952                        if rng.lock().gen_bool(0.3) {
4953                            log::info!("Guest {}: detaching code actions request", guest_id);
4954                            code_actions.detach();
4955                        } else {
4956                            code_actions.await;
4957                        }
4958                    }
4959                    30..=39 if buffer.read_with(&cx, |buffer, _| buffer.is_dirty()) => {
4960                        let (requested_version, save) = buffer.update(&mut cx, |buffer, cx| {
4961                            log::info!(
4962                                "Guest {}: saving buffer {:?}",
4963                                guest_id,
4964                                buffer.file().unwrap().full_path(cx)
4965                            );
4966                            (buffer.version(), buffer.save(cx))
4967                        });
4968                        let save = cx.spawn(|cx| async move {
4969                            let (saved_version, _) = save.await.expect("save request failed");
4970                            buffer.read_with(&cx, |buffer, _| {
4971                                assert!(buffer.version().observed_all(&saved_version));
4972                                assert!(saved_version.observed_all(&requested_version));
4973                            });
4974                        });
4975                        if rng.lock().gen_bool(0.3) {
4976                            log::info!("Guest {}: detaching save request", guest_id);
4977                            save.detach();
4978                        } else {
4979                            save.await;
4980                        }
4981                    }
4982                    40..=45 => {
4983                        let prepare_rename = project.update(&mut cx, |project, cx| {
4984                            log::info!(
4985                                "Guest {}: preparing rename for buffer {:?}",
4986                                guest_id,
4987                                buffer.read(cx).file().unwrap().full_path(cx)
4988                            );
4989                            let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
4990                            project.prepare_rename(buffer, offset, cx)
4991                        });
4992                        let prepare_rename = cx.background().spawn(async move {
4993                            prepare_rename.await.expect("prepare rename request failed");
4994                        });
4995                        if rng.lock().gen_bool(0.3) {
4996                            log::info!("Guest {}: detaching prepare rename request", guest_id);
4997                            prepare_rename.detach();
4998                        } else {
4999                            prepare_rename.await;
5000                        }
5001                    }
5002                    46..=49 => {
5003                        let definitions = project.update(&mut cx, |project, cx| {
5004                            log::info!(
5005                                "Guest {}: requesting defintions for buffer {:?}",
5006                                guest_id,
5007                                buffer.read(cx).file().unwrap().full_path(cx)
5008                            );
5009                            let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
5010                            project.definition(&buffer, offset, cx)
5011                        });
5012                        let definitions = cx.background().spawn(async move {
5013                            definitions.await.expect("definitions request failed");
5014                        });
5015                        if rng.lock().gen_bool(0.3) {
5016                            log::info!("Guest {}: detaching definitions request", guest_id);
5017                            definitions.detach();
5018                        } else {
5019                            definitions.await;
5020                        }
5021                    }
5022                    _ => {
5023                        buffer.update(&mut cx, |buffer, cx| {
5024                            log::info!(
5025                                "Guest {}: updating buffer {:?}",
5026                                guest_id,
5027                                buffer.file().unwrap().full_path(cx)
5028                            );
5029                            buffer.randomly_edit(&mut *rng.lock(), 5, cx)
5030                        });
5031                    }
5032                }
5033                cx.background().simulate_random_delay().await;
5034            }
5035
5036            self.project = Some(project);
5037            (self, cx)
5038        }
5039    }
5040
5041    impl Executor for Arc<gpui::executor::Background> {
5042        fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F) {
5043            self.spawn(future).detach();
5044        }
5045    }
5046
5047    fn channel_messages(channel: &Channel) -> Vec<(String, String, bool)> {
5048        channel
5049            .messages()
5050            .cursor::<()>()
5051            .map(|m| {
5052                (
5053                    m.sender.github_login.clone(),
5054                    m.body.clone(),
5055                    m.is_pending(),
5056                )
5057            })
5058            .collect()
5059    }
5060
5061    struct EmptyView;
5062
5063    impl gpui::Entity for EmptyView {
5064        type Event = ();
5065    }
5066
5067    impl gpui::View for EmptyView {
5068        fn ui_name() -> &'static str {
5069            "empty view"
5070        }
5071
5072        fn render(&mut self, _: &mut gpui::RenderContext<Self>) -> gpui::ElementBox {
5073            gpui::Element::boxed(gpui::elements::Empty)
5074        }
5075    }
5076}