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