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