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