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, _| {
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, _| {
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, _>(|_, _| {
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            Some(lsp::GotoDefinitionResponse::Scalar(lsp::Location::new(
2649                lsp::Url::from_file_path("/root-2/b.rs").unwrap(),
2650                lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
2651            )))
2652        });
2653
2654        let definitions_1 = project_b
2655            .update(cx_b, |p, cx| p.definition(&buffer_b, 23, cx))
2656            .await
2657            .unwrap();
2658        cx_b.read(|cx| {
2659            assert_eq!(definitions_1.len(), 1);
2660            assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
2661            let target_buffer = definitions_1[0].buffer.read(cx);
2662            assert_eq!(
2663                target_buffer.text(),
2664                "const TWO: usize = 2;\nconst THREE: usize = 3;"
2665            );
2666            assert_eq!(
2667                definitions_1[0].range.to_point(target_buffer),
2668                Point::new(0, 6)..Point::new(0, 9)
2669            );
2670        });
2671
2672        // Try getting more definitions for the same buffer, ensuring the buffer gets reused from
2673        // the previous call to `definition`.
2674        fake_language_server.handle_request::<lsp::request::GotoDefinition, _>(|_, _| {
2675            Some(lsp::GotoDefinitionResponse::Scalar(lsp::Location::new(
2676                lsp::Url::from_file_path("/root-2/b.rs").unwrap(),
2677                lsp::Range::new(lsp::Position::new(1, 6), lsp::Position::new(1, 11)),
2678            )))
2679        });
2680
2681        let definitions_2 = project_b
2682            .update(cx_b, |p, cx| p.definition(&buffer_b, 33, cx))
2683            .await
2684            .unwrap();
2685        cx_b.read(|cx| {
2686            assert_eq!(definitions_2.len(), 1);
2687            assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
2688            let target_buffer = definitions_2[0].buffer.read(cx);
2689            assert_eq!(
2690                target_buffer.text(),
2691                "const TWO: usize = 2;\nconst THREE: usize = 3;"
2692            );
2693            assert_eq!(
2694                definitions_2[0].range.to_point(target_buffer),
2695                Point::new(1, 6)..Point::new(1, 11)
2696            );
2697        });
2698        assert_eq!(definitions_1[0].buffer, definitions_2[0].buffer);
2699    }
2700
2701    #[gpui::test(iterations = 10)]
2702    async fn test_references(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2703        cx_a.foreground().forbid_parking();
2704        let mut lang_registry = Arc::new(LanguageRegistry::test());
2705        let fs = FakeFs::new(cx_a.background());
2706        fs.insert_tree(
2707            "/root-1",
2708            json!({
2709                ".zed.toml": r#"collaborators = ["user_b"]"#,
2710                "one.rs": "const ONE: usize = 1;",
2711                "two.rs": "const TWO: usize = one::ONE + one::ONE;",
2712            }),
2713        )
2714        .await;
2715        fs.insert_tree(
2716            "/root-2",
2717            json!({
2718                "three.rs": "const THREE: usize = two::TWO + one::ONE;",
2719            }),
2720        )
2721        .await;
2722
2723        // Set up a fake language server.
2724        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
2725        Arc::get_mut(&mut lang_registry)
2726            .unwrap()
2727            .add(Arc::new(Language::new(
2728                LanguageConfig {
2729                    name: "Rust".into(),
2730                    path_suffixes: vec!["rs".to_string()],
2731                    language_server: Some(language_server_config),
2732                    ..Default::default()
2733                },
2734                Some(tree_sitter_rust::language()),
2735            )));
2736
2737        // Connect to a server as 2 clients.
2738        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2739        let client_a = server.create_client(cx_a, "user_a").await;
2740        let client_b = server.create_client(cx_b, "user_b").await;
2741
2742        // Share a project as client A
2743        let project_a = cx_a.update(|cx| {
2744            Project::local(
2745                client_a.clone(),
2746                client_a.user_store.clone(),
2747                lang_registry.clone(),
2748                fs.clone(),
2749                cx,
2750            )
2751        });
2752        let (worktree_a, _) = project_a
2753            .update(cx_a, |p, cx| {
2754                p.find_or_create_local_worktree("/root-1", true, cx)
2755            })
2756            .await
2757            .unwrap();
2758        worktree_a
2759            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2760            .await;
2761        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
2762        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2763        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
2764
2765        // Join the worktree as client B.
2766        let project_b = Project::remote(
2767            project_id,
2768            client_b.clone(),
2769            client_b.user_store.clone(),
2770            lang_registry.clone(),
2771            fs.clone(),
2772            &mut cx_b.to_async(),
2773        )
2774        .await
2775        .unwrap();
2776
2777        // Open the file on client B.
2778        let buffer_b = cx_b
2779            .background()
2780            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "one.rs"), cx)))
2781            .await
2782            .unwrap();
2783
2784        // Request references to a symbol as the guest.
2785        let mut fake_language_server = fake_language_servers.next().await.unwrap();
2786        fake_language_server.handle_request::<lsp::request::References, _>(|params, _| {
2787            assert_eq!(
2788                params.text_document_position.text_document.uri.as_str(),
2789                "file:///root-1/one.rs"
2790            );
2791            Some(vec![
2792                lsp::Location {
2793                    uri: lsp::Url::from_file_path("/root-1/two.rs").unwrap(),
2794                    range: lsp::Range::new(lsp::Position::new(0, 24), lsp::Position::new(0, 27)),
2795                },
2796                lsp::Location {
2797                    uri: lsp::Url::from_file_path("/root-1/two.rs").unwrap(),
2798                    range: lsp::Range::new(lsp::Position::new(0, 35), lsp::Position::new(0, 38)),
2799                },
2800                lsp::Location {
2801                    uri: lsp::Url::from_file_path("/root-2/three.rs").unwrap(),
2802                    range: lsp::Range::new(lsp::Position::new(0, 37), lsp::Position::new(0, 40)),
2803                },
2804            ])
2805        });
2806
2807        let references = project_b
2808            .update(cx_b, |p, cx| p.references(&buffer_b, 7, cx))
2809            .await
2810            .unwrap();
2811        cx_b.read(|cx| {
2812            assert_eq!(references.len(), 3);
2813            assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
2814
2815            let two_buffer = references[0].buffer.read(cx);
2816            let three_buffer = references[2].buffer.read(cx);
2817            assert_eq!(
2818                two_buffer.file().unwrap().path().as_ref(),
2819                Path::new("two.rs")
2820            );
2821            assert_eq!(references[1].buffer, references[0].buffer);
2822            assert_eq!(
2823                three_buffer.file().unwrap().full_path(cx),
2824                Path::new("three.rs")
2825            );
2826
2827            assert_eq!(references[0].range.to_offset(&two_buffer), 24..27);
2828            assert_eq!(references[1].range.to_offset(&two_buffer), 35..38);
2829            assert_eq!(references[2].range.to_offset(&three_buffer), 37..40);
2830        });
2831    }
2832
2833    #[gpui::test(iterations = 10)]
2834    async fn test_project_search(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2835        cx_a.foreground().forbid_parking();
2836        let lang_registry = Arc::new(LanguageRegistry::test());
2837        let fs = FakeFs::new(cx_a.background());
2838        fs.insert_tree(
2839            "/root-1",
2840            json!({
2841                ".zed.toml": r#"collaborators = ["user_b"]"#,
2842                "a": "hello world",
2843                "b": "goodnight moon",
2844                "c": "a world of goo",
2845                "d": "world champion of clown world",
2846            }),
2847        )
2848        .await;
2849        fs.insert_tree(
2850            "/root-2",
2851            json!({
2852                "e": "disney world is fun",
2853            }),
2854        )
2855        .await;
2856
2857        // Connect to a server as 2 clients.
2858        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2859        let client_a = server.create_client(cx_a, "user_a").await;
2860        let client_b = server.create_client(cx_b, "user_b").await;
2861
2862        // Share a project as client A
2863        let project_a = cx_a.update(|cx| {
2864            Project::local(
2865                client_a.clone(),
2866                client_a.user_store.clone(),
2867                lang_registry.clone(),
2868                fs.clone(),
2869                cx,
2870            )
2871        });
2872        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
2873
2874        let (worktree_1, _) = project_a
2875            .update(cx_a, |p, cx| {
2876                p.find_or_create_local_worktree("/root-1", true, cx)
2877            })
2878            .await
2879            .unwrap();
2880        worktree_1
2881            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2882            .await;
2883        let (worktree_2, _) = project_a
2884            .update(cx_a, |p, cx| {
2885                p.find_or_create_local_worktree("/root-2", true, cx)
2886            })
2887            .await
2888            .unwrap();
2889        worktree_2
2890            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2891            .await;
2892
2893        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
2894
2895        // Join the worktree as client B.
2896        let project_b = Project::remote(
2897            project_id,
2898            client_b.clone(),
2899            client_b.user_store.clone(),
2900            lang_registry.clone(),
2901            fs.clone(),
2902            &mut cx_b.to_async(),
2903        )
2904        .await
2905        .unwrap();
2906
2907        let results = project_b
2908            .update(cx_b, |project, cx| {
2909                project.search(SearchQuery::text("world", false, false), cx)
2910            })
2911            .await
2912            .unwrap();
2913
2914        let mut ranges_by_path = results
2915            .into_iter()
2916            .map(|(buffer, ranges)| {
2917                buffer.read_with(cx_b, |buffer, cx| {
2918                    let path = buffer.file().unwrap().full_path(cx);
2919                    let offset_ranges = ranges
2920                        .into_iter()
2921                        .map(|range| range.to_offset(buffer))
2922                        .collect::<Vec<_>>();
2923                    (path, offset_ranges)
2924                })
2925            })
2926            .collect::<Vec<_>>();
2927        ranges_by_path.sort_by_key(|(path, _)| path.clone());
2928
2929        assert_eq!(
2930            ranges_by_path,
2931            &[
2932                (PathBuf::from("root-1/a"), vec![6..11]),
2933                (PathBuf::from("root-1/c"), vec![2..7]),
2934                (PathBuf::from("root-1/d"), vec![0..5, 24..29]),
2935                (PathBuf::from("root-2/e"), vec![7..12]),
2936            ]
2937        );
2938    }
2939
2940    #[gpui::test(iterations = 10)]
2941    async fn test_document_highlights(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2942        cx_a.foreground().forbid_parking();
2943        let lang_registry = Arc::new(LanguageRegistry::test());
2944        let fs = FakeFs::new(cx_a.background());
2945        fs.insert_tree(
2946            "/root-1",
2947            json!({
2948                ".zed.toml": r#"collaborators = ["user_b"]"#,
2949                "main.rs": "fn double(number: i32) -> i32 { number + number }",
2950            }),
2951        )
2952        .await;
2953
2954        // Set up a fake language server.
2955        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
2956        lang_registry.add(Arc::new(Language::new(
2957            LanguageConfig {
2958                name: "Rust".into(),
2959                path_suffixes: vec!["rs".to_string()],
2960                language_server: Some(language_server_config),
2961                ..Default::default()
2962            },
2963            Some(tree_sitter_rust::language()),
2964        )));
2965
2966        // Connect to a server as 2 clients.
2967        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2968        let client_a = server.create_client(cx_a, "user_a").await;
2969        let client_b = server.create_client(cx_b, "user_b").await;
2970
2971        // Share a project as client A
2972        let project_a = cx_a.update(|cx| {
2973            Project::local(
2974                client_a.clone(),
2975                client_a.user_store.clone(),
2976                lang_registry.clone(),
2977                fs.clone(),
2978                cx,
2979            )
2980        });
2981        let (worktree_a, _) = project_a
2982            .update(cx_a, |p, cx| {
2983                p.find_or_create_local_worktree("/root-1", true, cx)
2984            })
2985            .await
2986            .unwrap();
2987        worktree_a
2988            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2989            .await;
2990        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
2991        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2992        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
2993
2994        // Join the worktree as client B.
2995        let project_b = Project::remote(
2996            project_id,
2997            client_b.clone(),
2998            client_b.user_store.clone(),
2999            lang_registry.clone(),
3000            fs.clone(),
3001            &mut cx_b.to_async(),
3002        )
3003        .await
3004        .unwrap();
3005
3006        // Open the file on client B.
3007        let buffer_b = cx_b
3008            .background()
3009            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "main.rs"), cx)))
3010            .await
3011            .unwrap();
3012
3013        // Request document highlights as the guest.
3014        let mut fake_language_server = fake_language_servers.next().await.unwrap();
3015        fake_language_server.handle_request::<lsp::request::DocumentHighlightRequest, _>(
3016            |params, _| {
3017                assert_eq!(
3018                    params
3019                        .text_document_position_params
3020                        .text_document
3021                        .uri
3022                        .as_str(),
3023                    "file:///root-1/main.rs"
3024                );
3025                assert_eq!(
3026                    params.text_document_position_params.position,
3027                    lsp::Position::new(0, 34)
3028                );
3029                Some(vec![
3030                    lsp::DocumentHighlight {
3031                        kind: Some(lsp::DocumentHighlightKind::WRITE),
3032                        range: lsp::Range::new(
3033                            lsp::Position::new(0, 10),
3034                            lsp::Position::new(0, 16),
3035                        ),
3036                    },
3037                    lsp::DocumentHighlight {
3038                        kind: Some(lsp::DocumentHighlightKind::READ),
3039                        range: lsp::Range::new(
3040                            lsp::Position::new(0, 32),
3041                            lsp::Position::new(0, 38),
3042                        ),
3043                    },
3044                    lsp::DocumentHighlight {
3045                        kind: Some(lsp::DocumentHighlightKind::READ),
3046                        range: lsp::Range::new(
3047                            lsp::Position::new(0, 41),
3048                            lsp::Position::new(0, 47),
3049                        ),
3050                    },
3051                ])
3052            },
3053        );
3054
3055        let highlights = project_b
3056            .update(cx_b, |p, cx| p.document_highlights(&buffer_b, 34, cx))
3057            .await
3058            .unwrap();
3059        buffer_b.read_with(cx_b, |buffer, _| {
3060            let snapshot = buffer.snapshot();
3061
3062            let highlights = highlights
3063                .into_iter()
3064                .map(|highlight| (highlight.kind, highlight.range.to_offset(&snapshot)))
3065                .collect::<Vec<_>>();
3066            assert_eq!(
3067                highlights,
3068                &[
3069                    (lsp::DocumentHighlightKind::WRITE, 10..16),
3070                    (lsp::DocumentHighlightKind::READ, 32..38),
3071                    (lsp::DocumentHighlightKind::READ, 41..47)
3072                ]
3073            )
3074        });
3075    }
3076
3077    #[gpui::test(iterations = 10)]
3078    async fn test_project_symbols(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3079        cx_a.foreground().forbid_parking();
3080        let mut lang_registry = Arc::new(LanguageRegistry::test());
3081        let fs = FakeFs::new(cx_a.background());
3082        fs.insert_tree(
3083            "/code",
3084            json!({
3085                "crate-1": {
3086                    ".zed.toml": r#"collaborators = ["user_b"]"#,
3087                    "one.rs": "const ONE: usize = 1;",
3088                },
3089                "crate-2": {
3090                    "two.rs": "const TWO: usize = 2; const THREE: usize = 3;",
3091                },
3092                "private": {
3093                    "passwords.txt": "the-password",
3094                }
3095            }),
3096        )
3097        .await;
3098
3099        // Set up a fake language server.
3100        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
3101        Arc::get_mut(&mut lang_registry)
3102            .unwrap()
3103            .add(Arc::new(Language::new(
3104                LanguageConfig {
3105                    name: "Rust".into(),
3106                    path_suffixes: vec!["rs".to_string()],
3107                    language_server: Some(language_server_config),
3108                    ..Default::default()
3109                },
3110                Some(tree_sitter_rust::language()),
3111            )));
3112
3113        // Connect to a server as 2 clients.
3114        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3115        let client_a = server.create_client(cx_a, "user_a").await;
3116        let client_b = server.create_client(cx_b, "user_b").await;
3117
3118        // Share a project as client A
3119        let project_a = cx_a.update(|cx| {
3120            Project::local(
3121                client_a.clone(),
3122                client_a.user_store.clone(),
3123                lang_registry.clone(),
3124                fs.clone(),
3125                cx,
3126            )
3127        });
3128        let (worktree_a, _) = project_a
3129            .update(cx_a, |p, cx| {
3130                p.find_or_create_local_worktree("/code/crate-1", true, cx)
3131            })
3132            .await
3133            .unwrap();
3134        worktree_a
3135            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3136            .await;
3137        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
3138        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3139        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
3140
3141        // Join the worktree as client B.
3142        let project_b = Project::remote(
3143            project_id,
3144            client_b.clone(),
3145            client_b.user_store.clone(),
3146            lang_registry.clone(),
3147            fs.clone(),
3148            &mut cx_b.to_async(),
3149        )
3150        .await
3151        .unwrap();
3152
3153        // Cause the language server to start.
3154        let _buffer = cx_b
3155            .background()
3156            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "one.rs"), cx)))
3157            .await
3158            .unwrap();
3159
3160        let mut fake_language_server = fake_language_servers.next().await.unwrap();
3161        fake_language_server.handle_request::<lsp::request::WorkspaceSymbol, _>(|_, _| {
3162            #[allow(deprecated)]
3163            Some(vec![lsp::SymbolInformation {
3164                name: "TWO".into(),
3165                location: lsp::Location {
3166                    uri: lsp::Url::from_file_path("/code/crate-2/two.rs").unwrap(),
3167                    range: lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
3168                },
3169                kind: lsp::SymbolKind::CONSTANT,
3170                tags: None,
3171                container_name: None,
3172                deprecated: None,
3173            }])
3174        });
3175
3176        // Request the definition of a symbol as the guest.
3177        let symbols = project_b
3178            .update(cx_b, |p, cx| p.symbols("two", cx))
3179            .await
3180            .unwrap();
3181        assert_eq!(symbols.len(), 1);
3182        assert_eq!(symbols[0].name, "TWO");
3183
3184        // Open one of the returned symbols.
3185        let buffer_b_2 = project_b
3186            .update(cx_b, |project, cx| {
3187                project.open_buffer_for_symbol(&symbols[0], cx)
3188            })
3189            .await
3190            .unwrap();
3191        buffer_b_2.read_with(cx_b, |buffer, _| {
3192            assert_eq!(
3193                buffer.file().unwrap().path().as_ref(),
3194                Path::new("../crate-2/two.rs")
3195            );
3196        });
3197
3198        // Attempt to craft a symbol and violate host's privacy by opening an arbitrary file.
3199        let mut fake_symbol = symbols[0].clone();
3200        fake_symbol.path = Path::new("/code/secrets").into();
3201        let error = project_b
3202            .update(cx_b, |project, cx| {
3203                project.open_buffer_for_symbol(&fake_symbol, cx)
3204            })
3205            .await
3206            .unwrap_err();
3207        assert!(error.to_string().contains("invalid symbol signature"));
3208    }
3209
3210    #[gpui::test(iterations = 10)]
3211    async fn test_open_buffer_while_getting_definition_pointing_to_it(
3212        cx_a: &mut TestAppContext,
3213        cx_b: &mut TestAppContext,
3214        mut rng: StdRng,
3215    ) {
3216        cx_a.foreground().forbid_parking();
3217        let mut lang_registry = Arc::new(LanguageRegistry::test());
3218        let fs = FakeFs::new(cx_a.background());
3219        fs.insert_tree(
3220            "/root",
3221            json!({
3222                ".zed.toml": r#"collaborators = ["user_b"]"#,
3223                "a.rs": "const ONE: usize = b::TWO;",
3224                "b.rs": "const TWO: usize = 2",
3225            }),
3226        )
3227        .await;
3228
3229        // Set up a fake language server.
3230        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
3231
3232        Arc::get_mut(&mut lang_registry)
3233            .unwrap()
3234            .add(Arc::new(Language::new(
3235                LanguageConfig {
3236                    name: "Rust".into(),
3237                    path_suffixes: vec!["rs".to_string()],
3238                    language_server: Some(language_server_config),
3239                    ..Default::default()
3240                },
3241                Some(tree_sitter_rust::language()),
3242            )));
3243
3244        // Connect to a server as 2 clients.
3245        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3246        let client_a = server.create_client(cx_a, "user_a").await;
3247        let client_b = server.create_client(cx_b, "user_b").await;
3248
3249        // Share a project as client A
3250        let project_a = cx_a.update(|cx| {
3251            Project::local(
3252                client_a.clone(),
3253                client_a.user_store.clone(),
3254                lang_registry.clone(),
3255                fs.clone(),
3256                cx,
3257            )
3258        });
3259
3260        let (worktree_a, _) = project_a
3261            .update(cx_a, |p, cx| {
3262                p.find_or_create_local_worktree("/root", true, cx)
3263            })
3264            .await
3265            .unwrap();
3266        worktree_a
3267            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3268            .await;
3269        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
3270        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3271        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
3272
3273        // Join the worktree as client B.
3274        let project_b = Project::remote(
3275            project_id,
3276            client_b.clone(),
3277            client_b.user_store.clone(),
3278            lang_registry.clone(),
3279            fs.clone(),
3280            &mut cx_b.to_async(),
3281        )
3282        .await
3283        .unwrap();
3284
3285        let buffer_b1 = cx_b
3286            .background()
3287            .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
3288            .await
3289            .unwrap();
3290
3291        let mut fake_language_server = fake_language_servers.next().await.unwrap();
3292        fake_language_server.handle_request::<lsp::request::GotoDefinition, _>(|_, _| {
3293            Some(lsp::GotoDefinitionResponse::Scalar(lsp::Location::new(
3294                lsp::Url::from_file_path("/root/b.rs").unwrap(),
3295                lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
3296            )))
3297        });
3298
3299        let definitions;
3300        let buffer_b2;
3301        if rng.gen() {
3302            definitions = project_b.update(cx_b, |p, cx| p.definition(&buffer_b1, 23, cx));
3303            buffer_b2 = project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "b.rs"), cx));
3304        } else {
3305            buffer_b2 = project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "b.rs"), cx));
3306            definitions = project_b.update(cx_b, |p, cx| p.definition(&buffer_b1, 23, cx));
3307        }
3308
3309        let buffer_b2 = buffer_b2.await.unwrap();
3310        let definitions = definitions.await.unwrap();
3311        assert_eq!(definitions.len(), 1);
3312        assert_eq!(definitions[0].buffer, buffer_b2);
3313    }
3314
3315    #[gpui::test(iterations = 10)]
3316    async fn test_collaborating_with_code_actions(
3317        cx_a: &mut TestAppContext,
3318        cx_b: &mut TestAppContext,
3319    ) {
3320        cx_a.foreground().forbid_parking();
3321        let mut lang_registry = Arc::new(LanguageRegistry::test());
3322        let fs = FakeFs::new(cx_a.background());
3323        cx_b.update(|cx| editor::init(cx));
3324
3325        // Set up a fake language server.
3326        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
3327        Arc::get_mut(&mut lang_registry)
3328            .unwrap()
3329            .add(Arc::new(Language::new(
3330                LanguageConfig {
3331                    name: "Rust".into(),
3332                    path_suffixes: vec!["rs".to_string()],
3333                    language_server: Some(language_server_config),
3334                    ..Default::default()
3335                },
3336                Some(tree_sitter_rust::language()),
3337            )));
3338
3339        // Connect to a server as 2 clients.
3340        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3341        let client_a = server.create_client(cx_a, "user_a").await;
3342        let client_b = server.create_client(cx_b, "user_b").await;
3343
3344        // Share a project as client A
3345        fs.insert_tree(
3346            "/a",
3347            json!({
3348                ".zed.toml": r#"collaborators = ["user_b"]"#,
3349                "main.rs": "mod other;\nfn main() { let foo = other::foo(); }",
3350                "other.rs": "pub fn foo() -> usize { 4 }",
3351            }),
3352        )
3353        .await;
3354        let project_a = cx_a.update(|cx| {
3355            Project::local(
3356                client_a.clone(),
3357                client_a.user_store.clone(),
3358                lang_registry.clone(),
3359                fs.clone(),
3360                cx,
3361            )
3362        });
3363        let (worktree_a, _) = project_a
3364            .update(cx_a, |p, cx| {
3365                p.find_or_create_local_worktree("/a", true, cx)
3366            })
3367            .await
3368            .unwrap();
3369        worktree_a
3370            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3371            .await;
3372        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
3373        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3374        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
3375
3376        // Join the worktree as client B.
3377        let project_b = Project::remote(
3378            project_id,
3379            client_b.clone(),
3380            client_b.user_store.clone(),
3381            lang_registry.clone(),
3382            fs.clone(),
3383            &mut cx_b.to_async(),
3384        )
3385        .await
3386        .unwrap();
3387        let mut params = cx_b.update(WorkspaceParams::test);
3388        params.languages = lang_registry.clone();
3389        params.client = client_b.client.clone();
3390        params.user_store = client_b.user_store.clone();
3391        params.project = project_b;
3392
3393        let (_window_b, workspace_b) = cx_b.add_window(|cx| Workspace::new(&params, cx));
3394        let editor_b = workspace_b
3395            .update(cx_b, |workspace, cx| {
3396                workspace.open_path((worktree_id, "main.rs"), cx)
3397            })
3398            .await
3399            .unwrap()
3400            .downcast::<Editor>()
3401            .unwrap();
3402
3403        let mut fake_language_server = fake_language_servers.next().await.unwrap();
3404        fake_language_server
3405            .handle_request::<lsp::request::CodeActionRequest, _>(|params, _| {
3406                assert_eq!(
3407                    params.text_document.uri,
3408                    lsp::Url::from_file_path("/a/main.rs").unwrap(),
3409                );
3410                assert_eq!(params.range.start, lsp::Position::new(0, 0));
3411                assert_eq!(params.range.end, lsp::Position::new(0, 0));
3412                None
3413            })
3414            .next()
3415            .await;
3416
3417        // Move cursor to a location that contains code actions.
3418        editor_b.update(cx_b, |editor, cx| {
3419            editor.select_ranges([Point::new(1, 31)..Point::new(1, 31)], None, cx);
3420            cx.focus(&editor_b);
3421        });
3422
3423        fake_language_server
3424            .handle_request::<lsp::request::CodeActionRequest, _>(|params, _| {
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(1, 31));
3430                assert_eq!(params.range.end, lsp::Position::new(1, 31));
3431
3432                Some(vec![lsp::CodeActionOrCommand::CodeAction(
3433                    lsp::CodeAction {
3434                        title: "Inline into all callers".to_string(),
3435                        edit: Some(lsp::WorkspaceEdit {
3436                            changes: Some(
3437                                [
3438                                    (
3439                                        lsp::Url::from_file_path("/a/main.rs").unwrap(),
3440                                        vec![lsp::TextEdit::new(
3441                                            lsp::Range::new(
3442                                                lsp::Position::new(1, 22),
3443                                                lsp::Position::new(1, 34),
3444                                            ),
3445                                            "4".to_string(),
3446                                        )],
3447                                    ),
3448                                    (
3449                                        lsp::Url::from_file_path("/a/other.rs").unwrap(),
3450                                        vec![lsp::TextEdit::new(
3451                                            lsp::Range::new(
3452                                                lsp::Position::new(0, 0),
3453                                                lsp::Position::new(0, 27),
3454                                            ),
3455                                            "".to_string(),
3456                                        )],
3457                                    ),
3458                                ]
3459                                .into_iter()
3460                                .collect(),
3461                            ),
3462                            ..Default::default()
3463                        }),
3464                        data: Some(json!({
3465                            "codeActionParams": {
3466                                "range": {
3467                                    "start": {"line": 1, "column": 31},
3468                                    "end": {"line": 1, "column": 31},
3469                                }
3470                            }
3471                        })),
3472                        ..Default::default()
3473                    },
3474                )])
3475            })
3476            .next()
3477            .await;
3478
3479        // Toggle code actions and wait for them to display.
3480        editor_b.update(cx_b, |editor, cx| {
3481            editor.toggle_code_actions(&ToggleCodeActions(false), cx);
3482        });
3483        editor_b
3484            .condition(&cx_b, |editor, _| editor.context_menu_visible())
3485            .await;
3486
3487        fake_language_server.remove_request_handler::<lsp::request::CodeActionRequest>();
3488
3489        // Confirming the code action will trigger a resolve request.
3490        let confirm_action = workspace_b
3491            .update(cx_b, |workspace, cx| {
3492                Editor::confirm_code_action(workspace, &ConfirmCodeAction(Some(0)), cx)
3493            })
3494            .unwrap();
3495        fake_language_server.handle_request::<lsp::request::CodeActionResolveRequest, _>(|_, _| {
3496            lsp::CodeAction {
3497                title: "Inline into all callers".to_string(),
3498                edit: Some(lsp::WorkspaceEdit {
3499                    changes: Some(
3500                        [
3501                            (
3502                                lsp::Url::from_file_path("/a/main.rs").unwrap(),
3503                                vec![lsp::TextEdit::new(
3504                                    lsp::Range::new(
3505                                        lsp::Position::new(1, 22),
3506                                        lsp::Position::new(1, 34),
3507                                    ),
3508                                    "4".to_string(),
3509                                )],
3510                            ),
3511                            (
3512                                lsp::Url::from_file_path("/a/other.rs").unwrap(),
3513                                vec![lsp::TextEdit::new(
3514                                    lsp::Range::new(
3515                                        lsp::Position::new(0, 0),
3516                                        lsp::Position::new(0, 27),
3517                                    ),
3518                                    "".to_string(),
3519                                )],
3520                            ),
3521                        ]
3522                        .into_iter()
3523                        .collect(),
3524                    ),
3525                    ..Default::default()
3526                }),
3527                ..Default::default()
3528            }
3529        });
3530
3531        // After the action is confirmed, an editor containing both modified files is opened.
3532        confirm_action.await.unwrap();
3533        let code_action_editor = workspace_b.read_with(cx_b, |workspace, cx| {
3534            workspace
3535                .active_item(cx)
3536                .unwrap()
3537                .downcast::<Editor>()
3538                .unwrap()
3539        });
3540        code_action_editor.update(cx_b, |editor, cx| {
3541            assert_eq!(editor.text(cx), "\nmod other;\nfn main() { let foo = 4; }");
3542            editor.undo(&Undo, cx);
3543            assert_eq!(
3544                editor.text(cx),
3545                "pub fn foo() -> usize { 4 }\nmod other;\nfn main() { let foo = other::foo(); }"
3546            );
3547            editor.redo(&Redo, cx);
3548            assert_eq!(editor.text(cx), "\nmod other;\nfn main() { let foo = 4; }");
3549        });
3550    }
3551
3552    #[gpui::test(iterations = 10)]
3553    async fn test_collaborating_with_renames(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3554        cx_a.foreground().forbid_parking();
3555        let mut lang_registry = Arc::new(LanguageRegistry::test());
3556        let fs = FakeFs::new(cx_a.background());
3557        cx_b.update(|cx| editor::init(cx));
3558
3559        // Set up a fake language server.
3560        let (language_server_config, mut fake_language_servers) = LanguageServerConfig::fake();
3561        Arc::get_mut(&mut lang_registry)
3562            .unwrap()
3563            .add(Arc::new(Language::new(
3564                LanguageConfig {
3565                    name: "Rust".into(),
3566                    path_suffixes: vec!["rs".to_string()],
3567                    language_server: Some(language_server_config),
3568                    ..Default::default()
3569                },
3570                Some(tree_sitter_rust::language()),
3571            )));
3572
3573        // Connect to a server as 2 clients.
3574        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3575        let client_a = server.create_client(cx_a, "user_a").await;
3576        let client_b = server.create_client(cx_b, "user_b").await;
3577
3578        // Share a project as client A
3579        fs.insert_tree(
3580            "/dir",
3581            json!({
3582                ".zed.toml": r#"collaborators = ["user_b"]"#,
3583                "one.rs": "const ONE: usize = 1;",
3584                "two.rs": "const TWO: usize = one::ONE + one::ONE;"
3585            }),
3586        )
3587        .await;
3588        let project_a = cx_a.update(|cx| {
3589            Project::local(
3590                client_a.clone(),
3591                client_a.user_store.clone(),
3592                lang_registry.clone(),
3593                fs.clone(),
3594                cx,
3595            )
3596        });
3597        let (worktree_a, _) = project_a
3598            .update(cx_a, |p, cx| {
3599                p.find_or_create_local_worktree("/dir", true, cx)
3600            })
3601            .await
3602            .unwrap();
3603        worktree_a
3604            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3605            .await;
3606        let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
3607        let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3608        project_a.update(cx_a, |p, cx| p.share(cx)).await.unwrap();
3609
3610        // Join the worktree as client B.
3611        let project_b = Project::remote(
3612            project_id,
3613            client_b.clone(),
3614            client_b.user_store.clone(),
3615            lang_registry.clone(),
3616            fs.clone(),
3617            &mut cx_b.to_async(),
3618        )
3619        .await
3620        .unwrap();
3621        let mut params = cx_b.update(WorkspaceParams::test);
3622        params.languages = lang_registry.clone();
3623        params.client = client_b.client.clone();
3624        params.user_store = client_b.user_store.clone();
3625        params.project = project_b;
3626
3627        let (_window_b, workspace_b) = cx_b.add_window(|cx| Workspace::new(&params, cx));
3628        let editor_b = workspace_b
3629            .update(cx_b, |workspace, cx| {
3630                workspace.open_path((worktree_id, "one.rs"), cx)
3631            })
3632            .await
3633            .unwrap()
3634            .downcast::<Editor>()
3635            .unwrap();
3636        let mut fake_language_server = fake_language_servers.next().await.unwrap();
3637
3638        // Move cursor to a location that can be renamed.
3639        let prepare_rename = editor_b.update(cx_b, |editor, cx| {
3640            editor.select_ranges([7..7], None, cx);
3641            editor.rename(&Rename, cx).unwrap()
3642        });
3643
3644        fake_language_server
3645            .handle_request::<lsp::request::PrepareRenameRequest, _>(|params, _| {
3646                assert_eq!(params.text_document.uri.as_str(), "file:///dir/one.rs");
3647                assert_eq!(params.position, lsp::Position::new(0, 7));
3648                Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
3649                    lsp::Position::new(0, 6),
3650                    lsp::Position::new(0, 9),
3651                )))
3652            })
3653            .next()
3654            .await
3655            .unwrap();
3656        prepare_rename.await.unwrap();
3657        editor_b.update(cx_b, |editor, cx| {
3658            let rename = editor.pending_rename().unwrap();
3659            let buffer = editor.buffer().read(cx).snapshot(cx);
3660            assert_eq!(
3661                rename.range.start.to_offset(&buffer)..rename.range.end.to_offset(&buffer),
3662                6..9
3663            );
3664            rename.editor.update(cx, |rename_editor, cx| {
3665                rename_editor.buffer().update(cx, |rename_buffer, cx| {
3666                    rename_buffer.edit([0..3], "THREE", cx);
3667                });
3668            });
3669        });
3670
3671        let confirm_rename = workspace_b.update(cx_b, |workspace, cx| {
3672            Editor::confirm_rename(workspace, &ConfirmRename, cx).unwrap()
3673        });
3674        fake_language_server
3675            .handle_request::<lsp::request::Rename, _>(|params, _| {
3676                assert_eq!(
3677                    params.text_document_position.text_document.uri.as_str(),
3678                    "file:///dir/one.rs"
3679                );
3680                assert_eq!(
3681                    params.text_document_position.position,
3682                    lsp::Position::new(0, 6)
3683                );
3684                assert_eq!(params.new_name, "THREE");
3685                Some(lsp::WorkspaceEdit {
3686                    changes: Some(
3687                        [
3688                            (
3689                                lsp::Url::from_file_path("/dir/one.rs").unwrap(),
3690                                vec![lsp::TextEdit::new(
3691                                    lsp::Range::new(
3692                                        lsp::Position::new(0, 6),
3693                                        lsp::Position::new(0, 9),
3694                                    ),
3695                                    "THREE".to_string(),
3696                                )],
3697                            ),
3698                            (
3699                                lsp::Url::from_file_path("/dir/two.rs").unwrap(),
3700                                vec![
3701                                    lsp::TextEdit::new(
3702                                        lsp::Range::new(
3703                                            lsp::Position::new(0, 24),
3704                                            lsp::Position::new(0, 27),
3705                                        ),
3706                                        "THREE".to_string(),
3707                                    ),
3708                                    lsp::TextEdit::new(
3709                                        lsp::Range::new(
3710                                            lsp::Position::new(0, 35),
3711                                            lsp::Position::new(0, 38),
3712                                        ),
3713                                        "THREE".to_string(),
3714                                    ),
3715                                ],
3716                            ),
3717                        ]
3718                        .into_iter()
3719                        .collect(),
3720                    ),
3721                    ..Default::default()
3722                })
3723            })
3724            .next()
3725            .await
3726            .unwrap();
3727        confirm_rename.await.unwrap();
3728
3729        let rename_editor = workspace_b.read_with(cx_b, |workspace, cx| {
3730            workspace
3731                .active_item(cx)
3732                .unwrap()
3733                .downcast::<Editor>()
3734                .unwrap()
3735        });
3736        rename_editor.update(cx_b, |editor, cx| {
3737            assert_eq!(
3738                editor.text(cx),
3739                "const TWO: usize = one::THREE + one::THREE;\nconst THREE: usize = 1;"
3740            );
3741            editor.undo(&Undo, cx);
3742            assert_eq!(
3743                editor.text(cx),
3744                "const TWO: usize = one::ONE + one::ONE;\nconst ONE: usize = 1;"
3745            );
3746            editor.redo(&Redo, cx);
3747            assert_eq!(
3748                editor.text(cx),
3749                "const TWO: usize = one::THREE + one::THREE;\nconst THREE: usize = 1;"
3750            );
3751        });
3752
3753        // Ensure temporary rename edits cannot be undone/redone.
3754        editor_b.update(cx_b, |editor, cx| {
3755            editor.undo(&Undo, cx);
3756            assert_eq!(editor.text(cx), "const ONE: usize = 1;");
3757            editor.undo(&Undo, cx);
3758            assert_eq!(editor.text(cx), "const ONE: usize = 1;");
3759            editor.redo(&Redo, cx);
3760            assert_eq!(editor.text(cx), "const THREE: usize = 1;");
3761        })
3762    }
3763
3764    #[gpui::test(iterations = 10)]
3765    async fn test_basic_chat(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3766        cx_a.foreground().forbid_parking();
3767
3768        // Connect to a server as 2 clients.
3769        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3770        let client_a = server.create_client(cx_a, "user_a").await;
3771        let client_b = server.create_client(cx_b, "user_b").await;
3772
3773        // Create an org that includes these 2 users.
3774        let db = &server.app_state.db;
3775        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
3776        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
3777            .await
3778            .unwrap();
3779        db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
3780            .await
3781            .unwrap();
3782
3783        // Create a channel that includes all the users.
3784        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
3785        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
3786            .await
3787            .unwrap();
3788        db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
3789            .await
3790            .unwrap();
3791        db.create_channel_message(
3792            channel_id,
3793            client_b.current_user_id(&cx_b),
3794            "hello A, it's B.",
3795            OffsetDateTime::now_utc(),
3796            1,
3797        )
3798        .await
3799        .unwrap();
3800
3801        let channels_a = cx_a
3802            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
3803        channels_a
3804            .condition(cx_a, |list, _| list.available_channels().is_some())
3805            .await;
3806        channels_a.read_with(cx_a, |list, _| {
3807            assert_eq!(
3808                list.available_channels().unwrap(),
3809                &[ChannelDetails {
3810                    id: channel_id.to_proto(),
3811                    name: "test-channel".to_string()
3812                }]
3813            )
3814        });
3815        let channel_a = channels_a.update(cx_a, |this, cx| {
3816            this.get_channel(channel_id.to_proto(), cx).unwrap()
3817        });
3818        channel_a.read_with(cx_a, |channel, _| assert!(channel.messages().is_empty()));
3819        channel_a
3820            .condition(&cx_a, |channel, _| {
3821                channel_messages(channel)
3822                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
3823            })
3824            .await;
3825
3826        let channels_b = cx_b
3827            .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
3828        channels_b
3829            .condition(cx_b, |list, _| list.available_channels().is_some())
3830            .await;
3831        channels_b.read_with(cx_b, |list, _| {
3832            assert_eq!(
3833                list.available_channels().unwrap(),
3834                &[ChannelDetails {
3835                    id: channel_id.to_proto(),
3836                    name: "test-channel".to_string()
3837                }]
3838            )
3839        });
3840
3841        let channel_b = channels_b.update(cx_b, |this, cx| {
3842            this.get_channel(channel_id.to_proto(), cx).unwrap()
3843        });
3844        channel_b.read_with(cx_b, |channel, _| assert!(channel.messages().is_empty()));
3845        channel_b
3846            .condition(&cx_b, |channel, _| {
3847                channel_messages(channel)
3848                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
3849            })
3850            .await;
3851
3852        channel_a
3853            .update(cx_a, |channel, cx| {
3854                channel
3855                    .send_message("oh, hi B.".to_string(), cx)
3856                    .unwrap()
3857                    .detach();
3858                let task = channel.send_message("sup".to_string(), cx).unwrap();
3859                assert_eq!(
3860                    channel_messages(channel),
3861                    &[
3862                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
3863                        ("user_a".to_string(), "oh, hi B.".to_string(), true),
3864                        ("user_a".to_string(), "sup".to_string(), true)
3865                    ]
3866                );
3867                task
3868            })
3869            .await
3870            .unwrap();
3871
3872        channel_b
3873            .condition(&cx_b, |channel, _| {
3874                channel_messages(channel)
3875                    == [
3876                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
3877                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
3878                        ("user_a".to_string(), "sup".to_string(), false),
3879                    ]
3880            })
3881            .await;
3882
3883        assert_eq!(
3884            server
3885                .state()
3886                .await
3887                .channel(channel_id)
3888                .unwrap()
3889                .connection_ids
3890                .len(),
3891            2
3892        );
3893        cx_b.update(|_| drop(channel_b));
3894        server
3895            .condition(|state| state.channel(channel_id).unwrap().connection_ids.len() == 1)
3896            .await;
3897
3898        cx_a.update(|_| drop(channel_a));
3899        server
3900            .condition(|state| state.channel(channel_id).is_none())
3901            .await;
3902    }
3903
3904    #[gpui::test(iterations = 10)]
3905    async fn test_chat_message_validation(cx_a: &mut TestAppContext) {
3906        cx_a.foreground().forbid_parking();
3907
3908        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3909        let client_a = server.create_client(cx_a, "user_a").await;
3910
3911        let db = &server.app_state.db;
3912        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
3913        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
3914        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
3915            .await
3916            .unwrap();
3917        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
3918            .await
3919            .unwrap();
3920
3921        let channels_a = cx_a
3922            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
3923        channels_a
3924            .condition(cx_a, |list, _| list.available_channels().is_some())
3925            .await;
3926        let channel_a = channels_a.update(cx_a, |this, cx| {
3927            this.get_channel(channel_id.to_proto(), cx).unwrap()
3928        });
3929
3930        // Messages aren't allowed to be too long.
3931        channel_a
3932            .update(cx_a, |channel, cx| {
3933                let long_body = "this is long.\n".repeat(1024);
3934                channel.send_message(long_body, cx).unwrap()
3935            })
3936            .await
3937            .unwrap_err();
3938
3939        // Messages aren't allowed to be blank.
3940        channel_a.update(cx_a, |channel, cx| {
3941            channel.send_message(String::new(), cx).unwrap_err()
3942        });
3943
3944        // Leading and trailing whitespace are trimmed.
3945        channel_a
3946            .update(cx_a, |channel, cx| {
3947                channel
3948                    .send_message("\n surrounded by whitespace  \n".to_string(), cx)
3949                    .unwrap()
3950            })
3951            .await
3952            .unwrap();
3953        assert_eq!(
3954            db.get_channel_messages(channel_id, 10, None)
3955                .await
3956                .unwrap()
3957                .iter()
3958                .map(|m| &m.body)
3959                .collect::<Vec<_>>(),
3960            &["surrounded by whitespace"]
3961        );
3962    }
3963
3964    #[gpui::test(iterations = 10)]
3965    async fn test_chat_reconnection(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3966        cx_a.foreground().forbid_parking();
3967
3968        // Connect to a server as 2 clients.
3969        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3970        let client_a = server.create_client(cx_a, "user_a").await;
3971        let client_b = server.create_client(cx_b, "user_b").await;
3972        let mut status_b = client_b.status();
3973
3974        // Create an org that includes these 2 users.
3975        let db = &server.app_state.db;
3976        let org_id = db.create_org("Test Org", "test-org").await.unwrap();
3977        db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
3978            .await
3979            .unwrap();
3980        db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
3981            .await
3982            .unwrap();
3983
3984        // Create a channel that includes all the users.
3985        let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
3986        db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
3987            .await
3988            .unwrap();
3989        db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
3990            .await
3991            .unwrap();
3992        db.create_channel_message(
3993            channel_id,
3994            client_b.current_user_id(&cx_b),
3995            "hello A, it's B.",
3996            OffsetDateTime::now_utc(),
3997            2,
3998        )
3999        .await
4000        .unwrap();
4001
4002        let channels_a = cx_a
4003            .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
4004        channels_a
4005            .condition(cx_a, |list, _| list.available_channels().is_some())
4006            .await;
4007
4008        channels_a.read_with(cx_a, |list, _| {
4009            assert_eq!(
4010                list.available_channels().unwrap(),
4011                &[ChannelDetails {
4012                    id: channel_id.to_proto(),
4013                    name: "test-channel".to_string()
4014                }]
4015            )
4016        });
4017        let channel_a = channels_a.update(cx_a, |this, cx| {
4018            this.get_channel(channel_id.to_proto(), cx).unwrap()
4019        });
4020        channel_a.read_with(cx_a, |channel, _| assert!(channel.messages().is_empty()));
4021        channel_a
4022            .condition(&cx_a, |channel, _| {
4023                channel_messages(channel)
4024                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4025            })
4026            .await;
4027
4028        let channels_b = cx_b
4029            .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
4030        channels_b
4031            .condition(cx_b, |list, _| list.available_channels().is_some())
4032            .await;
4033        channels_b.read_with(cx_b, |list, _| {
4034            assert_eq!(
4035                list.available_channels().unwrap(),
4036                &[ChannelDetails {
4037                    id: channel_id.to_proto(),
4038                    name: "test-channel".to_string()
4039                }]
4040            )
4041        });
4042
4043        let channel_b = channels_b.update(cx_b, |this, cx| {
4044            this.get_channel(channel_id.to_proto(), cx).unwrap()
4045        });
4046        channel_b.read_with(cx_b, |channel, _| assert!(channel.messages().is_empty()));
4047        channel_b
4048            .condition(&cx_b, |channel, _| {
4049                channel_messages(channel)
4050                    == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4051            })
4052            .await;
4053
4054        // Disconnect client B, ensuring we can still access its cached channel data.
4055        server.forbid_connections();
4056        server.disconnect_client(client_b.current_user_id(&cx_b));
4057        cx_b.foreground().advance_clock(Duration::from_secs(3));
4058        while !matches!(
4059            status_b.next().await,
4060            Some(client::Status::ReconnectionError { .. })
4061        ) {}
4062
4063        channels_b.read_with(cx_b, |channels, _| {
4064            assert_eq!(
4065                channels.available_channels().unwrap(),
4066                [ChannelDetails {
4067                    id: channel_id.to_proto(),
4068                    name: "test-channel".to_string()
4069                }]
4070            )
4071        });
4072        channel_b.read_with(cx_b, |channel, _| {
4073            assert_eq!(
4074                channel_messages(channel),
4075                [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4076            )
4077        });
4078
4079        // Send a message from client B while it is disconnected.
4080        channel_b
4081            .update(cx_b, |channel, cx| {
4082                let task = channel
4083                    .send_message("can you see this?".to_string(), cx)
4084                    .unwrap();
4085                assert_eq!(
4086                    channel_messages(channel),
4087                    &[
4088                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4089                        ("user_b".to_string(), "can you see this?".to_string(), true)
4090                    ]
4091                );
4092                task
4093            })
4094            .await
4095            .unwrap_err();
4096
4097        // Send a message from client A while B is disconnected.
4098        channel_a
4099            .update(cx_a, |channel, cx| {
4100                channel
4101                    .send_message("oh, hi B.".to_string(), cx)
4102                    .unwrap()
4103                    .detach();
4104                let task = channel.send_message("sup".to_string(), cx).unwrap();
4105                assert_eq!(
4106                    channel_messages(channel),
4107                    &[
4108                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4109                        ("user_a".to_string(), "oh, hi B.".to_string(), true),
4110                        ("user_a".to_string(), "sup".to_string(), true)
4111                    ]
4112                );
4113                task
4114            })
4115            .await
4116            .unwrap();
4117
4118        // Give client B a chance to reconnect.
4119        server.allow_connections();
4120        cx_b.foreground().advance_clock(Duration::from_secs(10));
4121
4122        // Verify that B sees the new messages upon reconnection, as well as the message client B
4123        // sent while offline.
4124        channel_b
4125            .condition(&cx_b, |channel, _| {
4126                channel_messages(channel)
4127                    == [
4128                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4129                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4130                        ("user_a".to_string(), "sup".to_string(), false),
4131                        ("user_b".to_string(), "can you see this?".to_string(), false),
4132                    ]
4133            })
4134            .await;
4135
4136        // Ensure client A and B can communicate normally after reconnection.
4137        channel_a
4138            .update(cx_a, |channel, cx| {
4139                channel.send_message("you online?".to_string(), cx).unwrap()
4140            })
4141            .await
4142            .unwrap();
4143        channel_b
4144            .condition(&cx_b, |channel, _| {
4145                channel_messages(channel)
4146                    == [
4147                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4148                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4149                        ("user_a".to_string(), "sup".to_string(), false),
4150                        ("user_b".to_string(), "can you see this?".to_string(), false),
4151                        ("user_a".to_string(), "you online?".to_string(), false),
4152                    ]
4153            })
4154            .await;
4155
4156        channel_b
4157            .update(cx_b, |channel, cx| {
4158                channel.send_message("yep".to_string(), cx).unwrap()
4159            })
4160            .await
4161            .unwrap();
4162        channel_a
4163            .condition(&cx_a, |channel, _| {
4164                channel_messages(channel)
4165                    == [
4166                        ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4167                        ("user_a".to_string(), "oh, hi B.".to_string(), false),
4168                        ("user_a".to_string(), "sup".to_string(), false),
4169                        ("user_b".to_string(), "can you see this?".to_string(), false),
4170                        ("user_a".to_string(), "you online?".to_string(), false),
4171                        ("user_b".to_string(), "yep".to_string(), false),
4172                    ]
4173            })
4174            .await;
4175    }
4176
4177    #[gpui::test(iterations = 10)]
4178    async fn test_contacts(
4179        cx_a: &mut TestAppContext,
4180        cx_b: &mut TestAppContext,
4181        cx_c: &mut TestAppContext,
4182    ) {
4183        cx_a.foreground().forbid_parking();
4184        let lang_registry = Arc::new(LanguageRegistry::test());
4185        let fs = FakeFs::new(cx_a.background());
4186
4187        // Connect to a server as 3 clients.
4188        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4189        let client_a = server.create_client(cx_a, "user_a").await;
4190        let client_b = server.create_client(cx_b, "user_b").await;
4191        let client_c = server.create_client(cx_c, "user_c").await;
4192
4193        // Share a worktree as client A.
4194        fs.insert_tree(
4195            "/a",
4196            json!({
4197                ".zed.toml": r#"collaborators = ["user_b", "user_c"]"#,
4198            }),
4199        )
4200        .await;
4201
4202        let project_a = cx_a.update(|cx| {
4203            Project::local(
4204                client_a.clone(),
4205                client_a.user_store.clone(),
4206                lang_registry.clone(),
4207                fs.clone(),
4208                cx,
4209            )
4210        });
4211        let (worktree_a, _) = project_a
4212            .update(cx_a, |p, cx| {
4213                p.find_or_create_local_worktree("/a", true, cx)
4214            })
4215            .await
4216            .unwrap();
4217        worktree_a
4218            .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
4219            .await;
4220
4221        client_a
4222            .user_store
4223            .condition(&cx_a, |user_store, _| {
4224                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4225            })
4226            .await;
4227        client_b
4228            .user_store
4229            .condition(&cx_b, |user_store, _| {
4230                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4231            })
4232            .await;
4233        client_c
4234            .user_store
4235            .condition(&cx_c, |user_store, _| {
4236                contacts(user_store) == vec![("user_a", vec![("a", vec![])])]
4237            })
4238            .await;
4239
4240        let project_id = project_a
4241            .update(cx_a, |project, _| project.next_remote_id())
4242            .await;
4243        project_a
4244            .update(cx_a, |project, cx| project.share(cx))
4245            .await
4246            .unwrap();
4247
4248        let _project_b = Project::remote(
4249            project_id,
4250            client_b.clone(),
4251            client_b.user_store.clone(),
4252            lang_registry.clone(),
4253            fs.clone(),
4254            &mut cx_b.to_async(),
4255        )
4256        .await
4257        .unwrap();
4258
4259        client_a
4260            .user_store
4261            .condition(&cx_a, |user_store, _| {
4262                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4263            })
4264            .await;
4265        client_b
4266            .user_store
4267            .condition(&cx_b, |user_store, _| {
4268                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4269            })
4270            .await;
4271        client_c
4272            .user_store
4273            .condition(&cx_c, |user_store, _| {
4274                contacts(user_store) == vec![("user_a", vec![("a", vec!["user_b"])])]
4275            })
4276            .await;
4277
4278        project_a
4279            .condition(&cx_a, |project, _| {
4280                project.collaborators().contains_key(&client_b.peer_id)
4281            })
4282            .await;
4283
4284        cx_a.update(move |_| drop(project_a));
4285        client_a
4286            .user_store
4287            .condition(&cx_a, |user_store, _| contacts(user_store) == vec![])
4288            .await;
4289        client_b
4290            .user_store
4291            .condition(&cx_b, |user_store, _| contacts(user_store) == vec![])
4292            .await;
4293        client_c
4294            .user_store
4295            .condition(&cx_c, |user_store, _| contacts(user_store) == vec![])
4296            .await;
4297
4298        fn contacts(user_store: &UserStore) -> Vec<(&str, Vec<(&str, Vec<&str>)>)> {
4299            user_store
4300                .contacts()
4301                .iter()
4302                .map(|contact| {
4303                    let worktrees = contact
4304                        .projects
4305                        .iter()
4306                        .map(|p| {
4307                            (
4308                                p.worktree_root_names[0].as_str(),
4309                                p.guests.iter().map(|p| p.github_login.as_str()).collect(),
4310                            )
4311                        })
4312                        .collect();
4313                    (contact.user.github_login.as_str(), worktrees)
4314                })
4315                .collect()
4316        }
4317    }
4318
4319    #[gpui::test(iterations = 10)]
4320    async fn test_following(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4321        cx_a.foreground().forbid_parking();
4322        let fs = FakeFs::new(cx_a.background());
4323
4324        // 2 clients connect to a server.
4325        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4326        let mut client_a = server.create_client(cx_a, "user_a").await;
4327        let mut client_b = server.create_client(cx_b, "user_b").await;
4328        cx_a.update(editor::init);
4329        cx_b.update(editor::init);
4330
4331        // Client A shares a project.
4332        fs.insert_tree(
4333            "/a",
4334            json!({
4335                ".zed.toml": r#"collaborators = ["user_b"]"#,
4336                "1.txt": "one",
4337                "2.txt": "two",
4338                "3.txt": "three",
4339            }),
4340        )
4341        .await;
4342        let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
4343        project_a
4344            .update(cx_a, |project, cx| project.share(cx))
4345            .await
4346            .unwrap();
4347
4348        // Client B joins the project.
4349        let project_b = client_b
4350            .build_remote_project(
4351                project_a
4352                    .read_with(cx_a, |project, _| project.remote_id())
4353                    .unwrap(),
4354                cx_b,
4355            )
4356            .await;
4357
4358        // Client A opens some editors.
4359        let workspace_a = client_a.build_workspace(&project_a, cx_a);
4360        let pane_a = workspace_a.read_with(cx_a, |workspace, _| workspace.active_pane().clone());
4361        let editor_a1 = workspace_a
4362            .update(cx_a, |workspace, cx| {
4363                workspace.open_path((worktree_id, "1.txt"), cx)
4364            })
4365            .await
4366            .unwrap()
4367            .downcast::<Editor>()
4368            .unwrap();
4369        let editor_a2 = workspace_a
4370            .update(cx_a, |workspace, cx| {
4371                workspace.open_path((worktree_id, "2.txt"), cx)
4372            })
4373            .await
4374            .unwrap()
4375            .downcast::<Editor>()
4376            .unwrap();
4377
4378        // Client B opens an editor.
4379        let workspace_b = client_b.build_workspace(&project_b, cx_b);
4380        let editor_b1 = workspace_b
4381            .update(cx_b, |workspace, cx| {
4382                workspace.open_path((worktree_id, "1.txt"), cx)
4383            })
4384            .await
4385            .unwrap()
4386            .downcast::<Editor>()
4387            .unwrap();
4388
4389        let client_a_id = project_b.read_with(cx_b, |project, _| {
4390            project.collaborators().values().next().unwrap().peer_id
4391        });
4392        let client_b_id = project_a.read_with(cx_a, |project, _| {
4393            project.collaborators().values().next().unwrap().peer_id
4394        });
4395
4396        // When client B starts following client A, all visible view states are replicated to client B.
4397        editor_a1.update(cx_a, |editor, cx| editor.select_ranges([0..1], None, cx));
4398        editor_a2.update(cx_a, |editor, cx| editor.select_ranges([2..3], None, cx));
4399        workspace_b
4400            .update(cx_b, |workspace, cx| {
4401                workspace.toggle_follow(&client_a_id.into(), cx).unwrap()
4402            })
4403            .await
4404            .unwrap();
4405        let editor_b2 = workspace_b.read_with(cx_b, |workspace, cx| {
4406            workspace
4407                .active_item(cx)
4408                .unwrap()
4409                .downcast::<Editor>()
4410                .unwrap()
4411        });
4412        assert!(cx_b.read(|cx| editor_b2.is_focused(cx)));
4413        assert_eq!(
4414            editor_b2.read_with(cx_b, |editor, cx| editor.project_path(cx)),
4415            Some((worktree_id, "2.txt").into())
4416        );
4417        assert_eq!(
4418            editor_b2.read_with(cx_b, |editor, cx| editor.selected_ranges(cx)),
4419            vec![2..3]
4420        );
4421        assert_eq!(
4422            editor_b1.read_with(cx_b, |editor, cx| editor.selected_ranges(cx)),
4423            vec![0..1]
4424        );
4425
4426        // When client A activates a different editor, client B does so as well.
4427        workspace_a.update(cx_a, |workspace, cx| {
4428            workspace.activate_item(&editor_a1, cx)
4429        });
4430        workspace_b
4431            .condition(cx_b, |workspace, cx| {
4432                workspace.active_item(cx).unwrap().id() == editor_b1.id()
4433            })
4434            .await;
4435
4436        // Changes to client A's editor are reflected on client B.
4437        editor_a1.update(cx_a, |editor, cx| {
4438            editor.select_ranges([1..1, 2..2], None, cx);
4439        });
4440        editor_b1
4441            .condition(cx_b, |editor, cx| {
4442                editor.selected_ranges(cx) == vec![1..1, 2..2]
4443            })
4444            .await;
4445
4446        editor_a1.update(cx_a, |editor, cx| editor.set_text("TWO", cx));
4447        editor_b1
4448            .condition(cx_b, |editor, cx| editor.text(cx) == "TWO")
4449            .await;
4450
4451        editor_a1.update(cx_a, |editor, cx| {
4452            editor.select_ranges([3..3], None, cx);
4453            editor.set_scroll_position(vec2f(0., 100.), cx);
4454        });
4455        editor_b1
4456            .condition(cx_b, |editor, cx| editor.selected_ranges(cx) == vec![3..3])
4457            .await;
4458
4459        // After unfollowing, client B stops receiving updates from client A.
4460        workspace_b.update(cx_b, |workspace, cx| {
4461            workspace.unfollow(&workspace.active_pane().clone(), cx)
4462        });
4463        workspace_a.update(cx_a, |workspace, cx| {
4464            workspace.activate_item(&editor_a2, cx)
4465        });
4466        cx_a.foreground().run_until_parked();
4467        assert_eq!(
4468            workspace_b.read_with(cx_b, |workspace, cx| workspace
4469                .active_item(cx)
4470                .unwrap()
4471                .id()),
4472            editor_b1.id()
4473        );
4474
4475        // Client A starts following client B.
4476        workspace_a
4477            .update(cx_a, |workspace, cx| {
4478                workspace.toggle_follow(&client_b_id.into(), cx).unwrap()
4479            })
4480            .await
4481            .unwrap();
4482        assert_eq!(
4483            workspace_a.read_with(cx_a, |workspace, _| workspace.leader_for_pane(&pane_a)),
4484            Some(client_b_id)
4485        );
4486        assert_eq!(
4487            workspace_a.read_with(cx_a, |workspace, cx| workspace
4488                .active_item(cx)
4489                .unwrap()
4490                .id()),
4491            editor_a1.id()
4492        );
4493
4494        // Following interrupts when client B disconnects.
4495        client_b.disconnect(&cx_b.to_async()).unwrap();
4496        cx_a.foreground().run_until_parked();
4497        assert_eq!(
4498            workspace_a.read_with(cx_a, |workspace, _| workspace.leader_for_pane(&pane_a)),
4499            None
4500        );
4501    }
4502
4503    #[gpui::test(iterations = 10)]
4504    async fn test_peers_following_each_other(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4505        cx_a.foreground().forbid_parking();
4506        let fs = FakeFs::new(cx_a.background());
4507
4508        // 2 clients connect to a server.
4509        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4510        let mut client_a = server.create_client(cx_a, "user_a").await;
4511        let mut client_b = server.create_client(cx_b, "user_b").await;
4512        cx_a.update(editor::init);
4513        cx_b.update(editor::init);
4514
4515        // Client A shares a project.
4516        fs.insert_tree(
4517            "/a",
4518            json!({
4519                ".zed.toml": r#"collaborators = ["user_b"]"#,
4520                "1.txt": "one",
4521                "2.txt": "two",
4522                "3.txt": "three",
4523                "4.txt": "four",
4524            }),
4525        )
4526        .await;
4527        let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
4528        project_a
4529            .update(cx_a, |project, cx| project.share(cx))
4530            .await
4531            .unwrap();
4532
4533        // Client B joins the project.
4534        let project_b = client_b
4535            .build_remote_project(
4536                project_a
4537                    .read_with(cx_a, |project, _| project.remote_id())
4538                    .unwrap(),
4539                cx_b,
4540            )
4541            .await;
4542
4543        // Client A opens some editors.
4544        let workspace_a = client_a.build_workspace(&project_a, cx_a);
4545        let pane_a1 = workspace_a.read_with(cx_a, |workspace, _| workspace.active_pane().clone());
4546        let _editor_a1 = workspace_a
4547            .update(cx_a, |workspace, cx| {
4548                workspace.open_path((worktree_id, "1.txt"), cx)
4549            })
4550            .await
4551            .unwrap()
4552            .downcast::<Editor>()
4553            .unwrap();
4554
4555        // Client B opens an editor.
4556        let workspace_b = client_b.build_workspace(&project_b, cx_b);
4557        let pane_b1 = workspace_b.read_with(cx_b, |workspace, _| workspace.active_pane().clone());
4558        let _editor_b1 = workspace_b
4559            .update(cx_b, |workspace, cx| {
4560                workspace.open_path((worktree_id, "2.txt"), cx)
4561            })
4562            .await
4563            .unwrap()
4564            .downcast::<Editor>()
4565            .unwrap();
4566
4567        // Clients A and B follow each other in split panes
4568        workspace_a
4569            .update(cx_a, |workspace, cx| {
4570                workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
4571                assert_ne!(*workspace.active_pane(), pane_a1);
4572                let leader_id = *project_a.read(cx).collaborators().keys().next().unwrap();
4573                workspace
4574                    .toggle_follow(&workspace::ToggleFollow(leader_id), cx)
4575                    .unwrap()
4576            })
4577            .await
4578            .unwrap();
4579        workspace_b
4580            .update(cx_b, |workspace, cx| {
4581                workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
4582                assert_ne!(*workspace.active_pane(), pane_b1);
4583                let leader_id = *project_b.read(cx).collaborators().keys().next().unwrap();
4584                workspace
4585                    .toggle_follow(&workspace::ToggleFollow(leader_id), cx)
4586                    .unwrap()
4587            })
4588            .await
4589            .unwrap();
4590
4591        workspace_a
4592            .update(cx_a, |workspace, cx| {
4593                workspace.activate_next_pane(cx);
4594                assert_eq!(*workspace.active_pane(), pane_a1);
4595                workspace.open_path((worktree_id, "3.txt"), cx)
4596            })
4597            .await
4598            .unwrap();
4599        workspace_b
4600            .update(cx_b, |workspace, cx| {
4601                workspace.activate_next_pane(cx);
4602                assert_eq!(*workspace.active_pane(), pane_b1);
4603                workspace.open_path((worktree_id, "4.txt"), cx)
4604            })
4605            .await
4606            .unwrap();
4607        cx_a.foreground().run_until_parked();
4608
4609        // Ensure leader updates don't change the active pane of followers
4610        workspace_a.read_with(cx_a, |workspace, _| {
4611            assert_eq!(*workspace.active_pane(), pane_a1);
4612        });
4613        workspace_b.read_with(cx_b, |workspace, _| {
4614            assert_eq!(*workspace.active_pane(), pane_b1);
4615        });
4616
4617        // Ensure peers following each other doesn't cause an infinite loop.
4618        assert_eq!(
4619            workspace_a.read_with(cx_a, |workspace, cx| workspace
4620                .active_item(cx)
4621                .unwrap()
4622                .project_path(cx)),
4623            Some((worktree_id, "3.txt").into())
4624        );
4625        workspace_a.update(cx_a, |workspace, cx| {
4626            assert_eq!(
4627                workspace.active_item(cx).unwrap().project_path(cx),
4628                Some((worktree_id, "3.txt").into())
4629            );
4630            workspace.activate_next_pane(cx);
4631            assert_eq!(
4632                workspace.active_item(cx).unwrap().project_path(cx),
4633                Some((worktree_id, "4.txt").into())
4634            );
4635        });
4636        workspace_b.update(cx_b, |workspace, cx| {
4637            assert_eq!(
4638                workspace.active_item(cx).unwrap().project_path(cx),
4639                Some((worktree_id, "4.txt").into())
4640            );
4641            workspace.activate_next_pane(cx);
4642            assert_eq!(
4643                workspace.active_item(cx).unwrap().project_path(cx),
4644                Some((worktree_id, "3.txt").into())
4645            );
4646        });
4647    }
4648
4649    #[gpui::test(iterations = 10)]
4650    async fn test_auto_unfollowing(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4651        cx_a.foreground().forbid_parking();
4652        let fs = FakeFs::new(cx_a.background());
4653
4654        // 2 clients connect to a server.
4655        let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4656        let mut client_a = server.create_client(cx_a, "user_a").await;
4657        let mut client_b = server.create_client(cx_b, "user_b").await;
4658        cx_a.update(editor::init);
4659        cx_b.update(editor::init);
4660
4661        // Client A shares a project.
4662        fs.insert_tree(
4663            "/a",
4664            json!({
4665                ".zed.toml": r#"collaborators = ["user_b"]"#,
4666                "1.txt": "one",
4667                "2.txt": "two",
4668                "3.txt": "three",
4669            }),
4670        )
4671        .await;
4672        let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
4673        project_a
4674            .update(cx_a, |project, cx| project.share(cx))
4675            .await
4676            .unwrap();
4677
4678        // Client B joins the project.
4679        let project_b = client_b
4680            .build_remote_project(
4681                project_a
4682                    .read_with(cx_a, |project, _| project.remote_id())
4683                    .unwrap(),
4684                cx_b,
4685            )
4686            .await;
4687
4688        // Client A opens some editors.
4689        let workspace_a = client_a.build_workspace(&project_a, cx_a);
4690        let _editor_a1 = workspace_a
4691            .update(cx_a, |workspace, cx| {
4692                workspace.open_path((worktree_id, "1.txt"), cx)
4693            })
4694            .await
4695            .unwrap()
4696            .downcast::<Editor>()
4697            .unwrap();
4698
4699        // Client B starts following client A.
4700        let workspace_b = client_b.build_workspace(&project_b, cx_b);
4701        let pane_b = workspace_b.read_with(cx_b, |workspace, _| workspace.active_pane().clone());
4702        let leader_id = project_b.read_with(cx_b, |project, _| {
4703            project.collaborators().values().next().unwrap().peer_id
4704        });
4705        workspace_b
4706            .update(cx_b, |workspace, cx| {
4707                workspace.toggle_follow(&leader_id.into(), cx).unwrap()
4708            })
4709            .await
4710            .unwrap();
4711        assert_eq!(
4712            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4713            Some(leader_id)
4714        );
4715        let editor_b2 = workspace_b.read_with(cx_b, |workspace, cx| {
4716            workspace
4717                .active_item(cx)
4718                .unwrap()
4719                .downcast::<Editor>()
4720                .unwrap()
4721        });
4722
4723        // When client B moves, it automatically stops following client A.
4724        editor_b2.update(cx_b, |editor, cx| editor.move_right(&editor::MoveRight, cx));
4725        assert_eq!(
4726            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4727            None
4728        );
4729
4730        workspace_b
4731            .update(cx_b, |workspace, cx| {
4732                workspace.toggle_follow(&leader_id.into(), cx).unwrap()
4733            })
4734            .await
4735            .unwrap();
4736        assert_eq!(
4737            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4738            Some(leader_id)
4739        );
4740
4741        // When client B edits, it automatically stops following client A.
4742        editor_b2.update(cx_b, |editor, cx| editor.insert("X", cx));
4743        assert_eq!(
4744            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4745            None
4746        );
4747
4748        workspace_b
4749            .update(cx_b, |workspace, cx| {
4750                workspace.toggle_follow(&leader_id.into(), cx).unwrap()
4751            })
4752            .await
4753            .unwrap();
4754        assert_eq!(
4755            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4756            Some(leader_id)
4757        );
4758
4759        // When client B scrolls, it automatically stops following client A.
4760        editor_b2.update(cx_b, |editor, cx| {
4761            editor.set_scroll_position(vec2f(0., 3.), cx)
4762        });
4763        assert_eq!(
4764            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4765            None
4766        );
4767
4768        workspace_b
4769            .update(cx_b, |workspace, cx| {
4770                workspace.toggle_follow(&leader_id.into(), cx).unwrap()
4771            })
4772            .await
4773            .unwrap();
4774        assert_eq!(
4775            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4776            Some(leader_id)
4777        );
4778
4779        // When client B activates a different pane, it continues following client A in the original pane.
4780        workspace_b.update(cx_b, |workspace, cx| {
4781            workspace.split_pane(pane_b.clone(), SplitDirection::Right, cx)
4782        });
4783        assert_eq!(
4784            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4785            Some(leader_id)
4786        );
4787
4788        workspace_b.update(cx_b, |workspace, cx| workspace.activate_next_pane(cx));
4789        assert_eq!(
4790            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4791            Some(leader_id)
4792        );
4793
4794        // When client B activates a different item in the original pane, it automatically stops following client A.
4795        workspace_b
4796            .update(cx_b, |workspace, cx| {
4797                workspace.open_path((worktree_id, "2.txt"), cx)
4798            })
4799            .await
4800            .unwrap();
4801        assert_eq!(
4802            workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
4803            None
4804        );
4805    }
4806
4807    #[gpui::test(iterations = 100)]
4808    async fn test_random_collaboration(cx: &mut TestAppContext, rng: StdRng) {
4809        cx.foreground().forbid_parking();
4810        let max_peers = env::var("MAX_PEERS")
4811            .map(|i| i.parse().expect("invalid `MAX_PEERS` variable"))
4812            .unwrap_or(5);
4813        let max_operations = env::var("OPERATIONS")
4814            .map(|i| i.parse().expect("invalid `OPERATIONS` variable"))
4815            .unwrap_or(10);
4816
4817        let rng = Arc::new(Mutex::new(rng));
4818
4819        let guest_lang_registry = Arc::new(LanguageRegistry::test());
4820        let (language_server_config, _fake_language_servers) = LanguageServerConfig::fake();
4821
4822        let fs = FakeFs::new(cx.background());
4823        fs.insert_tree(
4824            "/_collab",
4825            json!({
4826                ".zed.toml": r#"collaborators = ["guest-1", "guest-2", "guest-3", "guest-4", "guest-5"]"#
4827            }),
4828        )
4829        .await;
4830
4831        let operations = Rc::new(Cell::new(0));
4832        let mut server = TestServer::start(cx.foreground(), cx.background()).await;
4833        let mut clients = Vec::new();
4834
4835        let mut next_entity_id = 100000;
4836        let mut host_cx = TestAppContext::new(
4837            cx.foreground_platform(),
4838            cx.platform(),
4839            cx.foreground(),
4840            cx.background(),
4841            cx.font_cache(),
4842            cx.leak_detector(),
4843            next_entity_id,
4844        );
4845        let host = server.create_client(&mut host_cx, "host").await;
4846        let host_project = host_cx.update(|cx| {
4847            Project::local(
4848                host.client.clone(),
4849                host.user_store.clone(),
4850                Arc::new(LanguageRegistry::test()),
4851                fs.clone(),
4852                cx,
4853            )
4854        });
4855        let host_project_id = host_project
4856            .update(&mut host_cx, |p, _| p.next_remote_id())
4857            .await;
4858
4859        let (collab_worktree, _) = host_project
4860            .update(&mut host_cx, |project, cx| {
4861                project.find_or_create_local_worktree("/_collab", true, cx)
4862            })
4863            .await
4864            .unwrap();
4865        collab_worktree
4866            .read_with(&host_cx, |tree, _| tree.as_local().unwrap().scan_complete())
4867            .await;
4868        host_project
4869            .update(&mut host_cx, |project, cx| project.share(cx))
4870            .await
4871            .unwrap();
4872
4873        clients.push(cx.foreground().spawn(host.simulate_host(
4874            host_project,
4875            language_server_config,
4876            operations.clone(),
4877            max_operations,
4878            rng.clone(),
4879            host_cx,
4880        )));
4881
4882        while operations.get() < max_operations {
4883            cx.background().simulate_random_delay().await;
4884            if clients.len() >= max_peers {
4885                break;
4886            } else if rng.lock().gen_bool(0.05) {
4887                operations.set(operations.get() + 1);
4888
4889                let guest_id = clients.len();
4890                log::info!("Adding guest {}", guest_id);
4891                next_entity_id += 100000;
4892                let mut guest_cx = TestAppContext::new(
4893                    cx.foreground_platform(),
4894                    cx.platform(),
4895                    cx.foreground(),
4896                    cx.background(),
4897                    cx.font_cache(),
4898                    cx.leak_detector(),
4899                    next_entity_id,
4900                );
4901                let guest = server
4902                    .create_client(&mut guest_cx, &format!("guest-{}", guest_id))
4903                    .await;
4904                let guest_project = Project::remote(
4905                    host_project_id,
4906                    guest.client.clone(),
4907                    guest.user_store.clone(),
4908                    guest_lang_registry.clone(),
4909                    FakeFs::new(cx.background()),
4910                    &mut guest_cx.to_async(),
4911                )
4912                .await
4913                .unwrap();
4914                clients.push(cx.foreground().spawn(guest.simulate_guest(
4915                    guest_id,
4916                    guest_project,
4917                    operations.clone(),
4918                    max_operations,
4919                    rng.clone(),
4920                    guest_cx,
4921                )));
4922
4923                log::info!("Guest {} added", guest_id);
4924            }
4925        }
4926
4927        let mut clients = futures::future::join_all(clients).await;
4928        cx.foreground().run_until_parked();
4929
4930        let (host_client, mut host_cx) = clients.remove(0);
4931        let host_project = host_client.project.as_ref().unwrap();
4932        let host_worktree_snapshots = host_project.read_with(&host_cx, |project, cx| {
4933            project
4934                .worktrees(cx)
4935                .map(|worktree| {
4936                    let snapshot = worktree.read(cx).snapshot();
4937                    (snapshot.id(), snapshot)
4938                })
4939                .collect::<BTreeMap<_, _>>()
4940        });
4941
4942        host_client
4943            .project
4944            .as_ref()
4945            .unwrap()
4946            .read_with(&host_cx, |project, cx| project.check_invariants(cx));
4947
4948        for (guest_client, mut guest_cx) in clients.into_iter() {
4949            let guest_id = guest_client.client.id();
4950            let worktree_snapshots =
4951                guest_client
4952                    .project
4953                    .as_ref()
4954                    .unwrap()
4955                    .read_with(&guest_cx, |project, cx| {
4956                        project
4957                            .worktrees(cx)
4958                            .map(|worktree| {
4959                                let worktree = worktree.read(cx);
4960                                (worktree.id(), worktree.snapshot())
4961                            })
4962                            .collect::<BTreeMap<_, _>>()
4963                    });
4964
4965            assert_eq!(
4966                worktree_snapshots.keys().collect::<Vec<_>>(),
4967                host_worktree_snapshots.keys().collect::<Vec<_>>(),
4968                "guest {} has different worktrees than the host",
4969                guest_id
4970            );
4971            for (id, host_snapshot) in &host_worktree_snapshots {
4972                let guest_snapshot = &worktree_snapshots[id];
4973                assert_eq!(
4974                    guest_snapshot.root_name(),
4975                    host_snapshot.root_name(),
4976                    "guest {} has different root name than the host for worktree {}",
4977                    guest_id,
4978                    id
4979                );
4980                assert_eq!(
4981                    guest_snapshot.entries(false).collect::<Vec<_>>(),
4982                    host_snapshot.entries(false).collect::<Vec<_>>(),
4983                    "guest {} has different snapshot than the host for worktree {}",
4984                    guest_id,
4985                    id
4986                );
4987            }
4988
4989            guest_client
4990                .project
4991                .as_ref()
4992                .unwrap()
4993                .read_with(&guest_cx, |project, cx| project.check_invariants(cx));
4994
4995            for guest_buffer in &guest_client.buffers {
4996                let buffer_id = guest_buffer.read_with(&guest_cx, |buffer, _| buffer.remote_id());
4997                let host_buffer = host_project.read_with(&host_cx, |project, cx| {
4998                    project.buffer_for_id(buffer_id, cx).expect(&format!(
4999                        "host does not have buffer for guest:{}, peer:{}, id:{}",
5000                        guest_id, guest_client.peer_id, buffer_id
5001                    ))
5002                });
5003                let path = host_buffer
5004                    .read_with(&host_cx, |buffer, cx| buffer.file().unwrap().full_path(cx));
5005
5006                assert_eq!(
5007                    guest_buffer.read_with(&guest_cx, |buffer, _| buffer.deferred_ops_len()),
5008                    0,
5009                    "guest {}, buffer {}, path {:?} has deferred operations",
5010                    guest_id,
5011                    buffer_id,
5012                    path,
5013                );
5014                assert_eq!(
5015                    guest_buffer.read_with(&guest_cx, |buffer, _| buffer.text()),
5016                    host_buffer.read_with(&host_cx, |buffer, _| buffer.text()),
5017                    "guest {}, buffer {}, path {:?}, differs from the host's buffer",
5018                    guest_id,
5019                    buffer_id,
5020                    path
5021                );
5022            }
5023
5024            guest_cx.update(|_| drop(guest_client));
5025        }
5026
5027        host_cx.update(|_| drop(host_client));
5028    }
5029
5030    struct TestServer {
5031        peer: Arc<Peer>,
5032        app_state: Arc<AppState>,
5033        server: Arc<Server>,
5034        foreground: Rc<executor::Foreground>,
5035        notifications: mpsc::UnboundedReceiver<()>,
5036        connection_killers: Arc<Mutex<HashMap<UserId, barrier::Sender>>>,
5037        forbid_connections: Arc<AtomicBool>,
5038        _test_db: TestDb,
5039    }
5040
5041    impl TestServer {
5042        async fn start(
5043            foreground: Rc<executor::Foreground>,
5044            background: Arc<executor::Background>,
5045        ) -> Self {
5046            let test_db = TestDb::fake(background);
5047            let app_state = Self::build_app_state(&test_db).await;
5048            let peer = Peer::new();
5049            let notifications = mpsc::unbounded();
5050            let server = Server::new(app_state.clone(), peer.clone(), Some(notifications.0));
5051            Self {
5052                peer,
5053                app_state,
5054                server,
5055                foreground,
5056                notifications: notifications.1,
5057                connection_killers: Default::default(),
5058                forbid_connections: Default::default(),
5059                _test_db: test_db,
5060            }
5061        }
5062
5063        async fn create_client(&mut self, cx: &mut TestAppContext, name: &str) -> TestClient {
5064            cx.update(|cx| {
5065                let settings = Settings::test(cx);
5066                cx.set_global(settings);
5067            });
5068
5069            let http = FakeHttpClient::with_404_response();
5070            let user_id = self.app_state.db.create_user(name, false).await.unwrap();
5071            let client_name = name.to_string();
5072            let mut client = Client::new(http.clone());
5073            let server = self.server.clone();
5074            let connection_killers = self.connection_killers.clone();
5075            let forbid_connections = self.forbid_connections.clone();
5076            let (connection_id_tx, mut connection_id_rx) = mpsc::channel(16);
5077
5078            Arc::get_mut(&mut client)
5079                .unwrap()
5080                .override_authenticate(move |cx| {
5081                    cx.spawn(|_| async move {
5082                        let access_token = "the-token".to_string();
5083                        Ok(Credentials {
5084                            user_id: user_id.0 as u64,
5085                            access_token,
5086                        })
5087                    })
5088                })
5089                .override_establish_connection(move |credentials, cx| {
5090                    assert_eq!(credentials.user_id, user_id.0 as u64);
5091                    assert_eq!(credentials.access_token, "the-token");
5092
5093                    let server = server.clone();
5094                    let connection_killers = connection_killers.clone();
5095                    let forbid_connections = forbid_connections.clone();
5096                    let client_name = client_name.clone();
5097                    let connection_id_tx = connection_id_tx.clone();
5098                    cx.spawn(move |cx| async move {
5099                        if forbid_connections.load(SeqCst) {
5100                            Err(EstablishConnectionError::other(anyhow!(
5101                                "server is forbidding connections"
5102                            )))
5103                        } else {
5104                            let (client_conn, server_conn, kill_conn) =
5105                                Connection::in_memory(cx.background());
5106                            connection_killers.lock().insert(user_id, kill_conn);
5107                            cx.background()
5108                                .spawn(server.handle_connection(
5109                                    server_conn,
5110                                    client_name,
5111                                    user_id,
5112                                    Some(connection_id_tx),
5113                                    cx.background(),
5114                                ))
5115                                .detach();
5116                            Ok(client_conn)
5117                        }
5118                    })
5119                });
5120
5121            client
5122                .authenticate_and_connect(false, &cx.to_async())
5123                .await
5124                .unwrap();
5125
5126            Channel::init(&client);
5127            Project::init(&client);
5128            cx.update(|cx| {
5129                workspace::init(&client, cx);
5130            });
5131
5132            let peer_id = PeerId(connection_id_rx.next().await.unwrap().0);
5133            let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http, cx));
5134
5135            let client = TestClient {
5136                client,
5137                peer_id,
5138                user_store,
5139                language_registry: Arc::new(LanguageRegistry::test()),
5140                project: Default::default(),
5141                buffers: Default::default(),
5142            };
5143            client.wait_for_current_user(cx).await;
5144            client
5145        }
5146
5147        fn disconnect_client(&self, user_id: UserId) {
5148            self.connection_killers.lock().remove(&user_id);
5149        }
5150
5151        fn forbid_connections(&self) {
5152            self.forbid_connections.store(true, SeqCst);
5153        }
5154
5155        fn allow_connections(&self) {
5156            self.forbid_connections.store(false, SeqCst);
5157        }
5158
5159        async fn build_app_state(test_db: &TestDb) -> Arc<AppState> {
5160            let mut config = Config::default();
5161            config.session_secret = "a".repeat(32);
5162            config.database_url = test_db.url.clone();
5163            let github_client = github::AppClient::test();
5164            Arc::new(AppState {
5165                db: test_db.db().clone(),
5166                handlebars: Default::default(),
5167                auth_client: auth::build_client("", ""),
5168                repo_client: github::RepoClient::test(&github_client),
5169                github_client,
5170                config,
5171            })
5172        }
5173
5174        async fn state<'a>(&'a self) -> RwLockReadGuard<'a, Store> {
5175            self.server.store.read()
5176        }
5177
5178        async fn condition<F>(&mut self, mut predicate: F)
5179        where
5180            F: FnMut(&Store) -> bool,
5181        {
5182            async_std::future::timeout(Duration::from_millis(500), async {
5183                while !(predicate)(&*self.server.store.read()) {
5184                    self.foreground.start_waiting();
5185                    self.notifications.next().await;
5186                    self.foreground.finish_waiting();
5187                }
5188            })
5189            .await
5190            .expect("condition timed out");
5191        }
5192    }
5193
5194    impl Drop for TestServer {
5195        fn drop(&mut self) {
5196            self.peer.reset();
5197        }
5198    }
5199
5200    struct TestClient {
5201        client: Arc<Client>,
5202        pub peer_id: PeerId,
5203        pub user_store: ModelHandle<UserStore>,
5204        language_registry: Arc<LanguageRegistry>,
5205        project: Option<ModelHandle<Project>>,
5206        buffers: HashSet<ModelHandle<language::Buffer>>,
5207    }
5208
5209    impl Deref for TestClient {
5210        type Target = Arc<Client>;
5211
5212        fn deref(&self) -> &Self::Target {
5213            &self.client
5214        }
5215    }
5216
5217    impl TestClient {
5218        pub fn current_user_id(&self, cx: &TestAppContext) -> UserId {
5219            UserId::from_proto(
5220                self.user_store
5221                    .read_with(cx, |user_store, _| user_store.current_user().unwrap().id),
5222            )
5223        }
5224
5225        async fn wait_for_current_user(&self, cx: &TestAppContext) {
5226            let mut authed_user = self
5227                .user_store
5228                .read_with(cx, |user_store, _| user_store.watch_current_user());
5229            while authed_user.next().await.unwrap().is_none() {}
5230        }
5231
5232        async fn build_local_project(
5233            &mut self,
5234            fs: Arc<FakeFs>,
5235            root_path: impl AsRef<Path>,
5236            cx: &mut TestAppContext,
5237        ) -> (ModelHandle<Project>, WorktreeId) {
5238            let project = cx.update(|cx| {
5239                Project::local(
5240                    self.client.clone(),
5241                    self.user_store.clone(),
5242                    self.language_registry.clone(),
5243                    fs,
5244                    cx,
5245                )
5246            });
5247            self.project = Some(project.clone());
5248            let (worktree, _) = project
5249                .update(cx, |p, cx| {
5250                    p.find_or_create_local_worktree(root_path, true, cx)
5251                })
5252                .await
5253                .unwrap();
5254            worktree
5255                .read_with(cx, |tree, _| tree.as_local().unwrap().scan_complete())
5256                .await;
5257            project
5258                .update(cx, |project, _| project.next_remote_id())
5259                .await;
5260            (project, worktree.read_with(cx, |tree, _| tree.id()))
5261        }
5262
5263        async fn build_remote_project(
5264            &mut self,
5265            project_id: u64,
5266            cx: &mut TestAppContext,
5267        ) -> ModelHandle<Project> {
5268            let project = Project::remote(
5269                project_id,
5270                self.client.clone(),
5271                self.user_store.clone(),
5272                self.language_registry.clone(),
5273                FakeFs::new(cx.background()),
5274                &mut cx.to_async(),
5275            )
5276            .await
5277            .unwrap();
5278            self.project = Some(project.clone());
5279            project
5280        }
5281
5282        fn build_workspace(
5283            &self,
5284            project: &ModelHandle<Project>,
5285            cx: &mut TestAppContext,
5286        ) -> ViewHandle<Workspace> {
5287            let (window_id, _) = cx.add_window(|_| EmptyView);
5288            cx.add_view(window_id, |cx| {
5289                let fs = project.read(cx).fs().clone();
5290                Workspace::new(
5291                    &WorkspaceParams {
5292                        fs,
5293                        project: project.clone(),
5294                        user_store: self.user_store.clone(),
5295                        languages: self.language_registry.clone(),
5296                        channel_list: cx.add_model(|cx| {
5297                            ChannelList::new(self.user_store.clone(), self.client.clone(), cx)
5298                        }),
5299                        client: self.client.clone(),
5300                    },
5301                    cx,
5302                )
5303            })
5304        }
5305
5306        fn simulate_host(
5307            mut self,
5308            project: ModelHandle<Project>,
5309            mut language_server_config: LanguageServerConfig,
5310            operations: Rc<Cell<usize>>,
5311            max_operations: usize,
5312            rng: Arc<Mutex<StdRng>>,
5313            mut cx: TestAppContext,
5314        ) -> impl Future<Output = (Self, TestAppContext)> {
5315            let files: Arc<Mutex<Vec<PathBuf>>> = Default::default();
5316
5317            // Set up a fake language server.
5318            language_server_config.set_fake_initializer({
5319                let rng = rng.clone();
5320                let files = files.clone();
5321                let project = project.downgrade();
5322                move |fake_server| {
5323                    fake_server.handle_request::<lsp::request::Completion, _>(|_, _| {
5324                        Some(lsp::CompletionResponse::Array(vec![lsp::CompletionItem {
5325                            text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
5326                                range: lsp::Range::new(
5327                                    lsp::Position::new(0, 0),
5328                                    lsp::Position::new(0, 0),
5329                                ),
5330                                new_text: "the-new-text".to_string(),
5331                            })),
5332                            ..Default::default()
5333                        }]))
5334                    });
5335
5336                    fake_server.handle_request::<lsp::request::CodeActionRequest, _>(|_, _| {
5337                        Some(vec![lsp::CodeActionOrCommand::CodeAction(
5338                            lsp::CodeAction {
5339                                title: "the-code-action".to_string(),
5340                                ..Default::default()
5341                            },
5342                        )])
5343                    });
5344
5345                    fake_server.handle_request::<lsp::request::PrepareRenameRequest, _>(
5346                        |params, _| {
5347                            Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
5348                                params.position,
5349                                params.position,
5350                            )))
5351                        },
5352                    );
5353
5354                    fake_server.handle_request::<lsp::request::GotoDefinition, _>({
5355                        let files = files.clone();
5356                        let rng = rng.clone();
5357                        move |_, _| {
5358                            let files = files.lock();
5359                            let mut rng = rng.lock();
5360                            let count = rng.gen_range::<usize, _>(1..3);
5361                            let files = (0..count)
5362                                .map(|_| files.choose(&mut *rng).unwrap())
5363                                .collect::<Vec<_>>();
5364                            log::info!("LSP: Returning definitions in files {:?}", &files);
5365                            Some(lsp::GotoDefinitionResponse::Array(
5366                                files
5367                                    .into_iter()
5368                                    .map(|file| lsp::Location {
5369                                        uri: lsp::Url::from_file_path(file).unwrap(),
5370                                        range: Default::default(),
5371                                    })
5372                                    .collect(),
5373                            ))
5374                        }
5375                    });
5376
5377                    fake_server.handle_request::<lsp::request::DocumentHighlightRequest, _>({
5378                        let rng = rng.clone();
5379                        let project = project.clone();
5380                        move |params, mut cx| {
5381                            if let Some(project) = project.upgrade(&cx) {
5382                                project.update(&mut cx, |project, cx| {
5383                                    let path = params
5384                                        .text_document_position_params
5385                                        .text_document
5386                                        .uri
5387                                        .to_file_path()
5388                                        .unwrap();
5389                                    let (worktree, relative_path) =
5390                                        project.find_local_worktree(&path, cx)?;
5391                                    let project_path =
5392                                        ProjectPath::from((worktree.read(cx).id(), relative_path));
5393                                    let buffer =
5394                                        project.get_open_buffer(&project_path, cx)?.read(cx);
5395
5396                                    let mut highlights = Vec::new();
5397                                    let highlight_count = rng.lock().gen_range(1..=5);
5398                                    let mut prev_end = 0;
5399                                    for _ in 0..highlight_count {
5400                                        let range =
5401                                            buffer.random_byte_range(prev_end, &mut *rng.lock());
5402                                        let start = buffer
5403                                            .offset_to_point_utf16(range.start)
5404                                            .to_lsp_position();
5405                                        let end = buffer
5406                                            .offset_to_point_utf16(range.end)
5407                                            .to_lsp_position();
5408                                        highlights.push(lsp::DocumentHighlight {
5409                                            range: lsp::Range::new(start, end),
5410                                            kind: Some(lsp::DocumentHighlightKind::READ),
5411                                        });
5412                                        prev_end = range.end;
5413                                    }
5414                                    Some(highlights)
5415                                })
5416                            } else {
5417                                None
5418                            }
5419                        }
5420                    });
5421                }
5422            });
5423
5424            project.update(&mut cx, |project, _| {
5425                project.languages().add(Arc::new(Language::new(
5426                    LanguageConfig {
5427                        name: "Rust".into(),
5428                        path_suffixes: vec!["rs".to_string()],
5429                        language_server: Some(language_server_config),
5430                        ..Default::default()
5431                    },
5432                    None,
5433                )));
5434            });
5435
5436            async move {
5437                let fs = project.read_with(&cx, |project, _| project.fs().clone());
5438                while operations.get() < max_operations {
5439                    operations.set(operations.get() + 1);
5440
5441                    let distribution = rng.lock().gen_range::<usize, _>(0..100);
5442                    match distribution {
5443                        0..=20 if !files.lock().is_empty() => {
5444                            let path = files.lock().choose(&mut *rng.lock()).unwrap().clone();
5445                            let mut path = path.as_path();
5446                            while let Some(parent_path) = path.parent() {
5447                                path = parent_path;
5448                                if rng.lock().gen() {
5449                                    break;
5450                                }
5451                            }
5452
5453                            log::info!("Host: find/create local worktree {:?}", path);
5454                            let find_or_create_worktree = project.update(&mut cx, |project, cx| {
5455                                project.find_or_create_local_worktree(path, true, cx)
5456                            });
5457                            let find_or_create_worktree = async move {
5458                                find_or_create_worktree.await.unwrap();
5459                            };
5460                            if rng.lock().gen() {
5461                                cx.background().spawn(find_or_create_worktree).detach();
5462                            } else {
5463                                find_or_create_worktree.await;
5464                            }
5465                        }
5466                        10..=80 if !files.lock().is_empty() => {
5467                            let buffer = if self.buffers.is_empty() || rng.lock().gen() {
5468                                let file = files.lock().choose(&mut *rng.lock()).unwrap().clone();
5469                                let (worktree, path) = project
5470                                    .update(&mut cx, |project, cx| {
5471                                        project.find_or_create_local_worktree(
5472                                            file.clone(),
5473                                            true,
5474                                            cx,
5475                                        )
5476                                    })
5477                                    .await
5478                                    .unwrap();
5479                                let project_path =
5480                                    worktree.read_with(&cx, |worktree, _| (worktree.id(), path));
5481                                log::info!(
5482                                    "Host: opening path {:?}, worktree {}, relative_path {:?}",
5483                                    file,
5484                                    project_path.0,
5485                                    project_path.1
5486                                );
5487                                let buffer = project
5488                                    .update(&mut cx, |project, cx| {
5489                                        project.open_buffer(project_path, cx)
5490                                    })
5491                                    .await
5492                                    .unwrap();
5493                                self.buffers.insert(buffer.clone());
5494                                buffer
5495                            } else {
5496                                self.buffers
5497                                    .iter()
5498                                    .choose(&mut *rng.lock())
5499                                    .unwrap()
5500                                    .clone()
5501                            };
5502
5503                            if rng.lock().gen_bool(0.1) {
5504                                cx.update(|cx| {
5505                                    log::info!(
5506                                        "Host: dropping buffer {:?}",
5507                                        buffer.read(cx).file().unwrap().full_path(cx)
5508                                    );
5509                                    self.buffers.remove(&buffer);
5510                                    drop(buffer);
5511                                });
5512                            } else {
5513                                buffer.update(&mut cx, |buffer, cx| {
5514                                    log::info!(
5515                                        "Host: updating buffer {:?} ({})",
5516                                        buffer.file().unwrap().full_path(cx),
5517                                        buffer.remote_id()
5518                                    );
5519                                    buffer.randomly_edit(&mut *rng.lock(), 5, cx)
5520                                });
5521                            }
5522                        }
5523                        _ => loop {
5524                            let path_component_count = rng.lock().gen_range::<usize, _>(1..=5);
5525                            let mut path = PathBuf::new();
5526                            path.push("/");
5527                            for _ in 0..path_component_count {
5528                                let letter = rng.lock().gen_range(b'a'..=b'z');
5529                                path.push(std::str::from_utf8(&[letter]).unwrap());
5530                            }
5531                            path.set_extension("rs");
5532                            let parent_path = path.parent().unwrap();
5533
5534                            log::info!("Host: creating file {:?}", path,);
5535
5536                            if fs.create_dir(&parent_path).await.is_ok()
5537                                && fs.create_file(&path, Default::default()).await.is_ok()
5538                            {
5539                                files.lock().push(path);
5540                                break;
5541                            } else {
5542                                log::info!("Host: cannot create file");
5543                            }
5544                        },
5545                    }
5546
5547                    cx.background().simulate_random_delay().await;
5548                }
5549
5550                log::info!("Host done");
5551
5552                self.project = Some(project);
5553                (self, cx)
5554            }
5555        }
5556
5557        pub async fn simulate_guest(
5558            mut self,
5559            guest_id: usize,
5560            project: ModelHandle<Project>,
5561            operations: Rc<Cell<usize>>,
5562            max_operations: usize,
5563            rng: Arc<Mutex<StdRng>>,
5564            mut cx: TestAppContext,
5565        ) -> (Self, TestAppContext) {
5566            while operations.get() < max_operations {
5567                let buffer = if self.buffers.is_empty() || rng.lock().gen() {
5568                    let worktree = if let Some(worktree) = project.read_with(&cx, |project, cx| {
5569                        project
5570                            .worktrees(&cx)
5571                            .filter(|worktree| {
5572                                let worktree = worktree.read(cx);
5573                                worktree.is_visible()
5574                                    && worktree.entries(false).any(|e| e.is_file())
5575                            })
5576                            .choose(&mut *rng.lock())
5577                    }) {
5578                        worktree
5579                    } else {
5580                        cx.background().simulate_random_delay().await;
5581                        continue;
5582                    };
5583
5584                    operations.set(operations.get() + 1);
5585                    let (worktree_root_name, project_path) =
5586                        worktree.read_with(&cx, |worktree, _| {
5587                            let entry = worktree
5588                                .entries(false)
5589                                .filter(|e| e.is_file())
5590                                .choose(&mut *rng.lock())
5591                                .unwrap();
5592                            (
5593                                worktree.root_name().to_string(),
5594                                (worktree.id(), entry.path.clone()),
5595                            )
5596                        });
5597                    log::info!(
5598                        "Guest {}: opening path {:?} in worktree {} ({})",
5599                        guest_id,
5600                        project_path.1,
5601                        project_path.0,
5602                        worktree_root_name,
5603                    );
5604                    let buffer = project
5605                        .update(&mut cx, |project, cx| {
5606                            project.open_buffer(project_path.clone(), cx)
5607                        })
5608                        .await
5609                        .unwrap();
5610                    log::info!(
5611                        "Guest {}: opened path {:?} in worktree {} ({}) with buffer id {}",
5612                        guest_id,
5613                        project_path.1,
5614                        project_path.0,
5615                        worktree_root_name,
5616                        buffer.read_with(&cx, |buffer, _| buffer.remote_id())
5617                    );
5618                    self.buffers.insert(buffer.clone());
5619                    buffer
5620                } else {
5621                    operations.set(operations.get() + 1);
5622
5623                    self.buffers
5624                        .iter()
5625                        .choose(&mut *rng.lock())
5626                        .unwrap()
5627                        .clone()
5628                };
5629
5630                let choice = rng.lock().gen_range(0..100);
5631                match choice {
5632                    0..=9 => {
5633                        cx.update(|cx| {
5634                            log::info!(
5635                                "Guest {}: dropping buffer {:?}",
5636                                guest_id,
5637                                buffer.read(cx).file().unwrap().full_path(cx)
5638                            );
5639                            self.buffers.remove(&buffer);
5640                            drop(buffer);
5641                        });
5642                    }
5643                    10..=19 => {
5644                        let completions = project.update(&mut cx, |project, cx| {
5645                            log::info!(
5646                                "Guest {}: requesting completions for buffer {} ({:?})",
5647                                guest_id,
5648                                buffer.read(cx).remote_id(),
5649                                buffer.read(cx).file().unwrap().full_path(cx)
5650                            );
5651                            let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
5652                            project.completions(&buffer, offset, cx)
5653                        });
5654                        let completions = cx.background().spawn(async move {
5655                            completions.await.expect("completions request failed");
5656                        });
5657                        if rng.lock().gen_bool(0.3) {
5658                            log::info!("Guest {}: detaching completions request", guest_id);
5659                            completions.detach();
5660                        } else {
5661                            completions.await;
5662                        }
5663                    }
5664                    20..=29 => {
5665                        let code_actions = project.update(&mut cx, |project, cx| {
5666                            log::info!(
5667                                "Guest {}: requesting code actions for buffer {} ({:?})",
5668                                guest_id,
5669                                buffer.read(cx).remote_id(),
5670                                buffer.read(cx).file().unwrap().full_path(cx)
5671                            );
5672                            let range = buffer.read(cx).random_byte_range(0, &mut *rng.lock());
5673                            project.code_actions(&buffer, range, cx)
5674                        });
5675                        let code_actions = cx.background().spawn(async move {
5676                            code_actions.await.expect("code actions request failed");
5677                        });
5678                        if rng.lock().gen_bool(0.3) {
5679                            log::info!("Guest {}: detaching code actions request", guest_id);
5680                            code_actions.detach();
5681                        } else {
5682                            code_actions.await;
5683                        }
5684                    }
5685                    30..=39 if buffer.read_with(&cx, |buffer, _| buffer.is_dirty()) => {
5686                        let (requested_version, save) = buffer.update(&mut cx, |buffer, cx| {
5687                            log::info!(
5688                                "Guest {}: saving buffer {} ({:?})",
5689                                guest_id,
5690                                buffer.remote_id(),
5691                                buffer.file().unwrap().full_path(cx)
5692                            );
5693                            (buffer.version(), buffer.save(cx))
5694                        });
5695                        let save = cx.background().spawn(async move {
5696                            let (saved_version, _) = save.await.expect("save request failed");
5697                            assert!(saved_version.observed_all(&requested_version));
5698                        });
5699                        if rng.lock().gen_bool(0.3) {
5700                            log::info!("Guest {}: detaching save request", guest_id);
5701                            save.detach();
5702                        } else {
5703                            save.await;
5704                        }
5705                    }
5706                    40..=44 => {
5707                        let prepare_rename = project.update(&mut cx, |project, cx| {
5708                            log::info!(
5709                                "Guest {}: preparing rename for buffer {} ({:?})",
5710                                guest_id,
5711                                buffer.read(cx).remote_id(),
5712                                buffer.read(cx).file().unwrap().full_path(cx)
5713                            );
5714                            let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
5715                            project.prepare_rename(buffer, offset, cx)
5716                        });
5717                        let prepare_rename = cx.background().spawn(async move {
5718                            prepare_rename.await.expect("prepare rename request failed");
5719                        });
5720                        if rng.lock().gen_bool(0.3) {
5721                            log::info!("Guest {}: detaching prepare rename request", guest_id);
5722                            prepare_rename.detach();
5723                        } else {
5724                            prepare_rename.await;
5725                        }
5726                    }
5727                    45..=49 => {
5728                        let definitions = project.update(&mut cx, |project, cx| {
5729                            log::info!(
5730                                "Guest {}: requesting definitions for buffer {} ({:?})",
5731                                guest_id,
5732                                buffer.read(cx).remote_id(),
5733                                buffer.read(cx).file().unwrap().full_path(cx)
5734                            );
5735                            let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
5736                            project.definition(&buffer, offset, cx)
5737                        });
5738                        let definitions = cx.background().spawn(async move {
5739                            definitions.await.expect("definitions request failed")
5740                        });
5741                        if rng.lock().gen_bool(0.3) {
5742                            log::info!("Guest {}: detaching definitions request", guest_id);
5743                            definitions.detach();
5744                        } else {
5745                            self.buffers
5746                                .extend(definitions.await.into_iter().map(|loc| loc.buffer));
5747                        }
5748                    }
5749                    50..=54 => {
5750                        let highlights = project.update(&mut cx, |project, cx| {
5751                            log::info!(
5752                                "Guest {}: requesting highlights for buffer {} ({:?})",
5753                                guest_id,
5754                                buffer.read(cx).remote_id(),
5755                                buffer.read(cx).file().unwrap().full_path(cx)
5756                            );
5757                            let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
5758                            project.document_highlights(&buffer, offset, cx)
5759                        });
5760                        let highlights = cx.background().spawn(async move {
5761                            highlights.await.expect("highlights request failed");
5762                        });
5763                        if rng.lock().gen_bool(0.3) {
5764                            log::info!("Guest {}: detaching highlights request", guest_id);
5765                            highlights.detach();
5766                        } else {
5767                            highlights.await;
5768                        }
5769                    }
5770                    55..=59 => {
5771                        let search = project.update(&mut cx, |project, cx| {
5772                            let query = rng.lock().gen_range('a'..='z');
5773                            log::info!("Guest {}: project-wide search {:?}", guest_id, query);
5774                            project.search(SearchQuery::text(query, false, false), cx)
5775                        });
5776                        let search = cx
5777                            .background()
5778                            .spawn(async move { search.await.expect("search request failed") });
5779                        if rng.lock().gen_bool(0.3) {
5780                            log::info!("Guest {}: detaching search request", guest_id);
5781                            search.detach();
5782                        } else {
5783                            self.buffers.extend(search.await.into_keys());
5784                        }
5785                    }
5786                    _ => {
5787                        buffer.update(&mut cx, |buffer, cx| {
5788                            log::info!(
5789                                "Guest {}: updating buffer {} ({:?})",
5790                                guest_id,
5791                                buffer.remote_id(),
5792                                buffer.file().unwrap().full_path(cx)
5793                            );
5794                            buffer.randomly_edit(&mut *rng.lock(), 5, cx)
5795                        });
5796                    }
5797                }
5798                cx.background().simulate_random_delay().await;
5799            }
5800
5801            log::info!("Guest {} done", guest_id);
5802
5803            self.project = Some(project);
5804            (self, cx)
5805        }
5806    }
5807
5808    impl Drop for TestClient {
5809        fn drop(&mut self) {
5810            self.client.tear_down();
5811        }
5812    }
5813
5814    impl Executor for Arc<gpui::executor::Background> {
5815        type Timer = gpui::executor::Timer;
5816
5817        fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F) {
5818            self.spawn(future).detach();
5819        }
5820
5821        fn timer(&self, duration: Duration) -> Self::Timer {
5822            self.as_ref().timer(duration)
5823        }
5824    }
5825
5826    fn channel_messages(channel: &Channel) -> Vec<(String, String, bool)> {
5827        channel
5828            .messages()
5829            .cursor::<()>()
5830            .map(|m| {
5831                (
5832                    m.sender.github_login.clone(),
5833                    m.body.clone(),
5834                    m.is_pending(),
5835                )
5836            })
5837            .collect()
5838    }
5839
5840    struct EmptyView;
5841
5842    impl gpui::Entity for EmptyView {
5843        type Event = ();
5844    }
5845
5846    impl gpui::View for EmptyView {
5847        fn ui_name() -> &'static str {
5848            "empty view"
5849        }
5850
5851        fn render(&mut self, _: &mut gpui::RenderContext<Self>) -> gpui::ElementBox {
5852            gpui::Element::boxed(gpui::elements::Empty)
5853        }
5854    }
5855}