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